7846461830

Committed 09 Feb 2024 03:58PM UTC coverage: 93.504% (-0.2%) from 93.737%

Build # 7846461830

Build Type

Pull #90

github

Committed by

web-flow

Commit Message

Merge 4e3e41de3 into 3de029959

Pull Request Pull Request #90: RepairSolver

Run Details

796 of 884 new or added lines in 7 files covered. (90.05%)

2 existing lines in 2 files now uncovered.

6852 of 7328 relevant lines covered (93.5%)

0.94 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.54

/src/isla/solver.py

# Copyright © 2022 CISPA Helmholtz Center for Information Security.
# Author: Dominic Steinhöfel.
#
# This file is part of ISLa.
#
# ISLa is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# ISLa is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with ISLa.  If not, see <http://www.gnu.org/licenses/>.

import copy
import functools
import heapq
import itertools
import logging
import math
import operator
import random
import sys
import time
from abc import ABC
from dataclasses import dataclass
from functools import reduce, lru_cache, partial
from typing import (
    Dict,
    List,
    Set,
    Optional,
    Tuple,
    Union,
    cast,
    Callable,
    Iterable,
    Sequence,
)

import pkg_resources
import z3
from grammar_graph import gg
from grammar_graph.gg import GrammarGraph
from grammar_to_regex.cfg2regex import RegexConverter
from grammar_to_regex.regex import regex_to_z3
from orderedset import OrderedSet
from packaging import version
from returns.converters import result_to_maybe
from returns.functions import compose, tap
from returns.maybe import Nothing, Some
from returns.pipeline import flow, is_successful
from returns.pointfree import lash
from returns.result import safe, Success

import isla.isla_shortcuts as sc
import isla.three_valued_truth
from isla import language
from isla.derivation_tree import DerivationTree
from isla.evaluator import (
    evaluate,
    quantified_formula_might_match,
    get_toplevel_quantified_formulas,
    eliminate_quantifiers,
)
from isla.evaluator import matches_for_quantified_formula
from isla.existential_helpers import (
    insert_tree,
    DIRECT_EMBEDDING,
    SELF_EMBEDDING,
    CONTEXT_ADDITION,
)
from isla.fuzzer import GrammarFuzzer, GrammarCoverageFuzzer, expansion_key
from isla.helpers import (
    delete_unreachable,
    shuffle,
    dict_of_lists_to_list_of_dicts,
    weighted_geometric_mean,
    assertions_activated,
    split_str_with_nonterminals,
    cluster_by_common_elements,
    is_nonterminal,
    canonical,
    lazyjoin,
    lazystr,
    Maybe,
    eliminate_suffixes,
    get_elem_by_equivalence,
    get_expansions,
    list_del,
    compute_nullable_nonterminals,
    eassert,
    merge_dict_of_sets,
    list_set,
)
from isla.isla_predicates import (
    STANDARD_STRUCTURAL_PREDICATES,
    STANDARD_SEMANTIC_PREDICATES,
    COUNT_PREDICATE,
)
from isla.language import (
    VariablesCollector,
    split_conjunction,
    split_disjunction,
    convert_to_nnf,
    ensure_unique_bound_variables,
    parse_isla,
    get_conjuncts,
    parse_bnf,
    ForallIntFormula,
    set_smt_auto_eval,
    NoopFormulaTransformer,
    set_smt_auto_subst,
    fresh_constant,
    to_dnf_clauses,
)
from isla.mutator import Mutator
from isla.parser import EarleyParser
from isla.type_defs import Grammar, Path, ImmutableList, CanonicalGrammar
from isla.z3_helpers import (
    z3_solve,
    z3_subst,
    z3_eq,
    z3_and,
    visit_z3_expr,
    smt_string_val_to_string,
    parent_relationships_in_z3_expr,
    numeric_intervals_from_regex,
    z3_or,
)


@dataclass(frozen=True)
class SolutionState:
    constraint: language.Formula
    tree: DerivationTree
    level: int = 0
    __hash: Optional[int] = None

    def formula_satisfied(
        self, grammar: Grammar
    ) -> isla.three_valued_truth.ThreeValuedTruth:
        if self.tree.is_open():
            # Have to instantiate variables first
            return isla.three_valued_truth.ThreeValuedTruth.unknown()

        return evaluate(self.constraint, self.tree, grammar)

    def complete(self) -> bool:
        if not self.tree.is_complete():
            return False

        # We assume that any universal quantifier has already been instantiated, if it
        # matches, and is thus satisfied, or another unsatisfied constraint resulted
        # from the instantiation. Existential, predicate, and SMT formulas have to be
        # eliminated first.

        return self.constraint == sc.true()

    # Less-than comparisons are needed for usage in the binary heap queue
    def __lt__(self, other: "SolutionState"):
        return hash(self) < hash(other)

    def __hash__(self):
        if self.__hash is None:
            result = hash((self.constraint, self.tree))
            object.__setattr__(self, "__hash", result)
            return result

        return self.__hash

    def __eq__(self, other):
        return (
            isinstance(other, SolutionState)
            and self.constraint == other.constraint
            and self.tree.structurally_equal(other.tree)
        )


@dataclass(frozen=True)
class CostWeightVector:
    """
    Collection of weights for the
    :class:`~isla.solver.GrammarBasedBlackboxCostComputer`.
    """

    tree_closing_cost: float = 0
    constraint_cost: float = 0
    derivation_depth_penalty: float = 0
    low_k_coverage_penalty: float = 0
    low_global_k_path_coverage_penalty: float = 0

    def __iter__(self):
        """
        Use tuple assignment for objects of this type:

        >>> v = CostWeightVector(1, 2, 3, 4, 5)
        >>> a, b, c, d, e = v
        >>> (a, b, c, d, e)
        (1, 2, 3, 4, 5)

        :return: An iterator of the fixed-size list of elements of the weight vector.
        """
        return iter(
            [
                self.tree_closing_cost,
                self.constraint_cost,
                self.derivation_depth_penalty,
                self.low_k_coverage_penalty,
                self.low_global_k_path_coverage_penalty,
            ]
        )

    def __getitem__(self, item: int) -> float:
        """
        Tuple-like access of elements of the vector.

        >>> v = CostWeightVector(1, 2, 3, 4, 5)
        >>> v[3]
        4

        :param item: A numeric index.
        :return: The element at index :code:`item`.
        """
        assert isinstance(item, int)
        return [
            self.tree_closing_cost,
            self.constraint_cost,
            self.derivation_depth_penalty,
            self.low_k_coverage_penalty,
            self.low_global_k_path_coverage_penalty,
        ][item]


@dataclass(frozen=True)
class CostSettings:
    weight_vector: CostWeightVector
    k: int = 3

    def __init__(self, weight_vector: CostWeightVector, k: int = 3):
        assert isinstance(weight_vector, CostWeightVector)
        assert isinstance(k, int)
        object.__setattr__(self, "weight_vector", weight_vector)
        object.__setattr__(self, "k", k)


STD_COST_SETTINGS = CostSettings(
    CostWeightVector(
        tree_closing_cost=6.5,
        constraint_cost=1,
        derivation_depth_penalty=4,
        low_k_coverage_penalty=2,
        low_global_k_path_coverage_penalty=19,
    ),
    k=3,
)


@dataclass(frozen=True)
class UnknownResultError(Exception):
    pass


@dataclass(frozen=True)
class SemanticError(Exception):
    pass


@dataclass(frozen=True)
class SolverDefaults:
    formula: Optional[language.Formula | str] = None
    structural_predicates: frozenset[language.StructuralPredicate] = (
        STANDARD_STRUCTURAL_PREDICATES
    )
    semantic_predicates: frozenset[language.SemanticPredicate] = (
        STANDARD_SEMANTIC_PREDICATES
    )
    max_number_free_instantiations: int = 10
    max_number_smt_instantiations: int = 10
    max_number_tree_insertion_results: int = 5
    enforce_unique_trees_in_queue: bool = False
    debug: bool = False
    cost_computer: Optional["CostComputer"] = None
    timeout_seconds: Optional[int] = None
    global_fuzzer: bool = False
    predicates_unique_in_int_arg: Tuple[language.SemanticPredicate, ...] = (
        COUNT_PREDICATE,
    )
    fuzzer_factory: Callable[[Grammar], GrammarFuzzer] = (
        lambda grammar: GrammarCoverageFuzzer(grammar)
    )
    tree_insertion_methods: Optional[int] = None
    activate_unsat_support: bool = False
    grammar_unwinding_threshold: int = 4
    initial_tree: Maybe[DerivationTree] = Nothing
    enable_optimized_z3_queries: bool = True
    start_symbol: Optional[str] = None


_DEFAULTS = SolverDefaults()


