Self-contained proof steps for EqualityProver and APRProver

src/pyk/proof/equality.py

@@ -12,7 +12,7 @@
 from ..prelude.kbool import BOOL, TRUE
 from ..prelude.ml import is_bottom, is_top, mlAnd, mlEquals, mlEqualsFalse
 from ..utils import ensure_dir_path
-from .proof import Proof, ProofStatus, ProofSummary, Prover
+from .proof import Proof, ProofStatus, ProofStep, ProofSummary, Prover, StepResult
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Mapping
@@ -349,46 +349,81 @@ def lines(self) -> list[str]:
         ]
 
 
+@dataclass
+class ImpliesProofResult(StepResult):
+    csubst: CSubst | None
+    simplified_antecedent: KInner | None
+    simplified_consequent: KInner | None
+
+
+@dataclass
+class ImpliesProofStep(ProofStep):
+    kcfg_explore: KCFGExplore
+    antecedent: KInner
+    consequent: KInner
+    proof_id: str
+
+    def exec(self) -> StepResult:
+        # to prove the equality, we check the implication of the form `constraints #Implies LHS #Equals RHS`, i.e.
+        # "LHS equals RHS under these constraints"
+        simplified_antecedent, _ = self.kcfg_explore.kast_simplify(self.antecedent)
+        simplified_consequent, _ = self.kcfg_explore.kast_simplify(self.consequent)
+        _LOGGER.info(f'Simplified antecedent: {self.kcfg_explore.kprint.pretty_print(simplified_antecedent)}')
+        _LOGGER.info(f'Simplified consequent: {self.kcfg_explore.kprint.pretty_print(simplified_consequent)}')
+
+        csubst: CSubst | None = None
+
+        if is_bottom(simplified_antecedent):
+            _LOGGER.warning(f'Antecedent of implication (proof constraints) simplifies to #Bottom {self.proof_id}')
+            csubst = CSubst(Subst({}), ())
+
+        elif is_top(simplified_consequent):
+            _LOGGER.warning(f'Consequent of implication (proof equality) simplifies to #Top {self.proof_id}')
+            csubst = CSubst(Subst({}), ())
+
+        else:
+            dummy_config = self.kcfg_explore.kprint.definition.empty_config(sort=GENERATED_TOP_CELL)
+            implies_result = self.kcfg_explore.cterm_implies(
+                antecedent=CTerm(config=dummy_config, constraints=[simplified_antecedent]),
+                consequent=CTerm(config=dummy_config, constraints=[simplified_consequent]),
+            )
+            if implies_result is not None:
+                csubst = implies_result
+
+        return ImpliesProofResult(
+            csubst=csubst, simplified_antecedent=simplified_antecedent, simplified_consequent=simplified_consequent
+        )
+
+
 class ImpliesProver(Prover):
     proof: ImpliesProof
 
     def __init__(self, proof: ImpliesProof, kcfg_explore: KCFGExplore):
         super().__init__(kcfg_explore)
         self.proof = proof
 
-    def step_proof(self) -> None:
+    def get_steps(self) -> Iterable[ProofStep]:
         proof_type = type(self.proof).__name__
         _LOGGER.info(f'Attempting {proof_type} {self.proof.id}')
 
         if self.proof.status is not ProofStatus.PENDING:
             _LOGGER.info(f'{proof_type} finished {self.proof.id}: {self.proof.status}')