class ISLaSolver:
    """
    The solver class for ISLa formulas/constraints. Its top-level methods are

    :meth:`~isla.solver.ISLaSolver.solve`
      Use to generate solutions for an ISLa constraint.

    :meth:`~isla.solver.ISLaSolver.check`
      Use to check if an ISLa constraint is satisfied for a given input.

    :meth:`~isla.solver.ISLaSolver.parse`
      Use to parse and validate an input.

    :meth:`~isla.solver.ISLaSolver.repair`
      Use to repair an input such that it satisfies a constraint.

    :meth:`~isla.solver.ISLaSolver.mutate`
      Use to mutate an input such that the result satisfies a constraint.
    """

    def __init__(
        self,
        grammar: Grammar | str,
        formula: Optional[language.Formula | str] = _DEFAULTS.formula,
        structural_predicates: Set[
            language.StructuralPredicate
        ] = _DEFAULTS.structural_predicates,
        semantic_predicates: Set[
            language.SemanticPredicate
        ] = _DEFAULTS.semantic_predicates,
        max_number_free_instantiations: int = _DEFAULTS.max_number_free_instantiations,
        max_number_smt_instantiations: int = _DEFAULTS.max_number_smt_instantiations,
        max_number_tree_insertion_results: int = _DEFAULTS.max_number_tree_insertion_results,
        enforce_unique_trees_in_queue: bool = _DEFAULTS.enforce_unique_trees_in_queue,
        debug: bool = _DEFAULTS.debug,
        cost_computer: Optional["CostComputer"] = _DEFAULTS.cost_computer,
        timeout_seconds: Optional[int] = _DEFAULTS.timeout_seconds,
        global_fuzzer: bool = _DEFAULTS.global_fuzzer,
        predicates_unique_in_int_arg: Tuple[
            language.SemanticPredicate, ...
        ] = _DEFAULTS.predicates_unique_in_int_arg,
        fuzzer_factory: Callable[[Grammar], GrammarFuzzer] = _DEFAULTS.fuzzer_factory,
        tree_insertion_methods: Optional[int] = _DEFAULTS.tree_insertion_methods,
        activate_unsat_support: bool = _DEFAULTS.activate_unsat_support,
        grammar_unwinding_threshold: int = _DEFAULTS.grammar_unwinding_threshold,
        initial_tree: Maybe[DerivationTree] = _DEFAULTS.initial_tree,
        enable_optimized_z3_queries: bool = _DEFAULTS.enable_optimized_z3_queries,
        start_symbol: Optional[str] = _DEFAULTS.start_symbol,
    ):
        """
        The constructor of :class:`~isla.solver.ISLaSolver` accepts a large number of
        parameters. However, all but the first one, :code:`grammar`, are *optional.*

        The simplest way to construct an ISLa solver is by only providing it with a
        grammar only; it then works like a grammar fuzzer.

        >>> import random
        >>> random.seed(1)

        >>> import string
        >>> LANG_GRAMMAR = {
        ...     "<start>":
        ...         ["<stmt>"],
        ...     "<stmt>":
        ...         ["<assgn> ; <stmt>", "<assgn>"],
        ...     "<assgn>":
        ...         ["<var> := <rhs>"],
        ...     "<rhs>":
        ...         ["<var>", "<digit>"],
        ...     "<var>": list(string.ascii_lowercase),
        ...     "<digit>": list(string.digits)
        ... }
        >>>
        >>> from isla.solver import ISLaSolver
        >>> solver = ISLaSolver(LANG_GRAMMAR)
        >>>
        >>> str(solver.solve())
        'd := 9'
        >>> str(solver.solve())
        'v := n ; s := r'

        :param grammar: The underlying grammar; either, as a "Fuzzing Book" dictionary
          or in BNF syntax.
        :param formula: The formula to solve; either a string or a readily parsed
          formula. If no formula is given, a default `true` constraint is assumed, and
          the solver falls back to a grammar fuzzer. The number of produced solutions
          will then be bound by `max_number_free_instantiations`.
        :param structural_predicates: Structural predicates to use when parsing a
          formula.
        :param semantic_predicates: Semantic predicates to use when parsing a formula.
        :param max_number_free_instantiations: Number of times that nonterminals that
          are not bound by any formula should be expanded by a coverage-based fuzzer.
        :param max_number_smt_instantiations: Number of solutions of SMT formulas that
          should be produced.
        :param max_number_tree_insertion_results: The maximum number of results when
          solving existential quantifiers by tree insertion.
        :param enforce_unique_trees_in_queue: If true, states with the same tree as an
          already existing tree in the queue are discarded, irrespectively of the
          constraint.
        :param debug: If true, debug information about the evolution of states is
          collected, notably in the field state_tree. The root of the tree is in the
          field state_tree_root. The field costs stores the computed cost values for
          all new nodes.
        :param cost_computer: The `CostComputer` class for computing the cost relevant
          to placing states in ISLa's queue.
        :param timeout_seconds: Number of seconds after which the solver will terminate.
        :param global_fuzzer: If set to True, only one coverage-guided grammar fuzzer
          object is used to finish off unconstrained open derivation trees throughout
          the whole generation time. This may be beneficial for some targets; e.g., we
          experienced that CSV works significantly faster. However, the achieved k-path
          coverage can be lower with that setting.
        :param predicates_unique_in_int_arg: This is needed in certain cases for
          instantiating universal integer quantifiers. The supplied predicates should
          have exactly one integer argument, and hold for exactly one integer value
          once all other parameters are fixed.
        :param fuzzer_factory: Constructor of the fuzzer to use for instantiating
          "free" nonterminals.
        :param tree_insertion_methods: Combination of methods to use for existential
          quantifier elimination by tree insertion. Full selection: `DIRECT_EMBEDDING &
          SELF_EMBEDDING & CONTEXT_ADDITION`.
        :param activate_unsat_support: Set to True if you assume that a formula might
          be unsatisfiable. This triggers additional tests for unsatisfiability that
          reduce input generation performance, but might ensure termination (with a
          negative solver result) for unsatisfiable problems for which the solver could
          otherwise diverge.
        :param grammar_unwinding_threshold: When querying the SMT solver, ISLa passes a
          regular expression for the syntax of the involved nonterminals. If this
          syntax is not regular, we unwind the respective part in the reference grammar
          up to a depth of `grammar_unwinding_threshold`. If this is too shallow, it can
          happen that an equation etc. cannot be solved; if it is too deep, it can
          negatively impact performance (and quite tremendously so).
        :param initial_tree: An initial input tree for the queue, if the solver shall
          not start from the tree `(<start>, None)`.
        :param enable_optimized_z3_queries: Enables preprocessing of Z3 queries (mainly
          numeric problems concerning things like length). This can improve performance
          significantly; however, it might happen that certain problems cannot be solved
          anymore. In that case, this option can/should be deactivated.
        :param start_symbol: This is an alternative to `initial_tree` for starting with
          a start symbol different form `<start>`. If `start_symbol` is provided, a tree
          consisting of a single root node with the value of `start_symbol` is chosen as
          initial tree.
        """
        self.logger = logging.getLogger(type(self).__name__)

        # We require at least z3 4.8.13.0. ISLa might work for some older versions, but
        # at least for 4.8.8.0, we have witnessed that certain rather easy constraints,
        # like equations, don't work since z3 cannot handle the restrictions to more
        # complicated regular expressions. This happened in the XML case study with
        # constraints of the kind <id> == <id_no_prefix>. At least with 4.8.13.0, this
        # works flawlessly, but times out for 4.8.8.0.
        #
        # This should be solved using Python requirements, which however cannot be done
        # currently since the fuzzingbook library inflexibly binds z3 to 4.8.8.0. Thus,
        # one has to manually install a newer version and ignore the warning.

        z3_version = pkg_resources.get_distribution("z3-solver").version
        assert version.parse(z3_version) >= version.parse("4.8.13.0"), (
            f"ISLa requires at least z3 4.8.13.0, present: {z3_version}. "
            "Please install a newer z3 version, e.g., using 'pip install z3-solver==4.8.14.0'."
        )

        if isinstance(grammar, str):
            self.grammar = parse_bnf(grammar)
        else:
            self.grammar = copy.deepcopy(grammar)

        assert start_symbol is None or not is_successful(
            initial_tree
        ), "You cannot supply a start symbol *and* an initial tree."

        if start_symbol is not None:
            self.grammar |= {"<start>": [start_symbol]}
            self.grammar = delete_unreachable(self.grammar)

        self.graph = GrammarGraph.from_grammar(self.grammar)
        self.canonical_grammar = canonical(self.grammar)
        self.timeout_seconds = timeout_seconds
        self.start_time: Optional[int] = None
        self.global_fuzzer = global_fuzzer
        self.fuzzer = fuzzer_factory(self.grammar)
        self.fuzzer_factory = fuzzer_factory
        self.predicates_unique_in_int_arg: Set[language.SemanticPredicate] = set(
            predicates_unique_in_int_arg
        )
        self.grammar_unwinding_threshold = grammar_unwinding_threshold
        self.enable_optimized_z3_queries = enable_optimized_z3_queries

        if activate_unsat_support and tree_insertion_methods is None:
            self.tree_insertion_methods = 0
        else:
            if activate_unsat_support:
                assert tree_insertion_methods is not None
                print(
                    "With activate_unsat_support set, a 0 value for tree_insertion_methods is recommended, "
                    f"the current value is: {tree_insertion_methods}",
                    file=sys.stderr,
                )

            self.tree_insertion_methods = (
                DIRECT_EMBEDDING + SELF_EMBEDDING + CONTEXT_ADDITION
            )
            if tree_insertion_methods is not None:
                self.tree_insertion_methods = tree_insertion_methods

        self.activate_unsat_support = activate_unsat_support
        self.currently_unsat_checking: bool = False

        self.cost_computer = (
            cost_computer
            if cost_computer is not None
            else GrammarBasedBlackboxCostComputer(STD_COST_SETTINGS, self.graph)
        )

        formula = (
            sc.true()
            if formula is None
            else (
                parse_isla(
                    formula, self.grammar, structural_predicates, semantic_predicates
                )
                if isinstance(formula, str)
                else formula
            )
        )

        self.formula = ensure_unique_bound_variables(formula)

        top_constants: Set[language.Constant] = set(
            [
                c
                for c in VariablesCollector.collect(self.formula)
                if isinstance(c, language.Constant) and not c.is_numeric()
            ]
        )

        assert len(top_constants) <= 1, (
            "ISLa only accepts up to one constant (free variable), "
            + f'found {len(top_constants)}: {", ".join(map(str, top_constants))}'
        )

        only_top_constant = Maybe.from_optional(next(iter(top_constants), None))
        self.top_constant = only_top_constant.map(
            lambda c: language.Constant(
                c.name, Maybe.from_optional(start_symbol).value_or(c.n_type)
            )
        )

        if only_top_constant != self.top_constant:
            assert is_successful(only_top_constant)
            assert is_successful(self.top_constant)
            self.formula = self.formula.substitute_variables(
                {only_top_constant.unwrap(): self.top_constant.unwrap()}
            )

        quantifier_chains: List[Tuple[language.ForallFormula, ...]] = [
            tuple([f for f in c if isinstance(f, language.ForallFormula)])
            for c in get_quantifier_chains(self.formula)
        ]
        # TODO: Remove?
        self.quantifier_chains: List[Tuple[language.ForallFormula, ...]] = [
            c for c in quantifier_chains if c
        ]

        self.max_number_free_instantiations: int = max_number_free_instantiations
        self.max_number_smt_instantiations: int = max_number_smt_instantiations
        self.max_number_tree_insertion_results = max_number_tree_insertion_results
        self.enforce_unique_trees_in_queue = enforce_unique_trees_in_queue

        # Initialize Queue
        self.initial_tree = (
            initial_tree.lash(
                lambda _: Maybe.from_optional(start_symbol)
                .map(lambda s: eassert(s, s in self.grammar))
                .map(lambda s: DerivationTree(s, None))
            ).lash(
                lambda _: Some(
                    DerivationTree(
                        self.top_constant.map(lambda c: c.n_type).value_or("<start>"),
                        None,
                    )
                )
            )
        ).unwrap()

        initial_formula = self.top_constant.map(
            lambda c: self.formula.substitute_expressions({c: self.initial_tree})
        ).value_or(self.formula)
        initial_state = SolutionState(initial_formula, self.initial_tree)
        initial_states = self.establish_invariant(initial_state)

        self.queue: List[Tuple[float, SolutionState]] = []
        self.tree_hashes_in_queue: Set[int] = {self.initial_tree.structural_hash()}
        self.state_hashes_in_queue: Set[int] = {hash(state) for state in initial_states}
        for state in initial_states:
            heapq.heappush(self.queue, (self.compute_cost(state), state))

        self.seen_coverages: Set[str] = set()
        self.current_level: int = 0
        self.step_cnt: int = 0
        self.last_cost_recomputation: int = 0

        self.regex_cache = {}

        self.solutions: List[DerivationTree] = []

        # Debugging stuff
        self.debug = debug
        self.state_tree: Dict[SolutionState, List[SolutionState]] = (
            {}
        )  # is only filled if self.debug
        self.state_tree_root = None
        self.current_state = None
        self.costs: Dict[SolutionState, float] = {}

        if self.debug:
            self.state_tree[initial_state] = initial_states
            self.state_tree_root = initial_state
            self.costs[initial_state] = 0
            for state in initial_states:
                self.costs[state] = self.compute_cost(state)

    def solve(self) -> DerivationTree:
        """
        Attempts to compute a solution to the given ISLa formula. Returns that solution,
        if any. This function can be called repeatedly to obtain more solutions until
        one of two exception types is raised: A :class:`StopIteration` indicates that
        no more solution can be found; a :class:`TimeoutError` is raised if a timeout
        occurred. After that, an exception will be raised every time.

        The timeout can be controlled by the :code:`timeout_seconds`
        :meth:`constructor <isla.solver.ISLaSolver.__init__>` parameter.

        :return: A solution for the ISLa formula passed to the
          :class:`isla.solver.ISLaSolver`.
        """
        if self.timeout_seconds is not None and self.start_time is None:
            self.start_time = int(time.time())

        while self.queue:
            self.step_cnt += 1

            # import dill as pickle
            # state_hash = 9107154106757938105
            # out_file = "/tmp/saved_debug_state"
            # if hash(self.queue[0][1]) == state_hash:
            #     with open(out_file, 'wb') as debug_state_file:
            #         pickle.dump(self, debug_state_file)
            #     print(f"Dumping state to {out_file}")
            #     exit()

            if self.timeout_seconds is not None:
                if int(time.time()) - self.start_time > self.timeout_seconds:
                    self.logger.debug("TIMEOUT")
                    raise TimeoutError(self.timeout_seconds)

            if self.solutions:
                solution = self.solutions.pop(0)
                self.logger.debug('Found solution "%s"', solution)
                return solution

            cost: int
            state: SolutionState
            cost, state = heapq.heappop(self.queue)

            self.current_level = state.level
            self.tree_hashes_in_queue.discard(state.tree.structural_hash())
            self.state_hashes_in_queue.discard(hash(state))

            if self.debug:
                self.current_state = state
                self.state_tree.setdefault(state, [])
            self.logger.debug(
                "Polling new state (%s, %s) (hash %d, cost %f)",
                state.constraint,
                state.tree.to_string(show_open_leaves=True, show_ids=True),
                hash(state),
                cost,
            )
            self.logger.debug("Queue length: %s", len(self.queue))

            assert not isinstance(state.constraint, language.DisjunctiveFormula)

            # Instantiate all top-level structural predicate formulas.
            state = self.instantiate_structural_predicates(state)

            # Apply the first elimination function that is applicable.
            # The later ones are ignored.
            def process_and_extend_solutions(
                result_states: List[SolutionState],
            ) -> Nothing:
                assert result_states is not None
                self.solutions.extend(self.process_new_states(result_states))
                return Nothing

            flow(
                Nothing,
                *map(
                    compose(lambda f: (lambda _: f(state)), lash),
                    [
                        self.noop_on_false_constraint,
                        self.eliminate_existential_integer_quantifiers,
                        self.instantiate_universal_integer_quantifiers,
                        self.match_all_universal_formulas,
                        self.expand_to_match_quantifiers,
                        self.eliminate_all_semantic_formulas,
                        self.eliminate_all_ready_semantic_predicate_formulas,
                        self.eliminate_and_match_first_existential_formula_and_expand,
                        self.assert_remaining_formulas_are_lazy_binding_semantic,
                        self.finish_unconstrained_trees,
                        self.expand,
                    ],
                ),
            ).bind(process_and_extend_solutions)

        if self.solutions:
            solution = self.solutions.pop(0)
            self.logger.debug('Found solution "%s"', solution)
            return solution
        else:
            self.logger.debug("UNSAT")
            raise StopIteration()

    def check(self, inp: DerivationTree | str) -> bool:
        """
        Evaluates whether the given derivation tree satisfies the constraint passed to
        the solver. Raises an `UnknownResultError` if this could not be evaluated
        (e.g., because of a solver timeout or a semantic predicate that cannot be
        evaluated).

        :param inp: The input to evaluate, either readily parsed or as a string.
        :return: A truth value.
        """
        if isinstance(inp, str):
            try:
                self.parse(inp)
                return True
            except (SyntaxError, SemanticError):
                return False

        assert isinstance(inp, DerivationTree)

        result = evaluate(self.formula, inp, self.grammar)

        if result.is_unknown():
            raise UnknownResultError()
        else:
            return bool(result)

    def parse(
        self,
        inp: str,
        nonterminal: str = "<start>",
        skip_check: bool = False,
        silent: bool = False,
    ) -> DerivationTree:
        """
        Parses the given input `inp`. Raises a `SyntaxError` if the input does not
        satisfy the grammar, a `SemanticError` if it does not satisfy the constraint
        (this is only checked if `nonterminal` is "<start>"), and returns the parsed
        `DerivationTree` otherwise.

        :param inp: The input to parse.
        :param nonterminal: The nonterminal to start parsing with, if a string
          corresponding to a sub-grammar shall be parsed. We don't check semantic
          correctness in that case.
        :param skip_check: If True, the semantic check is left out.
        :param silent: If True, no error is sent to the log stream in case of a
            failed parse.
        :return: A parsed `DerivationTree`.
        """
        grammar = copy.deepcopy(self.grammar)
        if nonterminal != "<start>":
            grammar |= {"<start>": [nonterminal]}
            grammar = delete_unreachable(grammar)

        parser = EarleyParser(grammar)
        try:
            parse_tree = next(parser.parse(inp))
            if nonterminal != "<start>":
                parse_tree = parse_tree[1][0]
            tree = DerivationTree.from_parse_tree(parse_tree)
        except SyntaxError as err:
            if not silent:
                self.logger.error(
                    f'Error parsing "{inp}" starting with "{nonterminal}"'
                )
            raise err

        if not skip_check and nonterminal == "<start>" and not self.check(tree):
            raise SemanticError()

        return tree

    def repair(
        self, inp: DerivationTree | str, fix_timeout_seconds: float = 3
    ) -> Maybe[DerivationTree]:
        """
        Attempts to repair the given input. The input must not violate syntactic
        (grammar) constraints. If semantic constraints are violated, this method
        gradually abstracts the input and tries to turn it into a valid one.
        Note that intensive structural manipulations are not performed; we merely
        try to satisfy SMT-LIB and semantic predicate constraints.

        :param inp: The input to fix.
        :param fix_timeout_seconds: A timeout used when calling the solver for an
          abstracted input. Usually, a low timeout suffices.
        :return: A fixed input (or the original, if it was not broken) or nothing.
        """

        inp = self.parse(inp, skip_check=True) if isinstance(inp, str) else inp

        try:
            if self.check(inp) or not is_successful(self.top_constant):
                return Some(inp)
        except UnknownResultError:
            pass

        formula = self.top_constant.map(
            lambda c: self.formula.substitute_expressions({c: inp})
        ).unwrap()

        set_smt_auto_eval(formula, False)
        set_smt_auto_subst(formula, False)

        qfr_free = eliminate_quantifiers(
            formula,
            grammar=self.grammar,
            numeric_constants={
                c
                for c in VariablesCollector.collect(formula)
                if isinstance(c, language.Constant) and c.is_numeric()
            },
        )

        # We first evaluate all structural predicates; for now, we do not interfere
        # with structure.
        semantic_only = qfr_free.transform(EvaluatePredicateFormulasTransformer(inp))

        if semantic_only == sc.false():
            # This cannot be repaired while preserving structure; for existential
            # problems, we could try tree insertion. We leave this for future work.
            return Nothing

        # We try to satisfy any of the remaining disjunctive elements, in random order
        for formula_to_satisfy in shuffle(split_disjunction(semantic_only)):
            # Now, we consider all combinations of 1, 2, ... of the derivation trees
            # participating in the formula. We successively prune deeper and deeper
            # subtrees until the resulting input evaluates to "unknown" for the given
            # formula.

            participating_paths = {
                inp.find_node(arg) for arg in formula_to_satisfy.tree_arguments()
            }

            def do_complete(tree: DerivationTree) -> Maybe[DerivationTree]:
                return result_to_maybe(
                    safe(
                        self.copy_without_queue(
                            initial_tree=Some(tree),
                            timeout_seconds=Some(fix_timeout_seconds),
                        ).solve,
                        (UnknownResultError, TimeoutError, StopIteration),
                    )()
                )

            # If p1, p2 are in participating_paths, then we consider the following
            # path combinations (roughly) in the listed order:
            # {p1}, {p2}, {p1, p2}, {p1[:-1]}, {p2[:-1]}, {p1[:-1], p2}, {p1, p2[:-1]},
            # {p1[:-1], p2[:-1]}, ...
            for abstracted_tree in generate_abstracted_trees(inp, participating_paths):
                match (
                    safe(lambda: self.check(abstracted_tree))()
                    .bind(lambda _: Nothing)
                    .lash(
                        lambda exc: (
                            Some(abstracted_tree)
                            if isinstance(exc, UnknownResultError)
                            else Nothing
                        )
                    )
                    .bind(do_complete)
                ):
                    case Some(completed):
                        return Some(completed)
                    case _:
                        pass

        return Nothing

    def mutate(
        self,
        inp: DerivationTree | str,
        min_mutations: int = 2,
        max_mutations: int = 5,
        fix_timeout_seconds: float = 1,
    ) -> DerivationTree:
        """
        Mutates `inp` such that the result satisfies the constraint.

        :param inp: The input to mutate.
        :param min_mutations: The minimum number of mutation steps to perform.
        :param max_mutations: The maximum number of mutation steps to perform.
        :param fix_timeout_seconds: A timeout used when calling the solver for fixing
          an abstracted input. Usually, a low timeout suffices.
        :return: A mutated input.
        """

        inp = self.parse(inp, skip_check=True) if isinstance(inp, str) else inp
        mutator = Mutator(
            self.grammar,
            min_mutations=min_mutations,
            max_mutations=max_mutations,
            graph=self.graph,
        )

        while True:
            mutated = mutator.mutate(inp)
            if mutated.structurally_equal(inp):
                continue
            maybe_fixed = self.repair(mutated, fix_timeout_seconds)
            if is_successful(maybe_fixed):
                return maybe_fixed.unwrap()

    def copy_without_queue(
        self,
        grammar: Maybe[Grammar | str] = Nothing,
        formula: Maybe[language.Formula | str] = Nothing,
        max_number_free_instantiations: Maybe[int] = Nothing,
        max_number_smt_instantiations: Maybe[int] = Nothing,
        max_number_tree_insertion_results: Maybe[int] = Nothing,
        enforce_unique_trees_in_queue: Maybe[bool] = Nothing,
        debug: Maybe[bool] = Nothing,
        cost_computer: Maybe["CostComputer"] = Nothing,
        timeout_seconds: Maybe[int] = Nothing,
        global_fuzzer: Maybe[bool] = Nothing,
        predicates_unique_in_int_arg: Maybe[
            Tuple[language.SemanticPredicate, ...]
        ] = Nothing,
        fuzzer_factory: Maybe[Callable[[Grammar], GrammarFuzzer]] = Nothing,
        tree_insertion_methods: Maybe[int] = Nothing,
        activate_unsat_support: Maybe[bool] = Nothing,
        grammar_unwinding_threshold: Maybe[int] = Nothing,
        initial_tree: Maybe[DerivationTree] = Nothing,
        enable_optimized_z3_queries: Maybe[bool] = Nothing,
        start_symbol: Optional[str] = None,
    ):
        result = ISLaSolver(
            grammar=grammar.value_or(self.grammar),
            formula=formula.value_or(self.formula),
            max_number_free_instantiations=max_number_free_instantiations.value_or(
                self.max_number_free_instantiations
            ),
            max_number_smt_instantiations=max_number_smt_instantiations.value_or(
                self.max_number_smt_instantiations
            ),
            max_number_tree_insertion_results=max_number_tree_insertion_results.value_or(
                self.max_number_tree_insertion_results
            ),
            enforce_unique_trees_in_queue=enforce_unique_trees_in_queue.value_or(
                self.enforce_unique_trees_in_queue
            ),
            debug=debug.value_or(self.debug),
            cost_computer=cost_computer.value_or(self.cost_computer),
            timeout_seconds=timeout_seconds.value_or(self.timeout_seconds),
            global_fuzzer=global_fuzzer.value_or(self.global_fuzzer),
            predicates_unique_in_int_arg=predicates_unique_in_int_arg.value_or(
                self.predicates_unique_in_int_arg
            ),
            fuzzer_factory=fuzzer_factory.value_or(self.fuzzer_factory),
            tree_insertion_methods=tree_insertion_methods.value_or(
                self.tree_insertion_methods
            ),
            activate_unsat_support=activate_unsat_support.value_or(
                self.activate_unsat_support
            ),
            grammar_unwinding_threshold=grammar_unwinding_threshold.value_or(
                self.grammar_unwinding_threshold
            ),
            initial_tree=initial_tree,
            enable_optimized_z3_queries=enable_optimized_z3_queries.value_or(
                self.enable_optimized_z3_queries
            ),
            start_symbol=start_symbol,
        )

        result.regex_cache = self.regex_cache

        return result

    @staticmethod
    def noop_on_false_constraint(
        state: SolutionState,
    ) -> Maybe[List[SolutionState]]:
        if state.constraint == sc.false():
            # This state can be silently discarded.
            return Some([state])

        return Nothing

    def expand_to_match_quantifiers(
        self,
        state: SolutionState,
    ) -> Maybe[List[SolutionState]]:
        if all(
            not isinstance(conjunct, language.ForallFormula)
            for conjunct in get_conjuncts(state.constraint)
        ):
            return Nothing

        expansion_result = self.expand_tree(state)

        assert len(expansion_result) > 0, f"State {state} will never leave the queue."
        self.logger.debug(
            "Expanding state %s (%d successors)", state, len(expansion_result)
        )

        return Some(expansion_result)

    def eliminate_and_match_first_existential_formula_and_expand(
        self,
        state: SolutionState,
    ) -> Maybe[List[SolutionState]]:
        elim_result = self.eliminate_and_match_first_existential_formula(state)
        if elim_result is None:
            return Nothing

        # Also add some expansions of the original state, to create a larger
        # solution stream (otherwise, it might be possible that only a small
        # finite number of solutions are generated for existential formulas).
        return Some(
            elim_result + self.expand_tree(state, limit=2, only_universal=False)
        )

    def assert_remaining_formulas_are_lazy_binding_semantic(
        self,
        state: SolutionState,
    ) -> Maybe[List[SolutionState]]:
        # SEMANTIC PREDICATE FORMULAS can remain if they bind lazily. In that case, we can choose a random
        # instantiation and let the predicate "fix" the resulting tree.
        assert state.constraint == sc.true() or all(
            isinstance(conjunct, language.SemanticPredicateFormula)
            or (
                isinstance(conjunct, language.NegatedFormula)
                and isinstance(conjunct.args[0], language.SemanticPredicateFormula)
            )
            for conjunct in get_conjuncts(state.constraint)
        ), (
            "Constraint is not true and contains formulas "
            f"other than semantic predicate formulas: {state.constraint}"
        )

        assert (
            state.constraint == sc.true()
            or all(
                not pred_formula.binds_tree(leaf)
                for pred_formula in get_conjuncts(state.constraint)
                if isinstance(pred_formula, language.SemanticPredicateFormula)
                for _, leaf in state.tree.open_leaves()
            )
            or all(
                not cast(
                    language.SemanticPredicateFormula, pred_formula.args[0]
                ).binds_tree(leaf)
                for pred_formula in get_conjuncts(state.constraint)
                if isinstance(pred_formula, language.NegatedFormula)
                and isinstance(pred_formula.args[0], language.SemanticPredicateFormula)
            )
            for _, leaf in state.tree.open_leaves()
        ), (
            "Constraint is not true and contains semantic predicate formulas binding open tree leaves: "
            f"{state.constraint}, leaves: "
            + ", ".join(
                [str(leaf) for _, leaf in state.tree.open_leaves()],
            )
        )

        return Nothing

    def finish_unconstrained_trees(
        self,
        state: SolutionState,
    ) -> Maybe[List[SolutionState]]:
        fuzzer = (
            self.fuzzer if self.global_fuzzer else self.fuzzer_factory(self.grammar)
        )

        if isinstance(fuzzer, GrammarCoverageFuzzer):
            fuzzer.covered_expansions.update(self.seen_coverages)

        if state.constraint != sc.true():
            return Nothing

        closed_results: List[SolutionState] = []
        for _ in range(self.max_number_free_instantiations):
            result = state.tree
            for path, leaf in state.tree.open_leaves():
                leaf_inst = fuzzer.expand_tree(DerivationTree(leaf.value, None))
                result = result.replace_path(path, leaf_inst)

            closed_results.append(SolutionState(state.constraint, result))

        return Some(closed_results)

    def expand(
        self,
        state: SolutionState,
    ) -> Maybe[List[SolutionState]]:
        fuzzer = (
            self.fuzzer if self.global_fuzzer else self.fuzzer_factory(self.grammar)
        )

        if isinstance(fuzzer, GrammarCoverageFuzzer):
            fuzzer.covered_expansions.update(self.seen_coverages)

        result: List[SolutionState] = []
        for _ in range(self.max_number_free_instantiations):
            substitutions: Dict[DerivationTree, DerivationTree] = {
                subtree: fuzzer.expand_tree(DerivationTree(subtree.value, None))
                for path, subtree in state.tree.open_leaves()
            }

            if substitutions:
                result.append(
                    SolutionState(
                        state.constraint.substitute_expressions(substitutions),
                        state.tree.substitute(substitutions),
                    )
                )

        return Some(result)

    def instantiate_structural_predicates(self, state: SolutionState) -> SolutionState:
        predicate_formulas = [
            pred_formula
            for pred_formula in language.FilterVisitor(
                lambda f: isinstance(f, language.StructuralPredicateFormula)
            ).collect(state.constraint)
            if (
                isinstance(pred_formula, language.StructuralPredicateFormula)
                and all(
                    not isinstance(arg, language.Variable) for arg in pred_formula.args
                )
            )
        ]

        formula = state.constraint
        for predicate_formula in predicate_formulas:
            instantiation = language.SMTFormula(
                z3.BoolVal(predicate_formula.evaluate(state.tree))
            )
            self.logger.debug(
                "Eliminating (-> %s) structural predicate formula %s",
                instantiation,
                predicate_formula,
            )
            formula = language.replace_formula(
                formula, predicate_formula, instantiation
            )

        return SolutionState(formula, state.tree)

    def eliminate_existential_integer_quantifiers(
        self, state: SolutionState
    ) -> Maybe[List[SolutionState]]:
        existential_int_formulas = [
            conjunct
            for conjunct in get_conjuncts(state.constraint)
            if isinstance(conjunct, language.ExistsIntFormula)
        ]

        if not existential_int_formulas:
            return Nothing

        formula = state.constraint
        for existential_int_formula in existential_int_formulas:
            # The following check for validity is not only a performance measure, but required
            # when existential integer formulas are re-inserted. Otherwise, new constants get
            # introduced, and the solver won't terminate.
            if evaluate(
                existential_int_formula,
                state.tree,
                self.grammar,
                assumptions={
                    f
                    for f in split_conjunction(state.constraint)
                    if f != existential_int_formula
                },
            ).is_true():
                self.logger.debug(
                    "Removing existential integer quantifier '%.30s', already implied "
                    "by tree and existing constraints",
                    existential_int_formula,
                )
                # This should simplify the process after quantifier re-insertion.
                return Some(
                    [
                        SolutionState(
                            language.replace_formula(
                                state.constraint, existential_int_formula, sc.true()
                            ),
                            state.tree,
                        )
                    ]
                )

            self.logger.debug(
                "Eliminating existential integer quantifier %s", existential_int_formula
            )
            used_vars = set(VariablesCollector.collect(formula))
            fresh = language.fresh_constant(
                used_vars,
                language.Constant(
                    existential_int_formula.bound_variable.name,
                    existential_int_formula.bound_variable.n_type,
                ),
            )
            instantiation = existential_int_formula.inner_formula.substitute_variables(
                {existential_int_formula.bound_variable: fresh}
            )
            formula = language.replace_formula(
                formula, existential_int_formula, instantiation
            )

        return Some([SolutionState(formula, state.tree)])

    def instantiate_universal_integer_quantifiers(
        self, state: SolutionState
    ) -> Maybe[List[SolutionState]]:
        universal_int_formulas = [
            conjunct
            for conjunct in get_conjuncts(state.constraint)
            if isinstance(conjunct, language.ForallIntFormula)
        ]

        if not universal_int_formulas:
            return Nothing

        results: List[SolutionState] = [state]
        for universal_int_formula in universal_int_formulas:
            results = [
                result
                for formula_list in [
                    self.instantiate_universal_integer_quantifier(
                        previous_result, universal_int_formula
                    )
                    for previous_result in results
                ]
                for result in formula_list
            ]

        return Some(results)

    def instantiate_universal_integer_quantifier(
        self, state: SolutionState, universal_int_formula: language.ForallIntFormula
    ) -> List[SolutionState]:
        results = self.instantiate_universal_integer_quantifier_by_enumeration(
            state, universal_int_formula
        )

        if results:
            return results

        return self.instantiate_universal_integer_quantifier_by_transformation(
            state, universal_int_formula
        )

    def instantiate_universal_integer_quantifier_by_transformation(
        self, state: SolutionState, universal_int_formula: language.ForallIntFormula
    ) -> List[SolutionState]:
        # If the enumeration approach was not successful, we con transform the universal int
        # quantifier to an existential one in a particular situation:
        #
        # Let phi(elem, i) be such that phi(elem) (for fixed first argument) is a unary
        # relation that holds for exactly one argument:
        #
        # forall <A> elem:
        #   exists int i:
        #     phi(elem, i) and
        #     forall int i':
        #       phi(elem, i) <==> i = i'
        #
        # Then, the following transformations are equivalence-preserving:
        #
        # forall int i:
        #   exists <A> elem:
        #     not phi(elem, i)
        #
        # <==> (*)
        #
        # exists int i:
        #   exists <A> elem':
        #     phi(elem', i) &
        #   exists <A> elem:
        #     not phi(elem, i) &
        #   forall int i':
        #     i != i' ->
        #     exists <A> elem'':
        #       not phi(elem'', i')
        #
        # <==> (+)
        #
        # exists int i:
        #   exists <A> elem':
        #     phi(elem', i) &
        #   exists <A> elem:
        #     not phi(elem, i)
        #
        # (*)
        # Problematic is only the first inner conjunct. However, for every elem, there
        # has to be an i such that phi(elem, i) holds. If there is no no in the first
        # place, also the original formula would be unsatisfiable. Without this conjunct,
        # the transformation is a simple "quantifier unwinding."
        #
        # (+)
        # Let i' != i. Choose elem'' := elem': Since phi(elem', i) holds and i != i',
        # "not phi(elem', i')" has to hold.

        if (
            isinstance(universal_int_formula.inner_formula, language.ExistsFormula)
            and isinstance(
                universal_int_formula.inner_formula.inner_formula,
                language.NegatedFormula,
            )
            and isinstance(
                universal_int_formula.inner_formula.inner_formula.args[0],
                language.SemanticPredicateFormula,
            )
            and cast(
                language.SemanticPredicateFormula,
                universal_int_formula.inner_formula.inner_formula.args[0],
            ).predicate
            in self.predicates_unique_in_int_arg
        ):
            inner_formula: language.ExistsFormula = universal_int_formula.inner_formula
            predicate_formula: language.SemanticPredicateFormula = cast(
                language.SemanticPredicateFormula,
                cast(language.NegatedFormula, inner_formula.inner_formula).args[0],
            )

            fresh_var = language.fresh_bound_variable(
                language.VariablesCollector().collect(state.constraint),
                inner_formula.bound_variable,
                add=False,
            )

            new_formula = language.ExistsIntFormula(
                universal_int_formula.bound_variable,
                language.ExistsFormula(
                    fresh_var,
                    inner_formula.in_variable,
                    predicate_formula.substitute_variables(
                        {inner_formula.bound_variable: fresh_var}
                    ),
                )
                & inner_formula,
            )

            self.logger.debug(
                "Transforming universal integer quantifier "
                "(special case, see code comments for explanation):\n%s ==> %s",
                universal_int_formula,
                new_formula,
            )

            return [
                SolutionState(
                    language.replace_formula(
                        state.constraint, universal_int_formula, new_formula
                    ),
                    state.tree,
                )
            ]

        self.logger.warning(
            "Did not find a way to instantiate formula %s!\n"
            + "Discarding this state. Please report this to your nearest ISLa developer.",
            universal_int_formula,
        )

        return []

    def instantiate_universal_integer_quantifier_by_enumeration(
        self, state: SolutionState, universal_int_formula: ForallIntFormula
    ) -> Optional[List[SolutionState]]:
        constant = language.Constant(
            universal_int_formula.bound_variable.name,
            universal_int_formula.bound_variable.n_type,
        )

        # noinspection PyTypeChecker
        inner_formula = universal_int_formula.inner_formula.substitute_variables(
            {universal_int_formula.bound_variable: constant}
        )

        instantiations: List[
            Dict[
                language.Constant | DerivationTree,
                int | language.Constant | DerivationTree,
            ]
        ] = []

        if isinstance(universal_int_formula.inner_formula, language.DisjunctiveFormula):
            # In the disjunctive case, we attempt to falsify all SMT formulas in the inner formula
            # (on top level) that contain the bound variable as argument.
            smt_disjuncts = [
                formula
                for formula in language.split_disjunction(inner_formula)
                if isinstance(formula, language.SMTFormula)
                and constant in formula.free_variables()
            ]

            if smt_disjuncts and len(smt_disjuncts) < len(
                language.split_disjunction(inner_formula)
            ):
                instantiation_values = (
                    self.infer_satisfying_assignments_for_smt_formula(
                        -reduce(language.SMTFormula.disjunction, smt_disjuncts),
                        constant,
                    )
                )

                # We also try to falsify (negated) semantic predicate formulas, if present,
                # if there exist any remaining disjuncts.
                semantic_predicate_formulas: List[
                    Tuple[language.SemanticPredicateFormula, bool]
                ] = [
                    (
                        (pred_formula, False)
                        if isinstance(pred_formula, language.SemanticPredicateFormula)
                        else (cast(language.NegatedFormula, pred_formula).args[0], True)
                    )
                    for pred_formula in language.FilterVisitor(
                        lambda f: (
                            constant in f.free_variables()
                            and (
                                isinstance(f, language.SemanticPredicateFormula)
                                or isinstance(f, language.NegatedFormula)
                                and isinstance(
                                    f.args[0], language.SemanticPredicateFormula
                                )
                            )
                        ),
                        do_continue=lambda f: isinstance(
                            f, language.DisjunctiveFormula
                        ),
                    ).collect(inner_formula)
                    if all(
                        not isinstance(var, language.BoundVariable)
                        for var in pred_formula.free_variables()
                    )
                ]

                if semantic_predicate_formulas and len(
                    semantic_predicate_formulas
                ) + len(smt_disjuncts) < len(language.split_disjunction(inner_formula)):
                    for value in instantiation_values:
                        instantiation: Dict[
                            language.Constant | DerivationTree,
                            int | language.Constant | DerivationTree,
                        ] = {constant: value}
                        for (
                            semantic_predicate_formula,
                            negated,
                        ) in semantic_predicate_formulas:
                            eval_result = cast(
                                language.SemanticPredicateFormula,
                                language.substitute(
                                    semantic_predicate_formula, {constant: value}
                                ),
                            ).evaluate(self.graph, negate=not negated)
                            if eval_result.ready() and not eval_result.is_boolean():
                                instantiation.update(eval_result.result)
                        instantiations.append(instantiation)
                else:
                    instantiations.extend(
                        [{constant: value} for value in instantiation_values]
                    )

        results: List[SolutionState] = []
        for instantiation in instantiations:
            self.logger.debug(
                "Instantiating universal integer quantifier (%s -> %s) %s",
                universal_int_formula.bound_variable,
                instantiation[constant],
                universal_int_formula,
            )

            formula = language.replace_formula(
                state.constraint,
                universal_int_formula,
                language.substitute(inner_formula, instantiation),
            )
            formula = language.substitute(formula, instantiation)

            tree = state.tree.substitute(
                {
                    tree: subst
                    for tree, subst in instantiation.items()
                    if isinstance(tree, DerivationTree)
                }
            )

            results.append(SolutionState(formula, tree))

        return results

    def infer_satisfying_assignments_for_smt_formula(
        self, smt_formula: language.SMTFormula, constant: language.Constant
    ) -> Set[int | language.Constant]:
        """
        This method returns `self.max_number_free_instantiations` many solutions for
        the given :class:`~isla.language.SMTFormula` if `constant` is the only free
        variable in `smt_formula`. The given formula must be a numeric formula, i.e.,
        all free variables must be numeric. If more than one free variables are
        present, at most one solution is returned (see example below).

        :param smt_formula: The :class:`~isla.language.SMTFormula` to solve. Must
          only contain numeric free variables.
        :param constant: One free variable in `smt_formula`.
        :return: A set of solutions (see explanation above & comment below).

        We create a solver with a dummy grammar (it's not needed for this example),
        choosing a value of 5 for `max_number_free_instantiations`.

        >>> solver = ISLaSolver(
        ...     '<start> ::= "x"',  # dummy grammar
        ...     max_number_free_instantiations=5,
        ... )

        The formula we're considering is `x > 10`.

        >>> from isla.language import Constant, SMTFormula, Variable, unparse_isla
        >>> x = Constant("x", Variable.NUMERIC_NTYPE)

        >>> formula = SMTFormula(z3.StrToInt(x.to_smt()) > z3.IntVal(10), x)
        >>> unparse_isla(formula)
        '(< 10 (str.to.int x))'

        We obtain five results (due to our choice of `max_number_free_instantiations`).

        >>> results = solver.infer_satisfying_assignments_for_smt_formula(formula, x)
        >>> len(results)
        5

        All results are `int`s...

        >>> all(isinstance(result, int) for result in results)
        True

        ...and all are strictly greater than 10.

        >>> all(result > 10 for result in results)
        True

        Now, lets consider `x == y`. This formula contains *two* free variables, `x`
        and `y`. It is the only type of formula with more than one variable for which
        this method will return a solution in the current state.

        >>> y = Constant("y", Variable.NUMERIC_NTYPE)
        >>> formula = SMTFormula(
        ...     z3_eq(z3.StrToInt(x.to_smt()), z3.StrToInt(y.to_smt())), x, y)
        >>> unparse_isla(formula)
        '(= (str.to.int x) (str.to.int y))'

        The solution is the singleton set with the variable `y`, which is an
        instantiation of the constant `x` solving the equation.

        >>> solver.infer_satisfying_assignments_for_smt_formula(formula, x)
        {Constant("y", "NUM")}

        If we choose a different type of formula (a greater-than relation), we obtain
        an empty solution set.

        >>> formula = SMTFormula(
        ...     z3.StrToInt(x.to_smt()) > z3.StrToInt(y.to_smt()), x, y)
        >>> unparse_isla(formula)
        '(> (str.to.int x) (str.to.int y))'
        >>> solver.infer_satisfying_assignments_for_smt_formula(formula, x)
        set()

        With a non-numeric formula, we obtain an AssertionError (if assertions are
        enabled). This method expects to be called only internally, so this should
        not happen (with or without activated assertions).

        >>> z = Constant("x", "<start>")
        >>> formula = SMTFormula(z3_eq(z.to_smt(), z3.StringVal("x")), z)
        >>> print(unparse_isla(formula))
        const x: <start>;
        <BLANKLINE>
        (= x "x")
        >>> results = solver.infer_satisfying_assignments_for_smt_formula(formula, z)
        Traceback (most recent call last):
        ...
        AssertionError: Expected numeric solution.
        """

        free_variables = smt_formula.free_variables()
        max_instantiations = (
            self.max_number_free_instantiations if len(free_variables) == 1 else 1
        )

        try:
            solver_result = self.solve_quantifier_free_formula(
                (smt_formula,), max_instantiations=max_instantiations
            )

            solutions: Dict[language.Constant, Set[int]] = {
                c: {
                    int(solution[cast(language.Constant, c)].value)
                    for solution in solver_result
                }
                for c in free_variables
            }
        except ValueError:
            assert False, "Expected numeric solution."

        if solutions:
            if len(free_variables) == 1:
                return solutions[constant]
            else:
                assert all(len(solution) == 1 for solution in solutions.values())
                # In situations with multiple variables, we might have to abstract from
                # concrete values. Currently, we only support simple equality inference
                # (based on one sample...). Note that for supporting *more complex*
                # terms (e.g., additions), we would have to extend the whole
                # infrastructure: Substitutions with complex terms, and complex terms
                # in semantic predicate arguments, are unsupported as of now.
                candidates = {
                    c
                    for c in solutions
                    if c != constant
                    and next(iter(solutions[c])) == next(iter(solutions[constant]))
                }

                # Filter working candidates
                return {
                    c
                    for c in candidates
                    if self.solve_quantifier_free_formula(
                        (
                            cast(
                                language.SMTFormula,
                                smt_formula.substitute_variables({constant: c}),
                            ),
                        ),
                        max_instantiations=1,
                    )
                }

    def eliminate_all_semantic_formulas(
        self, state: SolutionState, max_instantiations: Optional[int] = None
    ) -> Maybe[List[SolutionState]]:
        """
        Eliminates all SMT-LIB formulas that appear in `state`'s constraint as conjunctive elements.
        If, e.g., an SMT-LIB formula occurs as a disjunction, no solution is computed.

        :param state: The state in which to solve all SMT-LIB formulas.
        :param max_instantiations: The number of solutions the SMT solver should be asked for.
        :return: The discovered solutions.
        """

        conjuncts = split_conjunction(state.constraint)
        semantic_formulas = [
            conjunct
            for conjunct in conjuncts
            if isinstance(conjunct, language.SMTFormula)
            and not z3.is_true(conjunct.formula)
        ]

        if not semantic_formulas:
            return Nothing

        self.logger.debug(
            "Eliminating semantic formulas [%s]", lazyjoin(", ", semantic_formulas)
        )

        prefix_conjunction = reduce(lambda a, b: a & b, semantic_formulas, sc.true())
        new_disjunct = prefix_conjunction & reduce(
            lambda a, b: a & b,
            [conjunct for conjunct in conjuncts if conjunct not in semantic_formulas],
            sc.true(),
        )

        return Some(
            self.eliminate_semantic_formula(
                prefix_conjunction,
                SolutionState(new_disjunct, state.tree),
                max_instantiations,
            )
        )

    def eliminate_all_ready_semantic_predicate_formulas(
        self, state: SolutionState
    ) -> Maybe[List[SolutionState]]:
        semantic_predicate_formulas: List[
            language.NegatedFormula | language.SemanticPredicateFormula
        ] = [
            cast(
                language.NegatedFormula | language.SemanticPredicateFormula,
                pred_formula,
            )
            for pred_formula in language.FilterVisitor(
                lambda f: (
                    isinstance(f, language.SemanticPredicateFormula)
                    or isinstance(f, language.NegatedFormula)
                    and isinstance(f.args[0], language.SemanticPredicateFormula)
                ),
                do_continue=lambda f: (
                    not isinstance(f, language.NegatedFormula)
                    or not isinstance(f.args[0], language.SemanticPredicateFormula)
                ),
            ).collect(state.constraint)
            if all(
                not isinstance(var, language.BoundVariable)
                for var in pred_formula.free_variables()
            )
        ]

        semantic_predicate_formulas = sorted(
            semantic_predicate_formulas,
            key=lambda f: (
                2 * cast(language.SemanticPredicateFormula, f.args[0]).predicate.order
                + 100
                if isinstance(f, language.NegatedFormula)
                else f.predicate.order
            ),
        )

        if not semantic_predicate_formulas:
            return Nothing

        result = state

        changed = False
        for idx, possibly_negated_semantic_predicate_formula in enumerate(
            semantic_predicate_formulas
        ):
            negated = isinstance(
                possibly_negated_semantic_predicate_formula, language.NegatedFormula
            )
            semantic_predicate_formula: language.SemanticPredicateFormula = (
                cast(
                    language.NegatedFormula, possibly_negated_semantic_predicate_formula
                ).args[0]
                if negated
                else possibly_negated_semantic_predicate_formula
            )

            evaluation_result = semantic_predicate_formula.evaluate(
                self.graph, negate=negated
            )
            if not evaluation_result.ready():
                continue

            self.logger.debug(
                "Eliminating semantic predicate formula %s", semantic_predicate_formula
            )
            changed = True

            if evaluation_result.is_boolean():
                result = SolutionState(
                    language.replace_formula(
                        result.constraint,
                        semantic_predicate_formula,
                        language.smt_atom(evaluation_result.true()),
                    ),
                    result.tree,
                )
                continue

            substitution = subtree_solutions(evaluation_result.result)

            new_constraint = language.replace_formula(
                result.constraint,
                semantic_predicate_formula,
                sc.false() if negated else sc.true(),
            ).substitute_expressions(substitution)

            for k in range(idx + 1, len(semantic_predicate_formulas)):
                semantic_predicate_formulas[k] = cast(
                    language.SemanticPredicateFormula,
                    semantic_predicate_formulas[k].substitute_expressions(substitution),
                )

            result = SolutionState(new_constraint, result.tree.substitute(substitution))
            assert self.graph.tree_is_valid(result.tree)

        return Maybe.from_optional([result] if changed else None)

    def eliminate_and_match_first_existential_formula(
        self, state: SolutionState
    ) -> Optional[List[SolutionState]]:
        # We produce up to two groups of output states: One where the first existential
        # formula, if it can be matched, is matched, and one where the first existential
        # formula is eliminated by tree insertion.

        def do_eliminate(
            first_existential_formula_with_idx: Tuple[int, language.ExistsFormula]
        ) -> List[SolutionState]:
            first_matched = OrderedSet(
                self.match_existential_formula(
                    first_existential_formula_with_idx[0], state
                )
            )

            # Tree insertion can be deactivated by setting `self.tree_insertion_methods`
            # to 0.
            if not self.tree_insertion_methods:
                return list(first_matched)

            if first_matched:
                self.logger.debug(
                    "Matched first existential formulas, result: [%s]",
                    lazyjoin(
                        ", ",
                        [
                            lazystr(lambda: f"{s} (hash={hash(s)})")
                            for s in first_matched
                        ],
                    ),
                )

            # 3. Eliminate first existential formula by tree insertion.
            elimination_result = OrderedSet(
                self.eliminate_existential_formula(
                    first_existential_formula_with_idx[0], state
                )
            )
            elimination_result = OrderedSet(
                [
                    result
                    for result in elimination_result
                    if not any(
                        other_result.tree == result.tree
                        and self.propositionally_unsatisfiable(
                            result.constraint & -other_result.constraint
                        )
                        for other_result in first_matched
                    )
                ]
            )

            if not elimination_result and not first_matched:
                self.logger.warning(
                    "Existential qfr elimination: Could not eliminate existential formula %s "
                    "by matching or tree insertion",
                    first_existential_formula_with_idx[1],
                )

            if elimination_result:
                self.logger.debug(
                    "Eliminated existential formula %s by tree insertion, %d successors",
                    first_existential_formula_with_idx[1],
                    len(elimination_result),
                )

            return [
                result
                for result in first_matched | elimination_result
                if result != state
            ]

        return (
            result_to_maybe(
                safe(
                    lambda: next(
                        (idx, conjunct)
                        for idx, conjunct in enumerate(
                            split_conjunction(state.constraint)
                        )
                        if isinstance(conjunct, language.ExistsFormula)
                    )
                )()
            )
            .map(do_eliminate)
            .value_or(None)
        )

    def match_all_universal_formulas(
        self, state: SolutionState
    ) -> Maybe[List[SolutionState]]:
        universal_formulas = [
            conjunct
            for conjunct in split_conjunction(state.constraint)
            if isinstance(conjunct, language.ForallFormula)
        ]

        if not universal_formulas:
            return Nothing

        result = self.match_universal_formulas(state)
        if not result:
            return Nothing

        self.logger.debug(
            "Matched universal formulas [%s]", lazyjoin(", ", universal_formulas)
        )

        return Some(result)

    def expand_tree(
        self,
        state: SolutionState,
        only_universal: bool = True,
        limit: Optional[int] = None,
    ) -> List[SolutionState]:
        """
        Expands the given tree, but not at nonterminals that can be freely instantiated of those that directly
        correspond to the assignment constant.

        :param state: The current state.
        :param only_universal: If set to True, only nonterminals that might match universal quantifiers are
        expanded. If set to false, also nonterminals matching to existential quantifiers are expanded.
        :param limit: If set to a value, this will return only up to limit expansions.
        :return: A (possibly empty) list of expanded trees.
        """

        nonterminal_expansions: Dict[Path, List[List[DerivationTree]]] = {
            leaf_path: [
                [
                    DerivationTree(child, None if is_nonterminal(child) else [])
                    for child in expansion
                ]
                for expansion in self.canonical_grammar[leaf_node.value]
            ]
            for leaf_path, leaf_node in state.tree.open_leaves()
            if any(
                self.quantified_formula_might_match(formula, leaf_path, state.tree)
                for formula in get_conjuncts(state.constraint)
                if (only_universal and isinstance(formula, language.ForallFormula))
                or (
                    not only_universal
                    and isinstance(formula, language.QuantifiedFormula)
                )
            )
        }

        if not nonterminal_expansions:
            return []

        possible_expansions: List[Dict[Path, List[DerivationTree]]] = []
        if not limit:
            possible_expansions = dict_of_lists_to_list_of_dicts(nonterminal_expansions)
            assert len(possible_expansions) == math.prod(
                len(values) for values in nonterminal_expansions.values()
            )
        else:
            for _ in range(limit):
                curr_expansion = {}
                for path, expansions in nonterminal_expansions.items():
                    if not expansions:
                        continue

                    curr_expansion[path] = random.choice(expansions)
                possible_expansions.append(curr_expansion)

        # This replaces a previous `if` statement with the negated condition as guard,
        # which seems to be dead code (the guard can never hold true due to the check
        # of emptiness of `nonterminal_expansions` above). We keep this assertion here
        # to be sure.
        assert (
            len(possible_expansions) > 1
            or len(possible_expansions) == 1
            and possible_expansions[0]
        )

        result: List[SolutionState] = []
        for possible_expansion in possible_expansions:
            expanded_tree = state.tree
            for path, new_children in possible_expansion.items():
                leaf_node = expanded_tree.get_subtree(path)
                expanded_tree = expanded_tree.replace_path(
                    path, DerivationTree(leaf_node.value, new_children, leaf_node.id)
                )

                assert expanded_tree is not state.tree
                assert expanded_tree != state.tree
                assert expanded_tree.structural_hash() != state.tree.structural_hash()

            updated_constraint = state.constraint.substitute_expressions(
                {
                    state.tree.get_subtree(path[:idx]): expanded_tree.get_subtree(
                        path[:idx]
                    )
                    for path in possible_expansion
                    for idx in range(len(path) + 1)
                }
            )

            result.append(SolutionState(updated_constraint, expanded_tree))

        assert not limit or len(result) <= limit
        return result

    def match_universal_formulas(self, state: SolutionState) -> List[SolutionState]:
        instantiated_formulas: List[language.Formula] = []
        conjuncts = split_conjunction(state.constraint)

        for idx, universal_formula in enumerate(conjuncts):
            if not isinstance(universal_formula, language.ForallFormula):
                continue

            matches: List[Dict[language.Variable, Tuple[Path, DerivationTree]]] = [
                match
                for match in matches_for_quantified_formula(
                    universal_formula, self.grammar
                )
                if not universal_formula.is_already_matched(
                    match[universal_formula.bound_variable][1]
                )
            ]

            universal_formula_with_matches = universal_formula.add_already_matched(
                {match[universal_formula.bound_variable][1] for match in matches}
            )

            for match in matches:
                inst_formula = (
                    universal_formula_with_matches.inner_formula.substitute_expressions(
                        {
                            variable: match_tree
                            for variable, (_, match_tree) in match.items()
                        }
                    )
                )

                instantiated_formulas.append(inst_formula)
                conjuncts = list_set(conjuncts, idx, universal_formula_with_matches)

        if instantiated_formulas:
            return [
                SolutionState(
                    sc.conjunction(*instantiated_formulas) & sc.conjunction(*conjuncts),
                    state.tree,
                )
            ]
        else:
            return []

    def match_existential_formula(
        self, existential_formula_idx: int, state: SolutionState
    ) -> List[SolutionState]:
        result: List[SolutionState] = []

        conjuncts: ImmutableList[language.Formula] = tuple(
            split_conjunction(state.constraint)
        )
        existential_formula = cast(
            language.ExistsFormula, conjuncts[existential_formula_idx]
        )

        matches: List[Dict[language.Variable, Tuple[Path, DerivationTree]]] = (
            matches_for_quantified_formula(existential_formula, self.grammar)
        )

        for match in matches:
            inst_formula = existential_formula.inner_formula.substitute_expressions(
                {variable: match_tree for variable, (_, match_tree) in match.items()}
            )
            constraint = inst_formula & sc.conjunction(
                *list_del(conjuncts, existential_formula_idx)
            )
            result.append(SolutionState(constraint, state.tree))

        return result

    def eliminate_existential_formula(
        self, existential_formula_idx: int, state: SolutionState
    ) -> List[SolutionState]:
        conjuncts: ImmutableList[language.Formula] = tuple(
            split_conjunction(state.constraint)
        )
        existential_formula = cast(
            language.ExistsFormula, conjuncts[existential_formula_idx]
        )

        inserted_trees_and_bind_paths = (
            [(DerivationTree(existential_formula.bound_variable.n_type, None), {})]
            if existential_formula.bind_expression is None
            else (
                existential_formula.bind_expression.to_tree_prefix(
                    existential_formula.bound_variable.n_type, self.grammar
                )
            )
        )

        result: List[SolutionState] = []

        inserted_tree: DerivationTree
        bind_expr_paths: Dict[language.BoundVariable, Path]
        for inserted_tree, bind_expr_paths in inserted_trees_and_bind_paths:
            self.logger.debug(
                "process_insertion_results(self.canonical_grammar, %s, %s, self.graph, %s)",
                lazystr(lambda: repr(inserted_tree)),
                lazystr(lambda: repr(existential_formula.in_variable)),
                self.max_number_tree_insertion_results,
            )

            insertion_results = insert_tree(
                self.canonical_grammar,
                inserted_tree,
                existential_formula.in_variable,
                graph=self.graph,
                max_num_solutions=self.max_number_tree_insertion_results * 2,
                methods=self.tree_insertion_methods,
            )

            insertion_results = sorted(
                insertion_results,
                key=lambda t: compute_tree_closing_cost(t, self.graph),
            )
            insertion_results = insertion_results[
                : self.max_number_tree_insertion_results
            ]

            for insertion_result in insertion_results:
                replaced_path = state.tree.find_node(existential_formula.in_variable)
                resulting_tree = state.tree.replace_path(
                    replaced_path, insertion_result
                )

                tree_substitution: Dict[DerivationTree, DerivationTree] = {}
                for idx in range(len(replaced_path) + 1):
                    original_path = replaced_path[: idx - 1]
                    original_tree = state.tree.get_subtree(original_path)
                    if (
                        resulting_tree.is_valid_path(original_path)
                        and original_tree.value
                        == resulting_tree.get_subtree(original_path).value
                        and resulting_tree.get_subtree(original_path) != original_tree
                    ):
                        tree_substitution[original_tree] = resulting_tree.get_subtree(
                            original_path
                        )

                assert insertion_result.find_node(inserted_tree) is not None
                variable_substitutions = {
                    existential_formula.bound_variable: inserted_tree
                }

                if bind_expr_paths:
                    if assertions_activated():
                        dangling_bind_expr_vars = [
                            (var, path)
                            for var, path in bind_expr_paths.items()
                            if (
                                var
                                in existential_formula.bind_expression.bound_variables()
                                and insertion_result.find_node(
                                    inserted_tree.get_subtree(path)
                                )
                                is None
                            )
                        ]
                        assert not dangling_bind_expr_vars, (
                            f"Bound variables from match expression not found in tree: "
                            f"[{' ,'.join(map(repr, dangling_bind_expr_vars))}]"
                        )

                    variable_substitutions.update(
                        {
                            var: inserted_tree.get_subtree(path)
                            for var, path in bind_expr_paths.items()
                            if var
                            in existential_formula.bind_expression.bound_variables()
                        }
                    )

                instantiated_formula = (
                    existential_formula.inner_formula.substitute_expressions(
                        variable_substitutions
                    ).substitute_expressions(tree_substitution)
                )

                instantiated_original_constraint = sc.conjunction(
                    *list_del(conjuncts, existential_formula_idx)
                ).substitute_expressions(tree_substitution)

                new_tree = resulting_tree.substitute(tree_substitution)

                new_formula = (
                    instantiated_formula
                    & self.formula.substitute_expressions(
                        {self.top_constant.unwrap(): new_tree}
                    )
                    & instantiated_original_constraint
                )

                new_state = SolutionState(new_formula, new_tree)

                assert all(
                    new_state.tree.find_node(tree) is not None
                    for quantified_formula in split_conjunction(new_state.constraint)
                    if isinstance(quantified_formula, language.QuantifiedFormula)
                    for _, tree in quantified_formula.in_variable.filter(lambda t: True)
                )

                if assertions_activated() or self.debug:
                    lost_tree_predicate_arguments: List[DerivationTree] = [
                        arg
                        for invstate in self.establish_invariant(new_state)
                        for predicate_formula in get_conjuncts(invstate.constraint)
                        if isinstance(
                            predicate_formula, language.StructuralPredicateFormula
                        )
                        for arg in predicate_formula.args
                        if isinstance(arg, DerivationTree)
                        and invstate.tree.find_node(arg) is None
                    ]

                    if lost_tree_predicate_arguments:
                        previous_posititions = [
                            state.tree.find_node(t)
                            for t in lost_tree_predicate_arguments
                        ]
                        assert False, (
                            f"Dangling subtrees [{', '.join(map(repr, lost_tree_predicate_arguments))}], "
                            f"previously at positions [{', '.join(map(str, previous_posititions))}] "
                            f"in tree {repr(state.tree)} (hash: {hash(state)})."
                        )

                    lost_semantic_formula_arguments = [
                        arg
                        for invstate in self.establish_invariant(new_state)
                        for semantic_formula in get_conjuncts(new_state.constraint)
                        if isinstance(semantic_formula, language.SMTFormula)
                        for arg in semantic_formula.substitutions.values()
                        if invstate.tree.find_node(arg) is None
                    ]

                    if lost_semantic_formula_arguments:
                        previous_posititions = [
                            state.tree.find_node(t)
                            for t in lost_semantic_formula_arguments
                        ]
                        previous_posititions = [
                            p for p in previous_posititions if p is not None
                        ]
                        assert False, (
                            f"Dangling subtrees [{', '.join(map(repr, lost_semantic_formula_arguments))}], "
                            f"previously at positions [{', '.join(map(str, previous_posititions))}] "
                            f"in tree {repr(state.tree)} (hash: {hash(state)})."
                        )

                result.append(new_state)

        return result

    def eliminate_semantic_formula(
        self,
        semantic_formula: language.Formula,
        state: SolutionState,
        max_instantiations: Optional[int] = None,
    ) -> Optional[List[SolutionState]]:
        """
        Solves a semantic formula and, for each solution, substitutes the solution for
        the respective constant in each assignment of the state. Also instantiates all
        "free" constants in the given tree. The SMT solver is passed a regular
        expression approximating the language of the nonterminal of each considered
        constant. Returns an empty list for unsolvable constraints.

        :param semantic_formula: The semantic (i.e., only containing logical connectors and SMT Formulas)
        formula to solve.
        :param state: The original solution state.
        :param max_instantiations: The maximum number of solutions to ask the SMT solver for.
        :return: A list of instantiated SolutionStates.
        """

        assert all(
            isinstance(conjunct, language.SMTFormula)
            for conjunct in get_conjuncts(semantic_formula)
        )

        # NOTE: We need to cluster SMT formulas by tree substitutions. If there are two
        # formulas with a variable $var which is instantiated to different trees, we
        # need two separate solutions. If, however, $var is instantiated with the
        # *same* tree, we need one solution to both formulas together.

        smt_formulas = self.rename_instantiated_variables_in_smt_formulas(
            [
                smt_formula
                for smt_formula in get_conjuncts(semantic_formula)
                if isinstance(smt_formula, language.SMTFormula)
            ]
        )

        # Now, we also cluster formulas by common variables (and instantiated subtrees:
        # One formula might yield an instantiation of a subtree of the instantiation of
        # another formula. They need to appear in the same cluster). The solver can
        # better handle smaller constraints, and those which do not have variables in
        # common can be handled independently.

        def cluster_keys(smt_formula: language.SMTFormula):
            return (
                smt_formula.free_variables()
                | smt_formula.instantiated_variables
                | set(
                    [
                        subtree
                        for tree in smt_formula.substitutions.values()
                        for _, subtree in tree.paths()
                    ]
                )
            )

        formula_clusters: List[List[language.SMTFormula]] = cluster_by_common_elements(
            smt_formulas, cluster_keys
        )

        assert all(
            not cluster_keys(smt_formula)
            or any(smt_formula in cluster for cluster in formula_clusters)
            for smt_formula in smt_formulas
        )

        formula_clusters = [cluster for cluster in formula_clusters if cluster]
        remaining_clusters = [
            smt_formula for smt_formula in smt_formulas if not cluster_keys(smt_formula)
        ]
        if remaining_clusters:
            formula_clusters.append(remaining_clusters)

        # Note: We cannot ask for `max_instantiations` solutions for *each cluster;*
        #       this would imply that we get 10^4 solutions if `max_instantiations`
        #       is 10 and we have 4 clusters (we combine all these solutions to a
        #       product). Instead, we want 10 solutions; thus, we compute the
        #       #numCluster'th root of `max_instantiations` and ceil.
        #       For example, the ceil of the 4-root of 10 is 2, and 2^10 is 16. This
        #       is still within an acceptable range.

        solutions_per_cluster = math.ceil(
            (max_instantiations or self.max_number_smt_instantiations)
            ** (1 / len(formula_clusters))
        )

        all_solutions: List[
            List[Dict[Union[language.Constant, DerivationTree], DerivationTree]]
        ] = [
            self.solve_quantifier_free_formula(
                tuple(cluster),
                solutions_per_cluster,
            )
            for cluster in formula_clusters
        ]

        # These solutions are all independent, such that we can combine each solution
        # with all others.
        solutions: List[
            Dict[Union[language.Constant, DerivationTree], DerivationTree]
        ] = [
            functools.reduce(operator.or_, dicts)
            for dicts in itertools.product(*all_solutions)
        ]

        results = []
        # We also have to instantiate all subtrees of the substituted element.
        for solution in map(subtree_solutions, solutions):
            if solution:
                new_state = SolutionState(
                    state.constraint.substitute_expressions(solution),
                    state.tree.substitute(solution),
                )
            else:
                new_state = SolutionState(
                    language.replace_formula(
                        state.constraint, semantic_formula, sc.true()
                    ),
                    state.tree,
                )

            results.append(new_state)

        return results

    @lru_cache(100)
    def solve_quantifier_free_formula(
        self,
        smt_formulas: ImmutableList[language.SMTFormula],
        max_instantiations: Optional[int] = None,
    ) -> List[Dict[language.Constant | DerivationTree, DerivationTree]]:
        """
        Attempts to solve the given SMT-LIB formulas by calling Z3.

        Note that this function does not unify variables pointing to the same derivation
        trees. E.g., a solution may be returned for the formula `var_1 = "a" and
        var_2 = "b"`, though `var_1` and `var_2` point to the same `<var>` tree as
        defined by their substitutions maps. Unification is performed in
        `eliminate_all_semantic_formulas`.

        :param smt_formulas: The SMT-LIB formulas to solve.
        :param max_instantiations: The maximum number of instantiations to produce.
        :return: A (possibly empty) list of solutions.
        """

        # If any SMT formula refers to *sub*trees in the instantiations of other SMT
        # formulas, we have to instantiate those first.
        priority_formulas = smt_formulas_referring_to_subtrees(smt_formulas)

        if priority_formulas:
            smt_formulas = priority_formulas
            assert not smt_formulas_referring_to_subtrees(smt_formulas)

        tree_substitutions = reduce(
            lambda d1, d2: d1 | d2,
            [smt_formula.substitutions for smt_formula in smt_formulas],
            {},
        )

        constants = reduce(
            lambda d1, d2: d1 | d2,
            [
                smt_formula.free_variables() | smt_formula.instantiated_variables
                for smt_formula in smt_formulas
            ],
            set(),
        )

        solutions: List[
            Dict[Union[language.Constant, DerivationTree], DerivationTree]
        ] = []
        internal_solutions: List[Dict[language.Constant, z3.StringVal]] = []

        num_instantiations = max_instantiations or self.max_number_smt_instantiations
        for _ in range(num_instantiations):
            (
                solver_result,
                maybe_model,
            ) = self.solve_smt_formulas_with_language_constraints(
                constants,
                tuple([smt_formula.formula for smt_formula in smt_formulas]),
                tree_substitutions,
                internal_solutions,
            )

            if solver_result != z3.sat:
                if not solutions:
                    return []
                else:
                    return solutions

            assert maybe_model is not None

            new_solution = {
                tree_substitutions.get(constant, constant): maybe_model[constant]
                for constant in constants
            }

            new_internal_solution = {
                constant: z3.StringVal(str(maybe_model[constant]))
                for constant in constants
            }

            if new_solution in solutions:
                # This can happen for trivial solutions, i.e., if the formula is
                # logically valid. Then, the assignment for that constant will
                # always be {}
                return solutions
            else:
                solutions.append(new_solution)
                if new_internal_solution:
                    internal_solutions.append(new_internal_solution)
                else:
                    # Again, for a trivial solution (e.g., True), the assignment
                    # can be empty.
                    break

        return solutions

    def solve_smt_formulas_with_language_constraints(
        self,
        variables: Set[language.Variable],
        smt_formulas: ImmutableList[z3.BoolRef],
        tree_substitutions: Dict[language.Variable, DerivationTree],
        solutions_to_exclude: List[Dict[language.Variable, z3.StringVal]],
    ) -> Tuple[z3.CheckSatResult, Dict[language.Variable, DerivationTree]]:
        # We disable optimized Z3 queries if the SMT formulas contain "too concrete"
        # substitutions, that is, substitutions with a tree that is not merely an
        # open leaf. Example: we have a constrained `str.len(<chars>) < 10` and a
        # tree `<char><char>`; only the concrete length "10" is possible then. In fact,
        # we could simply finish of the tree and check the constraint, or restrict the
        # custom tree generation to admissible lengths, but we stay general here. The
        # SMT solution is more robust.

        if self.enable_optimized_z3_queries and not any(
            substitution.children for substitution in tree_substitutions.values()
        ):
            vars_in_context = self.infer_variable_contexts(variables, smt_formulas)
            length_vars = vars_in_context["length"]
            int_vars = vars_in_context["int"]
            flexible_vars = vars_in_context["flexible"]
        else:
            length_vars = set()
            int_vars = set()
            flexible_vars = set(variables)

        # Add language constraints for "flexible" variables
        formulas: List[z3.BoolRef] = self.generate_language_constraints(
            flexible_vars, tree_substitutions
        )

        # Create fresh variables for `str.len` and `str.to.int` variables.
        all_variables = set(variables)
        fresh_var_map: Dict[language.Variable, z3.ExprRef] = {}
        for var in length_vars | int_vars:
            fresh = fresh_constant(
                all_variables,
                language.Constant(var.name, "NOT-NEEDED"),
            )
            fresh_var_map[var] = z3.Int(fresh.name)

        # In `smt_formulas`, we replace all `length(...)` terms for "length variables"
        # with the corresponding fresh variable.
        replacement_map: Dict[z3.ExprRef, z3.ExprRef] = {
            expr: fresh_var_map[
                get_elem_by_equivalence(
                    expr.children()[0],
                    length_vars | int_vars,
                    lambda e1, e2: e1 == e2.to_smt(),
                )
            ]
            for formula in smt_formulas
            for expr in visit_z3_expr(formula)
            if expr.decl().kind() in {z3.Z3_OP_SEQ_LENGTH, z3.Z3_OP_STR_TO_INT}
            and expr.children()[0] in {var.to_smt() for var in length_vars | int_vars}
        }

        # Perform substitution, add formulas
        formulas.extend(
            [
                cast(z3.BoolRef, z3_subst(formula, replacement_map))
                for formula in smt_formulas
            ]
        )

        # Lengths must be positive
        formulas.extend(
            [
                cast(
                    z3.BoolRef,
                    replacement_map[z3.Length(length_var.to_smt())] >= z3.IntVal(0),
                )
                for length_var in length_vars
            ]
        )

        # Add custom intervals for int variables
        for int_var in int_vars:
            if int_var.n_type == language.Variable.NUMERIC_NTYPE:
                # "NUM" variables range over the full int domain
                continue

            regex = self.extract_regular_expression(int_var.n_type)
            maybe_intervals = numeric_intervals_from_regex(regex)
            repl_var = replacement_map[z3.StrToInt(int_var.to_smt())]
            maybe_intervals.map(
                tap(
                    lambda intervals: formulas.append(
                        z3_or(
                            [
                                z3.And(
                                    (
                                        repl_var >= z3.IntVal(interval[0])
                                        if interval[0] > -sys.maxsize
                                        else z3.BoolVal(True)
                                    ),
                                    (
                                        repl_var <= z3.IntVal(interval[1])
                                        if interval[1] < sys.maxsize
                                        else z3.BoolVal(True)
                                    ),
                                )
                                for interval in intervals
                            ]
                        )
                    )
                )
            )

        for prev_solution in solutions_to_exclude:
            prev_solution_formula = z3_and(
                [
                    self.previous_solution_formula(
                        var, string_val, fresh_var_map, length_vars, int_vars
                    )
                    for var, string_val in prev_solution.items()
                ]
            )

            formulas.append(z3.Not(prev_solution_formula))

        sat_result, maybe_model = z3_solve(formulas)

        if sat_result != z3.sat:
            return sat_result, {}

        assert maybe_model is not None

        return sat_result, {
            var: self.extract_model_value(
                var, maybe_model, fresh_var_map, length_vars, int_vars
            )
            for var in variables
        }

    @staticmethod
    def previous_solution_formula(
        var: language.Variable,
        string_val: z3.StringVal,
        fresh_var_map: Dict[language.Variable, z3.ExprRef],
        length_vars: Set[language.Variable],
        int_vars: Set[language.Variable],
    ) -> z3.BoolRef:
        """
        Computes a formula describing the previously found solution
        :code:`var == string_val` for an :class:`~isla.language.SMTFormula`.
        Considers the special cases that :code:`var` is a "length" or "int"
        variable, i.e., occurred only in these contexts in the formula this
        solution is about.

        >>> x = language.Variable("x", "<X>")
        >>> ISLaSolver.previous_solution_formula(
        ...     x, z3.StringVal("val"), {}, set(), set())
        x == "val"

        >>> ISLaSolver.previous_solution_formula(
        ...     x, z3.StringVal("val"), {x: z3.Int("x_0")}, {x}, set())
        x_0 == 3

        >>> ISLaSolver.previous_solution_formula(
        ...     x, z3.StringVal("10"), {x: z3.Int("x_0")}, set(), {x})
        x_0 == 10

        >>> x = language.Variable("x", language.Variable.NUMERIC_NTYPE)
        >>> ISLaSolver.previous_solution_formula(
        ...     x, z3.StringVal("10"), {x: z3.Int("x_0")}, set(), {x})
        x_0 == 10

        A "numeric" variable (of "NUM" type) is expected to always be an int variable,
        which also needs to be reflected in its inclusion in :code:`fresh_var_map`.

        >>> x = language.Variable("x", language.Variable.NUMERIC_NTYPE)
        >>> ISLaSolver.previous_solution_formula(
        ...     x, z3.StringVal("10"), {}, set(), set())
        Traceback (most recent call last):
        ...
        AssertionError

        :param var: The variable the solution is for.
        :param string_val: The solution for :code:`var`.
        :param fresh_var_map: A map from variables to fresh variables for "length" or
                              "int" variables.
        :param length_vars: The "length" variables.
        :param int_vars: The "int" variables.
        :return: An equation describing the previous solution.
        """

        if var in int_vars:
            return z3_eq(
                fresh_var_map[var],
                z3.IntVal(int(smt_string_val_to_string(string_val))),
            )
        elif var in length_vars:
            return z3_eq(
                fresh_var_map[var],
                z3.IntVal(len(smt_string_val_to_string(string_val))),
            )
        else:
            assert not var.is_numeric()
            return z3_eq(var.to_smt(), string_val)

    def safe_create_fixed_length_tree(
        self,
        var: language.Variable,
        model: z3.ModelRef,
        fresh_var_map: Dict[language.Variable, z3.ExprRef],
    ) -> DerivationTree:
        """
        Creates a :class:`~isla.derivation_tree.DerivationTree` for :code:`var` such
        that the type of the tree fits to :code:`var` and the length of its string
        representation fits to the length in :code:`model` for the fresh variable in
        :code:`fresh_var_map`. For example:

        >>> grammar = {
        ...     "<start>": ["<X>"],
        ...     "<X>": ["x", "x<X>"],
        ... }
        >>> x = language.Variable("x", "<X>")
        >>> x_0 = z3.Int("x_0")
        >>> f = z3_eq(x_0, z3.IntVal(5))
        >>> z3_solver = z3.Solver()
        >>> z3_solver.add(f)
        >>> z3_solver.check()
        sat
        >>> model = z3_solver.model()
        >>> solver = ISLaSolver(grammar)
        >>> tree = solver.safe_create_fixed_length_tree(x, model, {x: x_0})
        >>> tree.value
        '<X>'
        >>> str(tree)
        'xxxxx'

        :param var: The variable to create a
                    :class:`~isla.derivation_tree.DerivationTree` object for.
        :param model: The Z3 model to extract a solution to the length constraint.
        :param fresh_var_map: A map including a mapping :code:`var` -> :code:`var_0`,
                              where :code:`var_0` is an integer-valued variale included
                              in :code:`model`.
        :return: A tree of the type of :code:`var` and length as specified in
                :code:`model`.
        """

        assert var in fresh_var_map
        assert fresh_var_map[var].decl() in model.decls()

        fixed_length_tree = create_fixed_length_tree(
            start=var.n_type,
            canonical_grammar=self.canonical_grammar,
            target_length=model[fresh_var_map[var]].as_long(),
        )

        if fixed_length_tree is None:
            raise RuntimeError(
                f"Could not create a tree with the start symbol '{var.n_type}' "
                + f"of length {model[fresh_var_map[var]].as_long()}; try "
                + "running the solver without optimized Z3 queries or make "
                + "sure that lengths are restricted to syntactically valid "
                + "ones (according to the grammar).",
            )

        return fixed_length_tree

    def extract_model_value(
        self,
        var: language.Variable,
        model: z3.ModelRef,
        fresh_var_map: Dict[language.Variable, z3.ExprRef],
        length_vars: Set[language.Variable],
        int_vars: Set[language.Variable],
    ) -> DerivationTree:
        r"""
        Extracts a value for :code:`var` from :code:`model`. Considers the following
        special cases:

        Numeric Variables
            Returns a closed derivation tree of one node with a string representation
            of the numeric solution.

        "Length" Variables
            Returns a string of the length corresponding to the model and
            :code:`fresh_var_map`, see also
            :meth:`~isla.solver.ISLaSolver.safe_create_fixed_length_tree()`.

        "Int" Variables
            Tries to parse the numeric solution from the model (obtained via
            :code:`fresh_var_map`) into the type of :code:`var` and returns the
            corresponding derivation tree.

        >>> grammar = {
        ...     "<start>": ["<A>"],
        ...     "<A>": ["<X><Y>"],
        ...     "<X>": ["x", "x<X>"],
        ...     "<Y>": ["<digit>", "<digit><Y>"],
        ...     "<digit>": ["0", "1", "2", "3", "4", "5", "6", "7", "8", "9"],
        ... }
        >>> solver = ISLaSolver(grammar)

        **Numeric Variables:**

        >>> n = language.Variable("n", language.Variable.NUMERIC_NTYPE)
        >>> f = z3_eq(z3.StrToInt(n.to_smt()), z3.IntVal(15))
        >>> z3_solver = z3.Solver()
        >>> z3_solver.add(f)
        >>> z3_solver.check()
        sat
        >>> model = z3_solver.model()
        >>> DerivationTree.next_id = 1
        >>> solver.extract_model_value(n, model, {}, set(), set())
        DerivationTree('15', (), id=1)

        For a trivially true solution on numeric variables, we return a random number:

        >>> f = z3_eq(n.to_smt(), n.to_smt())
        >>> z3_solver = z3.Solver()
        >>> z3_solver.add(f)
        >>> z3_solver.check()
        sat

        >>> model = z3_solver.model()
        >>> DerivationTree.next_id = 1
        >>> random.seed(0)
        >>> solver.extract_model_value(n, model, {n: n.to_smt()}, set(), {n})
        DerivationTree('-2116850434379610162', (), id=1)

        **"Length" Variables:**

        >>> x = language.Variable("x", "<X>")
        >>> x_0 = z3.Int("x_0")
        >>> f = z3_eq(x_0, z3.IntVal(3))
        >>> z3_solver = z3.Solver()
        >>> z3_solver.add(f)
        >>> z3_solver.check()
        sat
        >>> model = z3_solver.model()
        >>> result = solver.extract_model_value(x, model, {x: x_0}, {x}, set())
        >>> result.value
        '<X>'
        >>> str(result)
        'xxx'

        **"Int" Variables:**

        >>> y = language.Variable("y", "<Y>")
        >>> y_0 = z3.Int("y_0")
        >>> f = z3_eq(y_0, z3.IntVal(5))
        >>> z3_solver = z3.Solver()
        >>> z3_solver.add(f)
        >>> z3_solver.check()
        sat
        >>> model = z3_solver.model()
        >>> DerivationTree.next_id = 1
        >>> solver.extract_model_value(y, model, {y: y_0}, set(), {y})
        DerivationTree('<Y>', (DerivationTree('<digit>', (DerivationTree('5', (), id=1),), id=2),), id=3)

        **"Flexible" Variables:**

        >>> f = z3_eq(x.to_smt(), z3.StringVal("xxxxx"))
        >>> z3_solver = z3.Solver()
        >>> z3_solver.add(f)
        >>> z3_solver.check()
        sat
        >>> model = z3_solver.model()
        >>> result = solver.extract_model_value(x, model, {}, set(), set())
        >>> result.value
        '<X>'
        >>> str(result)
        'xxxxx'

        **Special Number Formats**

        It may happen that the solver returns, e.g., "1" as a solution for an int
        variable, but the grammar only recognizes "+001". We support this case, i.e.,
        an optional "+" and optional 0 padding; an optional 0 padding for negative
        numbers is also supported.

        >>> grammar = {
        ...     "<start>": ["<int>"],
        ...     "<int>": ["<sign>00<leaddigit><digits>"],
        ...     "<sign>": ["-", "+"],
        ...     "<digits>": ["", "<digit><digits>"],
        ...     "<digit>": list("0123456789"),
        ...     "<leaddigit>": list("123456789"),
        ... }
        >>> solver = ISLaSolver(grammar)

        >>> i = language.Variable("i", "<int>")
        >>> i_0 = z3.Int("i_0")
        >>> f = z3_eq(i_0, z3.IntVal(5))

        >>> z3_solver = z3.Solver()
        >>> z3_solver.add(f)
        >>> z3_solver.check()
        sat

        The following test works when run from the IDE, but unfortunately not when
        started from CI/the `test_doctests.py` script. Thus, we only show it as code
        block (we added a unit test as a replacement)::

            model = z3_solver.model()
            print(solver.extract_model_value(i, model, {i: i_0}, set(), {i}))
            # Prints: +005

        :param var: The variable for which to extract a solution from the model.
        :param model: The model containing the solution.
        :param fresh_var_map: A map from variables to fresh symbols for "length" and
                              "int" variables.
        :param length_vars: The set of "length" variables.
        :param int_vars: The set of "int" variables.
        :return: A :class:`~isla.derivation_tree.DerivationTree` object corresponding
                 to the solution in :code:`model`.
        """

        f_flex_vars = self.extract_model_value_flexible_var
        f_int_vars = partial(self.extract_model_value_int_var, f_flex_vars)
        f_length_vars = partial(self.extract_model_value_length_var, f_int_vars)
        f_num_vars = partial(self.extract_model_value_numeric_var, f_length_vars)

        return f_num_vars(var, model, fresh_var_map, length_vars, int_vars)

    ExtractModelValueFallbackType = Callable[
        [
            language.Variable,
            z3.ModelRef,
            Dict[language.Variable, z3.ExprRef],
            Set[language.Variable],
            Set[language.Variable],
        ],
        DerivationTree,
    ]

    def extract_model_value_numeric_var(
        self,
        fallback: ExtractModelValueFallbackType,
        var: language.Variable,
        model: z3.ModelRef,
        fresh_var_map: Dict[language.Variable, z3.ExprRef],
        length_vars: Set[language.Variable],
        int_vars: Set[language.Variable],
    ) -> DerivationTree:
        """
        Addresses the case of numeric variables from
        :meth:`~isla.solver.ISLaSolver.extract_model_value`.

        :param fallback: The function to call if this function is not responsible.
        :param var: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :param model: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :param fresh_var_map: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :param length_vars: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :param int_vars: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :return: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        """
        if not var.is_numeric():
            return fallback(var, model, fresh_var_map, length_vars, int_vars)

        z3_var = z3.String(var.name)
        if z3_var.decl() in model.decls():
            model_value = model[z3_var]
        else:
            assert var in int_vars
            assert var in fresh_var_map

            model_value = model[fresh_var_map[var]]

            if model_value is None:
                # This can happen for universally true formulas, e.g., `x = x`.
                # In that case, we return a random integer.
                model_value = z3.IntVal(random.randint(-sys.maxsize, sys.maxsize))

        assert (
            model_value is not None
        ), f"No solution for variable {var} found in model {model}"

        string_value = smt_string_val_to_string(model_value)
        assert string_value
        assert (
            string_value.isnumeric()
            or string_value[0] == "-"
            and string_value[1:].isnumeric()
        )

        return DerivationTree(string_value, ())

    def extract_model_value_length_var(
        self,
        fallback: ExtractModelValueFallbackType,
        var: language.Variable,
        model: z3.ModelRef,
        fresh_var_map: Dict[language.Variable, z3.ExprRef],
        length_vars: Set[language.Variable],
        int_vars: Set[language.Variable],
    ) -> DerivationTree:
        """
        Addresses the case of length variables from
        :meth:`~isla.solver.ISLaSolver.extract_model_value`.

        :param fallback: The function to call if this function is not responsible.
        :param var: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :param model: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :param fresh_var_map: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :param length_vars: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :param int_vars: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :return: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        """
        if var not in length_vars:
            return fallback(var, model, fresh_var_map, length_vars, int_vars)

        return self.safe_create_fixed_length_tree(var, model, fresh_var_map)

    def extract_model_value_int_var(
        self,
        fallback: ExtractModelValueFallbackType,
        var: language.Variable,
        model: z3.ModelRef,
        fresh_var_map: Dict[language.Variable, z3.ExprRef],
        length_vars: Set[language.Variable],
        int_vars: Set[language.Variable],
    ) -> DerivationTree:
        """
        Addresses the case of int variables from
        :meth:`~isla.solver.ISLaSolver.extract_model_value`.

        :param fallback: The function to call if this function is not responsible.
        :param var: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :param model: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :param fresh_var_map: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :param length_vars: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :param int_vars: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :return: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        """
        if var not in int_vars:
            return fallback(var, model, fresh_var_map, length_vars, int_vars)

        str_model_value = model[fresh_var_map[var]].as_string()

        try:
            int_model_value = int(str_model_value)
        except ValueError:
            raise RuntimeError(f"Value {str_model_value} for {var} is not a number")

        var_type = var.n_type

        try:
            return self.parse(
                str(int_model_value),
                var_type,
                silent=True,
            )
        except SyntaxError:
            # This may happen, e.g, with padded values: Only "01" is a valid
            # solution, but not "1". Similarly, a grammar may expect "+1", but
            # "1" is returned by the solver. We support the number format
            # `[+-]0*<digits>`. Whenever the grammar recognizes at least this
            # set for the nonterminal in question, we return a derivation tree.
            # Otherwise, a RuntimeError is raised.

            z3_solver = z3.Solver()
            z3_solver.set("timeout", 300)

            maybe_plus_re = z3.Option(z3.Re("+"))
            zeroes_padding_re = z3.Star(z3.Re("0"))

            # TODO: Ensure symbols are fresh
            maybe_plus_var = z3.String("__plus")
            zeroes_padding_var = z3.String("__padding")

            z3_solver.add(z3.InRe(maybe_plus_var, maybe_plus_re))
            z3_solver.add(z3.InRe(zeroes_padding_var, zeroes_padding_re))

            z3_solver.add(
                z3.InRe(
                    z3.Concat(
                        maybe_plus_var if int_model_value >= 0 else z3.StringVal("-"),
                        zeroes_padding_var,
                        z3.StringVal(
                            str_model_value
                            if int_model_value >= 0
                            else str(-int_model_value)
                        ),
                    ),
                    self.extract_regular_expression(var.n_type),
                )
            )

            if z3_solver.check() != z3.sat:
                raise RuntimeError(
                    "Could not parse a numeric solution "
                    + f"({str_model_value}) for variable "
                    + f"{var} of type '{var.n_type}'; try "
                    + "running the solver without optimized Z3 queries or make "
                    + "sure that ranges are restricted to syntactically valid "
                    + "ones (according to the grammar).",
                )

            return self.parse(
                (
                    z3_solver.model()[maybe_plus_var].as_string()
                    if int_model_value >= 0
                    else "-"
                )
                + z3_solver.model()[zeroes_padding_var].as_string()
                + (str_model_value if int_model_value >= 0 else str(-int_model_value)),
                var.n_type,
            )

    def extract_model_value_flexible_var(
        self,
        var: language.Variable,
        model: z3.ModelRef,
        fresh_var_map: Dict[language.Variable, z3.ExprRef],
        length_vars: Set[language.Variable],
        int_vars: Set[language.Variable],
    ) -> DerivationTree:
        """
        Addresses the case of "flexible" variables from
        :meth:`~isla.solver.ISLaSolver.extract_model_value`.

        :param fallback: The function to call if this function is not responsible.
        :param var: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :param model: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :param fresh_var_map: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :param length_vars: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :param int_vars: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        :return: See :meth:`~isla.solver.ISLaSolver.extract_model_value`.
        """

        return self.parse(
            smt_string_val_to_string(model[z3.String(var.name)]),
            var.n_type,
        )

    @staticmethod
    def infer_variable_contexts(
        variables: Set[language.Variable], smt_formulas: ImmutableList[z3.BoolRef]
    ) -> Dict[str, Set[language.Variable]]:
        """
        Divides the given variables into

        1. those that occur only in :code:`length(...)` contexts,
        2. those that occur only in :code:`str.to.int(...)` contexts, and
        3. "flexible" constants occurring in other/various contexts.

        >>> x = language.Variable("x", "<X>")
        >>> y = language.Variable("y", "<Y>")

        Two variables in an arbitrary context.

        >>> f = z3_eq(x.to_smt(), y.to_smt())
        >>> contexts = ISLaSolver.infer_variable_contexts({x, y}, (f,))
        >>> contexts["length"]
        set()
        >>> contexts["int"]
        set()
        >>> contexts["flexible"] == {language.Variable("x", "<X>"), language.Variable("y", "<Y>")}
        True

        Variable x occurs in a length context, variable y in an arbitrary one.

        >>> f = z3.And(
        ...     z3.Length(x.to_smt()) > z3.IntVal(10),
        ...     z3_eq(y.to_smt(), z3.StringVal("y")))
        >>> ISLaSolver.infer_variable_contexts({x, y}, (f,))
        {'length': {Variable("x", "<X>")}, 'int': set(), 'flexible': {Variable("y", "<Y>")}}

        Variable x occurs in a length context, y does not occur.

        >>> f = z3.Length(x.to_smt()) > z3.IntVal(10)
        >>> ISLaSolver.infer_variable_contexts({x, y}, (f,))
        {'length': {Variable("x", "<X>")}, 'int': set(), 'flexible': {Variable("y", "<Y>")}}

        Variables x and y both occur in a length context.

        >>> f = z3.Length(x.to_smt()) > z3.Length(y.to_smt())
        >>> contexts = ISLaSolver.infer_variable_contexts({x, y}, (f,))
        >>> contexts["length"] == {language.Variable("x", "<X>"), language.Variable("y", "<Y>")}
        True
        >>> contexts["int"]
        set()
        >>> contexts["flexible"]
        set()

        Variable x occurs in a :code:`str.to.int` context.

        >>> f = z3.StrToInt(x.to_smt()) > z3.IntVal(17)
        >>> ISLaSolver.infer_variable_contexts({x}, (f,))
        {'length': set(), 'int': {Variable("x", "<X>")}, 'flexible': set()}

        Now, x also occurs in a different context; it's "flexible" now.

        >>> f = z3.And(
        ...     z3.StrToInt(x.to_smt()) > z3.IntVal(17),
        ...     z3_eq(x.to_smt(), z3.StringVal("17")))
        >>> ISLaSolver.infer_variable_contexts({x}, (f,))
        {'length': set(), 'int': set(), 'flexible': {Variable("x", "<X>")}}

        :param variables: The constants to divide/filter from.
        :param smt_formulas: The SMT formulas to consider in the filtering.
        :return: A pair of constants occurring in `str.len` contexts, and the
        remaining ones. The union of both sets equals `variables`, and both sets
        are disjoint.
        """

        parent_relationships = reduce(
            merge_dict_of_sets,
            [parent_relationships_in_z3_expr(formula) for formula in smt_formulas],
            {},
        )

        contexts: Dict[language.Variable, Set[int]] = {
            var: {
                expr.decl().kind()
                for expr in parent_relationships.get(var.to_smt(), set())
            }
            or {-1}
            for var in variables
        }

        # The set `length_vars` consists of all variables that only occur in
        # `str.len(...)` context.
        length_vars: Set[language.Variable] = {
            var
            for var in variables
            if all(context == z3.Z3_OP_SEQ_LENGTH for context in contexts[var])
        }

        # The set `int_vars` consists of all variables that only occur in
        # `str.to.int(...)` context.
        int_vars: Set[language.Variable] = {
            var
            for var in variables
            if all(context == z3.Z3_OP_STR_TO_INT for context in contexts[var])
        }

        # "Flexible" variables are the remaining ones.
        flexible_vars = variables.difference(length_vars).difference(int_vars)

        return {"length": length_vars, "int": int_vars, "flexible": flexible_vars}

    def generate_language_constraints(
        self,
        constants: Iterable[language.Variable],
        tree_substitutions: Dict[language.Variable, DerivationTree],
    ) -> List[z3.BoolRef]:
        formulas: List[z3.BoolRef] = []
        for constant in constants:
            if constant.is_numeric():
                regex = z3.Union(
                    z3.Re("0"),
                    z3.Concat(z3.Range("1", "9"), z3.Star(z3.Range("0", "9"))),
                )
            elif constant in tree_substitutions:
                # We have a more concrete shape of the desired instantiation available
                regexes = [
                    (
                        self.extract_regular_expression(t)
                        if is_nonterminal(t)
                        else z3.Re(t)
                    )
                    for t in split_str_with_nonterminals(
                        str(tree_substitutions[constant])
                    )
                ]
                assert regexes
                regex = z3.Concat(*regexes) if len(regexes) > 1 else regexes[0]
            else:
                regex = self.extract_regular_expression(constant.n_type)

            formulas.append(z3.InRe(z3.String(constant.name), regex))

        return formulas

    def rename_instantiated_variables_in_smt_formulas(self, smt_formulas):
        old_smt_formulas = smt_formulas
        smt_formulas = []
        for subformula in old_smt_formulas:
            subst_var: language.Variable
            subst_tree: DerivationTree

            new_smt_formula: z3.BoolRef = subformula.formula
            new_substitutions = subformula.substitutions
            new_instantiated_variables = subformula.instantiated_variables

            for subst_var, subst_tree in subformula.substitutions.items():
                new_name = f"{subst_tree.value}_{subst_tree.id}"
                new_var = language.BoundVariable(new_name, subst_var.n_type)

                new_smt_formula = cast(
                    z3.BoolRef,
                    z3_subst(new_smt_formula, {subst_var.to_smt(): new_var.to_smt()}),
                )
                new_substitutions = {
                    new_var if k == subst_var else k: v
                    for k, v in new_substitutions.items()
                }
                new_instantiated_variables = {
                    new_var if v == subst_var else v for v in new_instantiated_variables
                }

            smt_formulas.append(
                language.SMTFormula(
                    new_smt_formula,
                    *subformula.free_variables_,
                    instantiated_variables=new_instantiated_variables,
                    substitutions=new_substitutions,
                )
            )

        return smt_formulas

    def process_new_states(
        self, new_states: List[SolutionState]
    ) -> List[DerivationTree]:
        return [
            tree
            for new_state in new_states
            for tree in self.process_new_state(new_state)
        ]

    def process_new_state(self, new_state: SolutionState) -> List[DerivationTree]:
        new_state = self.instantiate_structural_predicates(new_state)
        new_states = self.establish_invariant(new_state)
        new_states = [
            self.remove_nonmatching_universal_quantifiers(new_state)
            for new_state in new_states
        ]
        new_states = [
            self.remove_infeasible_universal_quantifiers(new_state)
            for new_state in new_states
        ]

        if self.activate_unsat_support and not self.currently_unsat_checking:
            self.currently_unsat_checking = True

            for new_state in list(new_states):
                if new_state.constraint == sc.true():
                    continue

                # Remove states with unsatisfiable SMT-LIB formulas.
                if any(
                    isinstance(f, language.SMTFormula)
                    for f in split_conjunction(new_state.constraint)
                ) and not is_successful(
                    self.eliminate_all_semantic_formulas(
                        new_state, max_instantiations=1
                    ).bind(lambda a: Some(a) if a else Nothing)
                ):
                    new_states.remove(new_state)
                    self.logger.debug(
                        "Dropping state %s, unsatisfiable SMT formulas", new_state
                    )

                # Remove states with unsatisfiable existential formulas.
                existential_formulas = [
                    f
                    for f in split_conjunction(new_state.constraint)
                    if isinstance(f, language.ExistsFormula)
                ]
                for existential_formula in existential_formulas:
                    old_start_time = self.start_time
                    old_timeout_seconds = self.timeout_seconds
                    old_queue = list(self.queue)
                    old_solutions = list(self.solutions)

                    self.queue = []
                    self.solutions = []
                    check_state = SolutionState(existential_formula, new_state.tree)
                    heapq.heappush(self.queue, (0, check_state))
                    self.start_time = int(time.time())
                    self.timeout_seconds = 2

                    try:
                        self.solve()
                    except StopIteration:
                        new_states.remove(new_state)
                        self.logger.debug(
                            "Dropping state %s, unsatisfiable existential formula %s",
                            new_state,
                            existential_formula,
                        )
                        break
                    finally:
                        self.start_time = old_start_time
                        self.timeout_seconds = old_timeout_seconds
                        self.queue = old_queue
                        self.solutions = old_solutions

            self.currently_unsat_checking = False

        assert all(
            state.tree.find_node(tree) is not None
            for state in new_states
            for quantified_formula in split_conjunction(state.constraint)
            if isinstance(quantified_formula, language.QuantifiedFormula)
            for _, tree in quantified_formula.in_variable.filter(lambda t: True)
        )

        solution_trees = [
            new_state.tree
            for new_state in new_states
            if self.state_is_valid_or_enqueue(new_state)
        ]

        for tree in solution_trees:
            self.cost_computer.signal_tree_output(tree)

        return solution_trees

    def state_is_valid_or_enqueue(self, state: SolutionState) -> bool:
        """
        Returns True if the given state is valid, such that it can be yielded. Returns False and enqueues the state
        if the state is not yet complete, otherwise returns False and discards the state.
        """

        if state.complete():
            for _, subtree in state.tree.paths():
                if subtree.children:
                    self.seen_coverages.add(
                        expansion_key(subtree.value, subtree.children)
                    )

            assert state.formula_satisfied(self.grammar).is_true()
            return True

        # Helps in debugging below assertion:
        # [(predicate_formula, [
        #     arg for arg in predicate_formula.args
        #     if isinstance(arg, DerivationTree) and not state.tree.find_node(arg)])
        #  for predicate_formula in get_conjuncts(state.constraint)
        #  if isinstance(predicate_formula, language.StructuralPredicateFormula)]

        self.assert_no_dangling_predicate_argument_trees(state)
        self.assert_no_dangling_smt_formula_argument_trees(state)

        if (
            self.enforce_unique_trees_in_queue
            and state.tree.structural_hash() in self.tree_hashes_in_queue
        ):
            # Some structures can arise as well from tree insertion (existential
            # quantifier elimination) and expansion; also, tree insertion can yield
            # different trees that have intersecting expansions. We drop those to output
            # more diverse solutions (numbers for SMT solutions and free nonterminals
            # are configurable, so you get more outputs by playing with those!).
            self.logger.debug("Discarding state %s, tree already in queue", state)
            return False

        if hash(state) in self.state_hashes_in_queue:
            self.logger.debug("Discarding state %s, already in queue", state)
            return False

        if self.propositionally_unsatisfiable(state.constraint):
            self.logger.debug("Discarding state %s", state)
            return False

        state = SolutionState(
            state.constraint, state.tree, level=self.current_level + 1
        )

        self.recompute_costs()

        cost = self.compute_cost(state)
        heapq.heappush(self.queue, (cost, state))
        self.tree_hashes_in_queue.add(state.tree.structural_hash())
        self.state_hashes_in_queue.add(hash(state))

        if self.debug:
            self.state_tree[self.current_state].append(state)
            self.costs[state] = cost

        self.logger.debug(
            "Pushing new state (%s, %s) (hash %d, cost %f)",
            state.constraint,
            state.tree.to_string(show_open_leaves=True, show_ids=True),
            hash(state),
            cost,
        )
        self.logger.debug("Queue length: %d", len(self.queue))
        if len(self.queue) % 100 == 0:
            self.logger.info("Queue length: %d", len(self.queue))

        return False

    def recompute_costs(self):
        if self.step_cnt % 400 != 0 or self.step_cnt <= self.last_cost_recomputation:
            return

        self.last_cost_recomputation = self.step_cnt
        self.logger.info(
            f"Recomputing costs in queue after {self.step_cnt} solver steps"
        )
        old_queue = list(self.queue)
        self.queue = []
        for _, state in old_queue:
            cost = self.compute_cost(state)
            heapq.heappush(self.queue, (cost, state))

    def assert_no_dangling_smt_formula_argument_trees(
        self, state: SolutionState
    ) -> None:
        if not assertions_activated() and not self.debug:
            return

        dangling_smt_formula_argument_trees = [
            (smt_formula, arg)
            for smt_formula in language.FilterVisitor(
                lambda f: isinstance(f, language.SMTFormula)
            ).collect(state.constraint)
            for arg in cast(language.SMTFormula, smt_formula).substitutions.values()
            if isinstance(arg, DerivationTree) and state.tree.find_node(arg) is None
        ]

        if dangling_smt_formula_argument_trees:
            message = "Dangling SMT formula arguments: ["
            message += ", ".join(
                [
                    str(f) + ", " + repr(a)
                    for f, a in dangling_smt_formula_argument_trees
                ]
            )
            message += "]"
            assert False, message

    def assert_no_dangling_predicate_argument_trees(self, state: SolutionState) -> None:
        if not assertions_activated() and not self.debug:
            return

        dangling_predicate_argument_trees = [
            (predicate_formula, arg)
            for predicate_formula in language.FilterVisitor(
                lambda f: isinstance(f, language.StructuralPredicateFormula)
            ).collect(state.constraint)
            for arg in cast(language.StructuralPredicateFormula, predicate_formula).args
            if isinstance(arg, DerivationTree) and state.tree.find_node(arg) is None
        ]

        if dangling_predicate_argument_trees:
            message = "Dangling predicate arguments: ["
            message += ", ".join(
                [str(f) + ", " + repr(a) for f, a in dangling_predicate_argument_trees]
            )
            message += "]"
            assert False, message

    def propositionally_unsatisfiable(self, formula: language.Formula) -> bool:
        return formula == sc.false()

        # NOTE: Deactivated propositional check for performance reasons
        # z3_formula = language.isla_to_smt_formula(formula, replace_untranslatable_with_predicate=True)
        # solver = z3.Solver()
        # solver.add(z3_formula)
        # return solver.check() == z3.unsat

    def establish_invariant(self, state: SolutionState) -> List[SolutionState]:
        clauses = to_dnf_clauses(convert_to_nnf(state.constraint))
        return [
            SolutionState(reduce(lambda a, b: a & b, clause, sc.true()), state.tree)
            for clause in clauses
        ]

    def compute_cost(self, state: SolutionState) -> float:
        if state.constraint == sc.true():
            return 0

        return self.cost_computer.compute_cost(state)

    def remove_nonmatching_universal_quantifiers(
        self, state: SolutionState
    ) -> SolutionState:
        conjuncts = [conjunct for conjunct in get_conjuncts(state.constraint)]
        deleted = False

        for idx, universal_formula in reversed(list(enumerate(conjuncts))):
            if not isinstance(universal_formula, language.ForallFormula):
                continue

            if (
                universal_formula.in_variable.is_complete()
                and not matches_for_quantified_formula(universal_formula, self.grammar)
            ):
                deleted = True
                del conjuncts[idx]

        if not deleted:
            return state

        return SolutionState(sc.conjunction(*conjuncts), state.tree)

    def remove_infeasible_universal_quantifiers(
        self, state: SolutionState
    ) -> SolutionState:
        conjuncts = get_conjuncts(state.constraint)
        one_removed = False

        for idx, universal_formula in reversed(list(enumerate(conjuncts))):
            if not isinstance(universal_formula, language.ForallFormula):
                continue

            matches = matches_for_quantified_formula(universal_formula, self.grammar)

            all_matches_matched = all(
                universal_formula.is_already_matched(
                    match[universal_formula.bound_variable][1]
                )
                for match in matches
            )

            def some_leaf_might_match() -> bool:
                return any(
                    self.quantified_formula_might_match(
                        universal_formula, leaf_path, universal_formula.in_variable
                    )
                    for leaf_path, _ in universal_formula.in_variable.open_leaves()
                )

            if all_matches_matched and not some_leaf_might_match():
                one_removed = True
                del conjuncts[idx]

        return (
            state
            if not one_removed
            else SolutionState(
                reduce(lambda a, b: a & b, conjuncts, sc.true()),
                state.tree,
            )
        )

    def quantified_formula_might_match(
        self,
        qfd_formula: language.QuantifiedFormula,
        path_to_nonterminal: Path,
        tree: DerivationTree,
    ) -> bool:
        return quantified_formula_might_match(
            qfd_formula,
            path_to_nonterminal,
            tree,
            self.grammar,
            self.graph.reachable,
        )

    def extract_regular_expression(self, nonterminal: str) -> z3.ReRef:
        if nonterminal in self.regex_cache:
            return self.regex_cache[nonterminal]

        # For definitions like `<a> ::= <b>`, we only compute the regular expression
        # for `<b>`. That way, we might save some calls if `<b>` is used multiple times
        # (e.g., as in `<byte>`).
        canonical_expansions = self.canonical_grammar[nonterminal]

        if (
            len(canonical_expansions) == 1
            and len(canonical_expansions[0]) == 1
            and is_nonterminal(canonical_expansions[0][0])
        ):
            sub_nonterminal = canonical_expansions[0][0]
            assert (
                nonterminal != sub_nonterminal
            ), f"Expansion {nonterminal} => {sub_nonterminal}: Infinite recursion!"
            return self.regex_cache.setdefault(
                nonterminal, self.extract_regular_expression(sub_nonterminal)
            )

        # Similarly, for definitions like `<a> ::= <b> " x " <c>`, where `<b>` and `<c>`
        # don't reach `<a>`, we only compute the regular expressions for `<b>` and `<c>`
        # and return a concatenation. This also saves us expensive conversions (e.g.,
        # for `<seq> ::= <byte> <byte>`).
        if (
            len(canonical_expansions) == 1
            and any(is_nonterminal(elem) for elem in canonical_expansions[0])
            and all(
                not is_nonterminal(elem)
                or elem != nonterminal
                and not self.graph.reachable(elem, nonterminal)
                for elem in canonical_expansions[0]
            )
        ):
            result_elements: List[z3.ReRef] = [
                (
                    z3.Re(elem)
                    if not is_nonterminal(elem)
                    else self.extract_regular_expression(elem)
                )
                for elem in canonical_expansions[0]
            ]
            return self.regex_cache.setdefault(nonterminal, z3.Concat(*result_elements))

        regex_conv = RegexConverter(
            self.grammar,
            compress_unions=True,
            max_num_expansions=self.grammar_unwinding_threshold,
        )
        regex = regex_conv.to_regex(nonterminal, convert_to_z3=False)
        self.logger.debug(
            f"Computed regular expression for nonterminal {nonterminal}:\n{regex}"
        )
        z3_regex = regex_to_z3(regex)

        if assertions_activated():
            # Check correctness of regular expression
            grammar = self.graph.subgraph(nonterminal).to_grammar()

            # L(regex) \subseteq L(grammar)
            self.logger.debug(
                "Checking L(regex) \\subseteq L(grammar) for "
                + "nonterminal '%s' and regex '%s'",
                nonterminal,
                regex,
            )
            parser = EarleyParser(grammar)
            c = z3.String("c")
            prev: Set[str] = set()
            for _ in range(100):
                s = z3.Solver()
                s.add(z3.InRe(c, z3_regex))
                for inp in prev:
                    s.add(z3.Not(c == z3.StringVal(inp)))
                if s.check() != z3.sat:
                    self.logger.debug(
                        "Cannot find the %d-th solution for regex %s (timeout).\nThis is *not* a problem "
                        "if there not that many solutions (for regexes with finite language), or if we "
                        "are facing a meaningless timeout of the solver.",
                        len(prev) + 1,
                        regex,
                    )
                    break
                new_inp = smt_string_val_to_string(s.model()[c])
                try:
                    next(parser.parse(new_inp))
                except SyntaxError:
                    assert (
                        False
                    ), f"Input '{new_inp}' from regex language is not in grammar language."
                prev.add(new_inp)

        self.regex_cache[nonterminal] = z3_regex

        return z3_regex


class CostComputer(ABC):
    def compute_cost(self, state: SolutionState) -> float:
        """
        Computes a cost value for the given state. States with lower cost
        will be preferred in the analysis.

        :param state: The state for which to compute a cost.
        :return: The cost value.
        """
        raise NotImplementedError()

    def signal_tree_output(self, tree: DerivationTree) -> None:
        """
        Should be called when a tree is output as a solution. Used to
        update internal information for cost computation.

        :param tree The tree that is output as a solution.
        :return: Nothing.
        """
        raise NotImplementedError()


class GrammarBasedBlackboxCostComputer(CostComputer):
    def __init__(
        self,
        cost_settings: CostSettings,
        graph: gg.GrammarGraph,
        reset_coverage_after_n_round_with_no_coverage: int = 100,
        symbol_costs: Optional[Dict[str, int]] = None,
    ):
        self.cost_settings = cost_settings
        self.graph = graph

        self.covered_k_paths: Set[Tuple[gg.Node, ...]] = set()
        self.rounds_with_no_new_coverage = 0
        self.reset_coverage_after_n_round_with_no_coverage = (
            reset_coverage_after_n_round_with_no_coverage
        )
        self.symbol_costs: Optional[Dict[str, int]] = symbol_costs

        self.logger = logging.getLogger(type(self).__name__)

    def __repr__(self):
        return (
            "GrammarBasedBlackboxCostComputer("
            + f"{repr(self.cost_settings)}, "
            + "graph, "
            + f"{self.reset_coverage_after_n_round_with_no_coverage}, "
            + f"{self.symbol_costs})"
        )

    def compute_cost(self, state: SolutionState) -> float:
        # How costly is it to finish the tree?
        tree_closing_cost = self.compute_tree_closing_cost(state.tree)

        # Quantifiers are expensive (universal formulas have to be matched, tree insertion for existential
        # formulas is even more costly). TODO: Penalize nested quantifiers more.
        constraint_cost = sum(
            [
                idx * (2 if isinstance(f, language.ExistsFormula) else 1) + 1
                for c in get_quantifier_chains(state.constraint)
                for idx, f in enumerate(c)
            ]
        )

        # k-Path coverage: Fewer covered -> higher penalty
        k_cov_cost = self._compute_k_coverage_cost(state)

        # Covered k-paths: Fewer contributed -> higher penalty
        global_k_path_cost = self._compute_global_k_coverage_cost(state)

        costs = [
            tree_closing_cost,
            constraint_cost,
            state.level,
            k_cov_cost,
            global_k_path_cost,
        ]
        assert tree_closing_cost >= 0, f"tree_closing_cost == {tree_closing_cost}!"
        assert constraint_cost >= 0, f"constraint_cost == {constraint_cost}!"
        assert state.level >= 0, f"state.level == {state.level}!"
        assert k_cov_cost >= 0, f"k_cov_cost == {k_cov_cost}!"
        assert global_k_path_cost >= 0, f"global_k_path_cost == {global_k_path_cost}!"

        # Compute geometric mean
        result = weighted_geometric_mean(costs, list(self.cost_settings.weight_vector))

        self.logger.debug(
            "Computed cost for state %s:\n%f, individual costs: %s, weights: %s",
            lazystr(lambda: f"({(str(state.constraint)[:50] + '...')}, {state.tree})"),
            result,
            costs,
            self.cost_settings.weight_vector,
        )

        return result

    def signal_tree_output(self, tree: DerivationTree) -> None:
        self._update_covered_k_paths(tree)

    def _symbol_costs(self):
        if self.symbol_costs is None:
            self.symbol_costs = compute_symbol_costs(self.graph)
        return self.symbol_costs

    def _update_covered_k_paths(self, tree: DerivationTree):
        if self.cost_settings.weight_vector.low_global_k_path_coverage_penalty > 0:
            old_covered_k_paths = copy.copy(self.covered_k_paths)

            self.covered_k_paths.update(
                tree.k_paths(
                    self.graph, self.cost_settings.k, include_potential_paths=False
                )
            )

            if old_covered_k_paths == self.covered_k_paths:
                self.rounds_with_no_new_coverage += 1

            graph_paths = self.graph.k_paths(
                self.cost_settings.k, include_terminals=False
            )
            if (
                self.rounds_with_no_new_coverage
                >= self.reset_coverage_after_n_round_with_no_coverage
                or self.covered_k_paths == graph_paths
            ):
                if self.covered_k_paths == graph_paths:
                    self.logger.debug("ALL PATHS COVERED")
                else:
                    self.logger.debug(
                        "COVERAGE RESET SINCE NO CHANGE IN COVERED PATHS SINCE %d "
                        + "ROUNDS (%d path(s) uncovered)",
                        self.reset_coverage_after_n_round_with_no_coverage,
                        len(graph_paths) - len(self.covered_k_paths),
                    )

                    # uncovered_paths = (
                    #     self.graph.k_paths(
                    #         self.cost_settings.k, include_terminals=False
                    #     )
                    #     - self.covered_k_paths
                    # )
                    # self.logger.debug(
                    #     "\n".join(
                    #         [
                    #             ", ".join(f"'{n.symbol}'" for n in p)
                    #             for p in uncovered_paths
                    #         ]
                    #     )
                    # )

                self.covered_k_paths = set()
            else:
                pass
                # uncovered_paths = (
                #     self.graph.k_paths(self.cost_settings.k, include_terminals=False)
                #     - self.covered_k_paths
                # )
                # self.logger.debug("%d uncovered paths", len(uncovered_paths))
                # self.logger.debug(
                #     "\n"
                #     + "\n".join(
                #         [", ".join(f"'{n.symbol}'" for n in p)
                #         for p in uncovered_paths]
                #     )
                #     + "\n"
                # )

            if (
                self.rounds_with_no_new_coverage
                >= self.reset_coverage_after_n_round_with_no_coverage
            ):
                self.rounds_with_no_new_coverage = 0

    def _compute_global_k_coverage_cost(self, state: SolutionState):
        if self.cost_settings.weight_vector.low_global_k_path_coverage_penalty == 0:
            return 0

        tree_k_paths = state.tree.k_paths(
            self.graph, self.cost_settings.k, include_potential_paths=False
        )
        all_graph_k_paths = self.graph.k_paths(
            self.cost_settings.k, include_terminals=False
        )

        contributed_k_paths = {
            path
            for path in all_graph_k_paths
            if path in tree_k_paths and path not in self.covered_k_paths
        }

        num_contributed_k_paths = len(contributed_k_paths)
        num_missing_k_paths = len(all_graph_k_paths) - len(self.covered_k_paths)

        # self.logger.debug(
        #     'k-Paths contributed by input %s:\n%s',
        #     state.tree,
        #     '\n'.join(map(
        #         lambda path: ' '.join(map(
        #             lambda n: n.symbol,
        #             filter(lambda n: not isinstance(n, gg.ChoiceNode), path))),
        #         contributed_k_paths)))
        # self.logger.debug('Missing k paths: %s', num_missing_k_paths)

        assert 0 <= num_contributed_k_paths <= num_missing_k_paths, (
            f"num_contributed_k_paths == {num_contributed_k_paths}, "
            f"num_missing_k_paths == {num_missing_k_paths}"
        )

        # return 1 - (num_contributed_k_paths / num_missing_k_paths)

        potential_tree_k_paths = state.tree.k_paths(
            self.graph, self.cost_settings.k, include_potential_paths=True
        )
        contributed_k_paths = {
            path
            for path in all_graph_k_paths
            if path in potential_tree_k_paths and path not in self.covered_k_paths
        }

        num_contributed_potential_k_paths = len(contributed_k_paths)

        if not num_missing_k_paths:
            return 0

        return 1 - weighted_geometric_mean(
            [
                num_contributed_k_paths / num_missing_k_paths,
                num_contributed_potential_k_paths / num_missing_k_paths,
            ],
            [0.2, 0.8],
        )

    def _compute_k_coverage_cost(self, state: SolutionState) -> float:
        if self.cost_settings.weight_vector.low_k_coverage_penalty == 0:
            return 0

        coverages = []
        for k in range(1, self.cost_settings.k + 1):
            coverage = state.tree.k_coverage(
                self.graph, k, include_potential_paths=False
            )
            assert 0 <= coverage <= 1, f"coverage == {coverage}"

            coverages.append(1 - coverage)

        return math.prod(coverages) ** (1 / float(self.cost_settings.k))

    def compute_tree_closing_cost(self, tree: DerivationTree) -> float:
        nonterminals = [leaf.value for _, leaf in tree.open_leaves()]
        return sum([self._symbol_costs()[nonterminal] for nonterminal in nonterminals])