-            return
-
-        # to prove the equality, we check the implication of the form `constraints #Implies LHS #Equals RHS`, i.e.
-        # "LHS equals RHS under these constraints"
-        antecedent_simplified_kast, _ = self.kcfg_explore.kast_simplify(self.proof.antecedent)
-        consequent_simplified_kast, _ = self.kcfg_explore.kast_simplify(self.proof.consequent)
-        self.proof.simplified_antecedent = antecedent_simplified_kast
-        self.proof.simplified_consequent = consequent_simplified_kast
-        _LOGGER.info(f'Simplified antecedent: {self.kcfg_explore.kprint.pretty_print(antecedent_simplified_kast)}')
-        _LOGGER.info(f'Simplified consequent: {self.kcfg_explore.kprint.pretty_print(consequent_simplified_kast)}')
-
-        if is_bottom(antecedent_simplified_kast):
-            _LOGGER.warning(f'Antecedent of implication (proof constraints) simplifies to #Bottom {self.proof.id}')
-            self.proof.csubst = CSubst(Subst({}), ())
-
-        elif is_top(consequent_simplified_kast):
-            _LOGGER.warning(f'Consequent of implication (proof equality) simplifies to #Top {self.proof.id}')
-            self.proof.csubst = CSubst(Subst({}), ())
-
-        else:
-            # TODO: we should not be forced to include the dummy configuration in the antecedent and consequent
-            dummy_config = self.kcfg_explore.kprint.definition.empty_config(sort=GENERATED_TOP_CELL)
-            result = self.kcfg_explore.cterm_implies(
-                antecedent=CTerm(config=dummy_config, constraints=[self.proof.simplified_antecedent]),
-                consequent=CTerm(config=dummy_config, constraints=[self.proof.simplified_consequent]),
+            return []
+        return [
+            ImpliesProofStep(
+                antecedent=self.proof.antecedent,
+                consequent=self.proof.consequent,
+                kcfg_explore=self.kcfg_explore,
+                proof_id=self.proof.id,
             )
-            if result is not None:
-                self.proof.csubst = result
+        ]
 
-        _LOGGER.info(f'{proof_type} finished {self.proof.id}: {self.proof.status}')
+    def commit(self, result: StepResult) -> None:
+        proof_type = type(self.proof).__name__
+        if isinstance(result, ImpliesProofResult):
+            self.proof.csubst = result.csubst
+            self.proof.simplified_antecedent = result.simplified_antecedent
+            self.proof.simplified_consequent = result.simplified_consequent
+            _LOGGER.info(f'{proof_type} finished {self.proof.id}: {self.proof.status}')
+        else:
+            raise ValueError('Incorrect result type')

src/pyk/proof/proof.py

@@ -286,6 +286,16 @@ def lines(self) -> list[str]:
         return [line for lines in (summary.lines for summary in self.summaries) for line in lines]
 
 
+class StepResult:
+    ...
+
+
+class ProofStep:
+    @abstractmethod
+    def exec(self) -> StepResult:
+        ...
+
+
 class Prover:
     kcfg_explore: KCFGExplore
     proof: Proof
@@ -294,17 +304,24 @@ def __init__(self, kcfg_explore: KCFGExplore):
         self.kcfg_explore = kcfg_explore
 
     @abstractmethod
-    def step_proof(self) -> None:
+    def get_steps(self) -> Iterable[ProofStep]:
+        ...
+
+    @abstractmethod
+    def commit(self, result: StepResult) -> None:
         ...
 
     def advance_proof(self, max_iterations: int | None = None, fail_fast: bool = False) -> None:
         iterations = 0
         while self.proof.can_progress:
-            if fail_fast and self.proof.failed:
-                _LOGGER.warning(f'Terminating proof early because fail_fast is set: {self.proof.id}')
-                return
-            if max_iterations is not None and max_iterations <= iterations:
-                return
-            iterations += 1
-            self.step_proof()
+            steps = self.get_steps()
+            for step in steps:
+                iterations += 1
+                result = step.exec()
+                self.commit(result)
+                if fail_fast and self.proof.failed:
+                    _LOGGER.warning(f'Terminating proof early because fail_fast is set: {self.proof.id}')
+                    return
+                if max_iterations is not None and max_iterations <= iterations:
+                    return
             self.proof.write_proof_data()

src/pyk/proof/reachability.py

@@ -19,15 +19,17 @@
 from ..prelude.ml import mlAnd, mlEquals, mlTop
 from ..utils import FrozenDict, ensure_dir_path, hash_str, shorten_hashes, single
 from .equality import ProofSummary, Prover, RefutationProof
-from .proof import CompositeSummary, Proof, ProofStatus
+from .proof import CompositeSummary, Proof, ProofStatus, ProofStep, StepResult
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Mapping
     from pathlib import Path
     from typing import Any, Final, TypeVar
 
+    from ..cterm import CSubst, CTerm
     from ..kast.outer import KClaim, KDefinition, KFlatModuleList, KRuleLike
     from ..kcfg import KCFGExplore
+    from ..kcfg.explore import ExtendResult
     from ..kcfg.kcfg import NodeIdLike
 
     T = TypeVar('T', bound='Proof')
@@ -685,19 +687,76 @@ def advance_pending_node(
             module_name=module_name,
         )
 
-    def step_proof(self) -> None:
-        self._check_all_terminals()
-
+    def get_steps(self) -> Iterable[APRProofStep]:
         if not self.proof.pending:
-            return
-        curr_node = self.proof.pending[0]
+            return []
+        steps = []
+        target_node = self.proof.kcfg.node(self.proof.target)
+        for curr_node in self.proof.pending:
+            if self.proof.bmc_depth is not None and curr_node.id not in self._checked_for_bounded:
+                _LOGGER.info(f'Checking bmc depth for node {self.proof.id}: {curr_node.id}')
+                self._checked_for_bounded.add(curr_node.id)
+                _prior_loops = [
+                    succ.source.id
+                    for succ in self.proof.shortest_path_to(curr_node.id)
+                    if self.kcfg_explore.kcfg_semantics.same_loop(succ.source.cterm, curr_node.cterm)
+                ]
+                prior_loops: list[NodeIdLike] = []
+                for _pl in _prior_loops:
+                    if not (
+                        self.proof.kcfg.zero_depth_between(_pl, curr_node.id)
+                        or any(self.proof.kcfg.zero_depth_between(_pl, pl) for pl in prior_loops)
+                    ):
+                        prior_loops.append(_pl)
+                _LOGGER.info(f'Prior loop heads for node {self.proof.id}: {(curr_node.id, prior_loops)}')
+                if len(prior_loops) > self.proof.bmc_depth:
+                    _LOGGER.warning(f'Bounded node {self.proof.id}: {curr_node.id} at bmc depth {self.proof.bmc_depth}')
+                    self.proof.add_bounded(curr_node.id)
+                    continue
+            steps.append(
+                APRProofStep(
+                    always_check_subsumption=self.always_check_subsumption,
+                    bmc_depth=self.proof.bmc_depth,
+                    checked_for_bounded=self._checked_for_bounded,
+                    cterm=curr_node.cterm,
+                    cut_point_rules=self.cut_point_rules,
+                    execute_depth=self.execute_depth,
+                    fast_check_subsumption=self.fast_check_subsumption,
+                    is_terminal=self.kcfg_explore.kcfg_semantics.is_terminal(curr_node.cterm),
+                    kcfg_explore=self.kcfg_explore,
+                    module_name=(
+                        self.circularities_module_name
+                        if self.nonzero_depth(curr_node)
+                        else self.dependencies_module_name
+                    ),
+                    node_id=curr_node.id,
+                    proof_id=self.proof.id,
+                    target_cterm=target_node.cterm,
+                    target_is_terminal=self.proof.target in self.proof._terminal,
+                    target_node_id=target_node.id,
+                    terminal_rules=self.terminal_rules,
+                )
+            )
+        return steps
 
-        self.advance_pending_node(
-            node=curr_node,
-            execute_depth=self.execute_depth,
-            cut_point_rules=self.cut_point_rules,
-            terminal_rules=self.terminal_rules,
-        )
+    def commit(self, result: StepResult) -> None:
+        if isinstance(result, APRProofExtendResult):
+            self.kcfg_explore.extend_kcfg(
+                result.extend_result, self.proof.kcfg, self.proof.kcfg.node(result.node_id), self.proof.logs
+            )
+        elif isinstance(result, APRProofSubsumeResult):
+            if result.csubst is None:
+                self.proof._terminal.add(result.node_id)
+            else:
+                self.proof.kcfg.create_cover(result.node_id, self.proof.target, csubst=result.csubst)
+                _LOGGER.info(
+                    f'Subsumed into target node {self.proof.id}: {shorten_hashes((result.node_id, self.proof.target))}'
+                )
+        else:
+            raise ValueError('Incorrect result type')
+
+        if self.proof.failed:
+            self.proof.failure_info = self.failure_info()
 
         self._check_all_terminals()
 
@@ -710,13 +769,79 @@ def step_proof(self) -> None:
                 self._checked_for_subsumption.add(node.id)
                 self._check_subsume(node)
 
-        if self.proof.failed:
-            self.proof.failure_info = self.failure_info()
-
     def failure_info(self) -> APRFailureInfo:
         return APRFailureInfo.from_proof(self.proof, self.kcfg_explore, counterexample_info=self.counterexample_info)
 
 
+@dataclass
+class APRProofResult(StepResult):
+    node_id: int
+
+
+@dataclass
+class APRProofExtendResult(APRProofResult):
+    extend_result: ExtendResult
+
+
+@dataclass
+class APRProofSubsumeResult(APRProofResult):
+    csubst: CSubst | None
+
+
+@dataclass(frozen=True, eq=True)
+class APRProofStep(ProofStep):
+    bmc_depth: int | None
+    node_id: int
+    target_node_id: int
+    proof_id: str
+    fast_check_subsumption: bool
+    cterm: CTerm
+    target_cterm: CTerm
+    kcfg_explore: KCFGExplore
+    checked_for_bounded: Iterable[int]
+    target_is_terminal: bool
+    always_check_subsumption: bool
+    module_name: str
+    cut_point_rules: Iterable[str]
+    terminal_rules: Iterable[str]
+    execute_depth: int | None
+    is_terminal: bool
+
+    def _may_subsume(self) -> bool:
+        node_k_cell = self.cterm.try_cell('K_CELL')
+        target_k_cell = self.target_cterm.try_cell('K_CELL')
+        if node_k_cell and target_k_cell and not target_k_cell.match(node_k_cell):
+            return False
+        return True
+
+    def _check_subsume(self) -> CSubst | None:
+        _LOGGER.info(
+            f'Checking subsumption into target state {self.proof_id}: {shorten_hashes((self.node_id, self.target_node_id))}'
+        )
+        if self.fast_check_subsumption and not self._may_subsume():
+            _LOGGER.info(
+                f'Skipping full subsumption check because of fast may subsume check {self.proof_id}: {self.node_id}'
+            )
+            return None
+        return self.kcfg_explore.cterm_implies(self.cterm, self.target_cterm)
+
+    def exec(self) -> APRProofResult:
+        if self.is_terminal or not self.target_is_terminal:
+            if self.always_check_subsumption:
+                csubst = self._check_subsume()
+                if csubst is not None or self.is_terminal:
+                    return APRProofSubsumeResult(csubst=csubst, node_id=self.node_id)
+
+        extend_result = self.kcfg_explore.extend_cterm(
+            self.cterm,
+            execute_depth=self.execute_depth,
+            cut_point_rules=self.cut_point_rules,
+            terminal_rules=self.terminal_rules,
+            module_name=self.module_name,
+        )
+        return APRProofExtendResult(node_id=self.node_id, extend_result=extend_result)
+
+
 @dataclass(frozen=True)
 class APRSummary(ProofSummary):
     id: str