def smt_formulas_referring_to_subtrees(
    smt_formulas: Sequence[language.SMTFormula],
) -> List[language.SMTFormula]:
    """
    Returns a list of SMT formulas whose solutions address subtrees of other SMT
    formulas, but whose own substitution subtrees are in turn *not* referred by
    top-level substitution trees of other formulas. Those must be solved first to avoid
    inconsistencies.

    :param smt_formulas: The formulas to search for references to subtrees.
    :return: The list of conflicting formulas that must be solved first.
    """

    def subtree_ids(formula: language.SMTFormula) -> Set[int]:
        return {
            subtree.id
            for tree in formula.substitutions.values()
            for _, subtree in tree.paths()
            if subtree.id != tree.id
        }

    def tree_ids(formula: language.SMTFormula) -> Set[int]:
        return {tree.id for tree in formula.substitutions.values()}

    subtree_ids_for_formula: Dict[language.SMTFormula, Set[int]] = {
        formula: subtree_ids(formula) for formula in smt_formulas
    }

    tree_ids_for_formula: Dict[language.SMTFormula, Set[int]] = {
        formula: tree_ids(formula) for formula in smt_formulas
    }

    def independent_from_solutions_of_other_formula(
        idx: int, formula: language.SMTFormula
    ) -> bool:
        return all(
            not tree_ids_for_formula[other_formula].intersection(
                subtree_ids_for_formula[formula]
            )
            for other_idx, other_formula in enumerate(smt_formulas)
            if other_idx != idx
        )

    def refers_to_subtree_of_other_formula(
        idx: int, formula: language.SMTFormula
    ) -> bool:
        return any(
            tree_ids_for_formula[formula].intersection(
                subtree_ids_for_formula[other_formula]
            )
            for other_idx, other_formula in enumerate(smt_formulas)
            if other_idx != idx
        )

    return [
        formula
        for idx, formula in enumerate(smt_formulas)
        if refers_to_subtree_of_other_formula(idx, formula)
        and independent_from_solutions_of_other_formula(idx, formula)
    ]


def compute_tree_closing_cost(tree: DerivationTree, graph: GrammarGraph) -> float:
    nonterminals = [leaf.value for _, leaf in tree.open_leaves()]
    return sum(
        [compute_symbol_costs(graph)[nonterminal] for nonterminal in nonterminals]
    )


def get_quantifier_chains(
    formula: language.Formula,
) -> List[Tuple[Union[language.QuantifiedFormula, language.ExistsIntFormula], ...]]:
    univ_toplevel_formulas = get_toplevel_quantified_formulas(formula)
    return [
        (f,) + c
        for f in univ_toplevel_formulas
        for c in (get_quantifier_chains(f.inner_formula) or [()])
    ]


def shortest_derivations(graph: gg.GrammarGraph) -> Dict[str, int]:
    def avg(it) -> int:
        elems = [elem for elem in it if elem is not None]
        return math.ceil(math.prod(elems) ** (1 / len(elems)))

    parent_relation = {node: set() for node in graph.all_nodes}
    for parent, child in graph.all_edges:
        parent_relation[child].add(parent)

    shortest_node_derivations: Dict[gg.Node, int] = {}
    stack: List[gg.Node] = graph.filter(lambda node: isinstance(node, gg.TerminalNode))
    while stack:
        node = stack.pop()

        old_min = shortest_node_derivations.get(node, None)

        if isinstance(node, gg.TerminalNode):
            shortest_node_derivations[node] = 0
        elif isinstance(node, gg.ChoiceNode):
            shortest_node_derivations[node] = max(
                shortest_node_derivations.get(child, 0) for child in node.children
            )
        elif isinstance(node, gg.NonterminalNode):
            assert not isinstance(node, gg.ChoiceNode)

            shortest_node_derivations[node] = (
                avg(
                    shortest_node_derivations.get(child, None)
                    for child in node.children
                )
                + 1
            )

        if (old_min or sys.maxsize) > shortest_node_derivations[node]:
            stack.extend(parent_relation[node])

    return {
        nonterminal: shortest_node_derivations[graph.get_node(nonterminal)]
        for nonterminal in graph.grammar
    }


@lru_cache()
def compute_symbol_costs(graph: GrammarGraph) -> Dict[str, int]:
    grammar = graph.to_grammar()
    canonical_grammar = canonical(grammar)

    result: Dict[str, int] = shortest_derivations(graph)

    nonterminal_parents = [
        nonterminal
        for nonterminal in canonical_grammar
        if any(
            is_nonterminal(symbol)
            for expansion in canonical_grammar[nonterminal]
            for symbol in expansion
        )
    ]

    # Sometimes this computation results in some nonterminals having lower cost values
    # than nonterminals that are reachable from those (but not vice versa), which is
    # undesired. We counteract this by assuring that on paths with at most one cycle
    # from the root to any nonterminal parent, the costs are strictly monotonically
    # decreasing.
    for nonterminal_parent in nonterminal_parents:
        # noinspection PyTypeChecker
        for path in all_paths(graph, graph.root, graph.get_node(nonterminal_parent)):
            for idx in reversed(range(1, len(path))):
                source: gg.Node = path[idx - 1]
                target: gg.Node = path[idx]

                if result[source.symbol] <= result[target.symbol]:
                    result[source.symbol] = result[target.symbol] + 1

    return result


def all_paths(
    graph,
    from_node: gg.NonterminalNode,
    to_node: gg.NonterminalNode,
    cycles_allowed: int = 2,
) -> List[List[gg.NonterminalNode]]:
    """Compute all paths between two nodes. Note: We allow to visit each nonterminal twice.
    This is not really allowing up to `cycles_allowed` cycles (which was the original intention
    of the parameter), since then we would have to check per path; yet, the number of paths would
    explode then and the current implementation provides reasonably good results."""
    result: List[List[gg.NonterminalNode]] = []
    visited: Dict[gg.NonterminalNode, int] = {n: 0 for n in graph.all_nodes}

    queue: List[List[gg.NonterminalNode]] = [[from_node]]
    while queue:
        p = queue.pop(0)
        if p[-1] == to_node:
            result.append(p)
            continue

        for child in p[-1].children:
            if (
                not isinstance(child, gg.NonterminalNode)
                or visited[child] > cycles_allowed + 1
            ):
                continue

            visited[child] += 1
            queue.append(p + [child])

    return [[n for n in p if not isinstance(n, gg.ChoiceNode)] for p in result]


def implies(
    f1: language.Formula, f2: language.Formula, grammar: Grammar, timeout_seconds=5
) -> bool:
    solver = ISLaSolver(
        grammar, f1 & -f2, activate_unsat_support=True, timeout_seconds=timeout_seconds
    )

    return (
        safe(solver.solve, exceptions=(StopIteration,))()
        .map(lambda _: False)
        .lash(lambda _: Success(True))
    ).unwrap()


def equivalent(
    f1: language.Formula, f2: language.Formula, grammar: Grammar, timeout_seconds=5
) -> bool:
    solver = ISLaSolver(
        grammar,
        -(f1 & f2 | -f1 & -f2),
        activate_unsat_support=True,
        timeout_seconds=timeout_seconds,
    )

    return (
        safe(solver.solve)()
        .map(lambda _: False)
        .lash(lambda e: Success(isinstance(e, StopIteration)))
    ).unwrap()


def generate_abstracted_trees(
    inp: DerivationTree, participating_paths: Set[Path]
) -> List[DerivationTree]:
    """
    Yields trees that are more and more "abstracted," i.e., pruned, at prefixes of the
    paths specified in `participating_paths`.

    :param inp: The unabstracted input.
    :param participating_paths: The paths to abstract.
    :return: A generator of more and more abstract trees, beginning with the most
    concrete and ending with the most abstract ones.
    """
    parent_paths: Set[ImmutableList[Path]] = {
        tuple(
            [tuple(path[:i]) for i in reversed(range(1, len(path) + 1))]
            if path
            else [()]
        )
        for path in participating_paths
    }

    abstraction_candidate_combinations: Set[ImmutableList[Path]] = {
        tuple(eliminate_suffixes(combination))
        for k in range(1, len(participating_paths) + 1)
        for paths in itertools.product(*parent_paths)
        for combination in itertools.combinations(paths, k)
    }

    result: Dict[int, DerivationTree] = {}
    for paths_to_abstract in abstraction_candidate_combinations:
        abstracted_tree = inp.substitute(
            {
                inp.get_subtree(path_to_abstract): DerivationTree(
                    inp.get_subtree(path_to_abstract).value
                )
                for path_to_abstract in paths_to_abstract
            }
        )
        result[abstracted_tree.structural_hash()] = abstracted_tree

    return sorted(result.values(), key=lambda tree: -len(tree))


class EvaluatePredicateFormulasTransformer(NoopFormulaTransformer):
    def __init__(self, inp: DerivationTree):
        super().__init__()
        self.inp = inp

    def transform_predicate_formula(
        self, sub_formula: language.StructuralPredicateFormula
    ) -> language.Formula:
        return sc.true() if sub_formula.evaluate(self.inp) else sc.false()

    def transform_conjunctive_formula(
        self, sub_formula: language.ConjunctiveFormula
    ) -> language.Formula:
        return reduce(language.Formula.__and__, sub_formula.args)

    def transform_disjunctive_formula(
        self, sub_formula: language.DisjunctiveFormula
    ) -> language.Formula:
        return reduce(language.Formula.__or__, sub_formula.args)

    def transform_smt_formula(
        self, sub_formula: language.SMTFormula
    ) -> language.Formula:
        # We instantiate the formula and check whether it evaluates to
        # True (or False in a negation scope); in that case, we replace
        # it by "true." Otherwise, we keep it for later analysis.

        instantiated_formula = copy.deepcopy(sub_formula)
        set_smt_auto_subst(instantiated_formula, True)
        set_smt_auto_eval(instantiated_formula, True)
        instantiated_formula = instantiated_formula.substitute_expressions(
            sub_formula.substitutions, force=True
        )

        assert instantiated_formula in {sc.true(), sc.false()}

        return (
            sc.true()
            if (instantiated_formula == sc.true()) ^ self.in_negation_scope
            else sub_formula
        )


def create_fixed_length_tree(
    start: DerivationTree | str,
    canonical_grammar: CanonicalGrammar,
    target_length: int,
) -> Optional[DerivationTree]:
    nullable = compute_nullable_nonterminals(canonical_grammar)
    start = DerivationTree(start) if isinstance(start, str) else start
    stack: List[
        Tuple[DerivationTree, int, ImmutableList[Tuple[Path, DerivationTree]]]
    ] = [
        (start, int(start.value not in nullable), (((), start),)),
    ]

    while stack:
        tree, curr_len, open_leaves = stack.pop()

        if not open_leaves:
            if curr_len == target_length:
                return tree
            else:
                continue

        if curr_len > target_length:
            continue

        idx: int
        path: Path
        leaf: DerivationTree
        for idx, (path, leaf) in reversed(list(enumerate(open_leaves))):
            terminal_expansions, expansions = get_expansions(
                leaf.value, canonical_grammar
            )

            if terminal_expansions:
                expansions.append(random.choice(terminal_expansions))

            # Only choose one random terminal expansion; keep all nonterminal expansions
            expansions = sorted(
                expansions,
                key=lambda expansion: len(
                    [elem for elem in expansion if is_nonterminal(elem)]
                ),
            )

            for expansion in reversed(expansions):
                new_children = tuple(
                    [
                        DerivationTree(elem, None if is_nonterminal(elem) else ())
                        for elem in expansion
                    ]
                )

                expanded_tree = tree.replace_path(
                    path,
                    DerivationTree(
                        leaf.value,
                        new_children,
                    ),
                )

                stack.append(
                    (
                        expanded_tree,
                        curr_len
                        + sum(
                            [
                                (
                                    len(child.value)
                                    if child.children == ()
                                    else (1 if child.value not in nullable else 0)
                                )
                                for child in new_children
                            ]
                        )
                        - int(leaf.value not in nullable),
                        open_leaves[:idx]
                        + tuple(
                            [
                                (path + (child_idx,), new_child)
                                for child_idx, new_child in enumerate(new_children)
                                if is_nonterminal(new_child.value)
                            ]
                        )
                        + open_leaves[idx + 1 :],
                    )
                )

    return None


def subtree_solutions(
    solution: Dict[language.Constant | DerivationTree, DerivationTree]
) -> Dict[language.Variable | DerivationTree, DerivationTree]:
    solution_with_subtrees: Dict[language.Variable | DerivationTree, DerivationTree] = (
        {}
    )
    for orig, subst in solution.items():
        if isinstance(orig, language.Variable):
            solution_with_subtrees[orig] = subst
            continue

        assert isinstance(
            orig, DerivationTree
        ), f"Expected a DerivationTree, given: {type(orig).__name__}"

        # Note: It can happen that a path in the original tree is not valid in the
        #       substitution, e.g., if we happen to replace a larger with a smaller
        #       tree.
        for path, tree in [
            (p, t)
            for p, t in orig.paths()
            if t not in solution_with_subtrees and subst.is_valid_path(p)
        ]:
            solution_with_subtrees[tree] = subst.get_subtree(path)

    return solution_with_subtrees

rindPHI / isla / 7846461830

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous