Reusing backend information to construct KCFGs more efficiently (#4590)

This PR leverages information provided by the backend as part of an
`execute` request to construct more than one KCFG element: specifically,
vacuous nodes, stuck nodes, edges resulting from cut-nodes, branches,
and non-deterministic branches.

In addition, it moves the associated printing to the KCFG class,
allowing for more informative prints: for example, we can now print the
target nodes of edges and branches, the absence of which (personally)
was making debugging of large proofs slightly more difficult.

The PR also removes the `always_check_subsumption` flag, which was not
used and was made obsolete by the subsumption logic in `step_proof`.

---------

Co-authored-by: Tamás Tóth <[email protected]>

pyk/src/pyk/kcfg/explore.py

@@ -218,19 +218,15 @@ def extend_cterm(
         cut_point_rules: Iterable[str] = (),
         terminal_rules: Iterable[str] = (),
         module_name: str | None = None,
-    ) -> KCFGExtendResult:
-        def log(message: str, *, warning: bool = False) -> None:
-            _LOGGER.log(logging.WARNING if warning else logging.INFO, f'Extend result for {self.id}: {message}')
+    ) -> list[KCFGExtendResult]:
 
         custom_step_result = self.kcfg_semantics.custom_step(_cterm)
         if custom_step_result is not None:
-            log(f'custom step node: {node_id}')
-            return custom_step_result
+            return [custom_step_result]
 
         abstract_cterm = self.kcfg_semantics.abstract_node(_cterm)
         if _cterm != abstract_cterm:
-            log(f'abstraction node: {node_id}')
-            return Abstract(abstract_cterm)
+            return [Abstract(abstract_cterm)]
 
         cterm, next_states, depth, vacuous, next_node_logs = self.cterm_symbolic.execute(
             _cterm,
@@ -240,33 +236,27 @@ def log(message: str, *, warning: bool = False) -> None:
             module_name=module_name,
         )
 
+        extend_results: list[KCFGExtendResult] = []
+
         # Basic block
         if depth > 0:
-            log(f'basic block at depth {depth}: {node_id}')
-            return Step(cterm, depth, next_node_logs, self._extract_rule_labels(next_node_logs))
+            extend_results.append(Step(cterm, depth, next_node_logs, self._extract_rule_labels(next_node_logs)))
 
         # Stuck or vacuous
         if not next_states:
             if vacuous:
-                log(f'vacuous node: {node_id}', warning=True)
-                return Vacuous()
-            log(f'stuck node: {node_id}')
-            return Stuck()
-
+                extend_results.append(Vacuous())
+            elif depth == 0:
+                extend_results.append(Stuck())
         # Cut rule
-        if len(next_states) == 1:
-            log(f'cut-rule basic block at depth {depth}: {node_id}')
-            return Step(
-                next_states[0].state,
-                1,
-                next_node_logs,
-                self._extract_rule_labels(next_node_logs),
-                cut=True,
-            )
-
+        elif len(next_states) == 1:
+            if not self.kcfg_semantics.can_make_custom_step(cterm):
+                (next_node_logs, rules) = (
+                    ((), []) if extend_results else (next_node_logs, self._extract_rule_labels(next_node_logs))
+                )
+                extend_results.append(Step(next_states[0].state, 1, next_node_logs, rules, cut=True, info='cut-rule'))
         # Branch
-        assert len(next_states) > 1
-        if all(branch_constraint for _, branch_constraint in next_states):
+        elif all(branch_constraint for _, branch_constraint in next_states):
             branch_preds = [flatten_label('#And', not_none(rule_predicate)) for _, rule_predicate in next_states]
             common_preds = list(
                 unique(
@@ -277,15 +267,14 @@ def log(message: str, *, warning: bool = False) -> None:
                 )
             )
             branches = [mlAnd(pred for pred in branch_pred if pred not in common_preds) for branch_pred in branch_preds]
-            if common_preds:
-                log(
-                    f'Common predicates found in branches: {[self.pretty_print(ml_pred_to_bool(cp)) for cp in common_preds]}'
-                )
-            constraint_strs = [self.pretty_print(ml_pred_to_bool(bc)) for bc in branches]
-            log(f'{len(branches)} branches: {node_id} -> {constraint_strs}')
-            return Branch(branches)
+            info = f'{[self.pretty_print(ml_pred_to_bool(bc)) for bc in branches]}'
+            extend_results.append(Branch(branches, info=info))
         else:
             # NDBranch
-            log(f'{len(next_states)} non-deterministic branches: {node_id}')
             next_cterms = [cterm for cterm, _ in next_states]
-            return NDBranch(next_cterms, next_node_logs, self._extract_rule_labels(next_node_logs))
+            (next_node_logs, rules) = (
+                ((), []) if extend_results else (next_node_logs, self._extract_rule_labels(next_node_logs))
+            )
+            extend_results.append(NDBranch(next_cterms, next_node_logs, rules))
+
+        return extend_results

pyk/src/pyk/kcfg/kcfg.py

@@ -509,30 +509,45 @@ def extend(
         node: KCFG.Node,
         logs: dict[int, tuple[LogEntry, ...]],
     ) -> None:
+
+        def log(message: str, *, warning: bool = False) -> None:
+            result_info = extend_result.info if type(extend_result) is Step or type(extend_result) is Branch else ''
+            result_info_message = f': {result_info}' if result_info else ''
+            _LOGGER.log(
+                logging.WARNING if warning else logging.INFO,
+                f'Extending current KCFG with the following: {message}{result_info_message}',
+            )
+
         match extend_result:
             case Vacuous():
                 self.add_vacuous(node.id)
+                log(f'vacuous node: {node.id}', warning=True)
 
             case Stuck():
                 self.add_stuck(node.id)
+                log(f'stuck node: {node.id}')
 
             case Abstract(cterm):
                 new_node = self.create_node(cterm)
                 self.create_cover(node.id, new_node.id)
+                log(f'abstraction node: {node.id}')
 
             case Step(cterm, depth, next_node_logs, rule_labels, _):
                 next_node = self.create_node(cterm)
                 logs[next_node.id] = next_node_logs
                 self.create_edge(node.id, next_node.id, depth, rules=rule_labels)
+                log(f'basic block at depth {depth}: {node.id} --> {next_node.id}')
 
-            case Branch(constraints, _):
-                self.split_on_constraints(node.id, constraints)
+            case Branch(branches, _):
+                branch_node_ids = self.split_on_constraints(node.id, branches)
+                log(f'{len(branches)} branches: {node.id} --> {branch_node_ids}')
 
             case NDBranch(cterms, next_node_logs, rule_labels):
                 next_ids = [self.create_node(cterm).id for cterm in cterms]
                 for i in next_ids:
                     logs[i] = next_node_logs
                 self.create_ndbranch(node.id, next_ids, rules=rule_labels)
+                log(f'{len(cterms)} non-deterministic branches: {node.id} --> {next_ids}')
 
             case _:
                 raise AssertionError()
@@ -1333,17 +1348,20 @@ class Step(KCFGExtendResult):
     logs: tuple[LogEntry, ...]
     rule_labels: list[str]
     cut: bool = field(default=False)
+    info: str = field(default='')
 
 
 @final
 @dataclass(frozen=True)
 class Branch(KCFGExtendResult):
     constraints: tuple[KInner, ...]
     heuristic: bool
+    info: str = field(default='')
 
-    def __init__(self, constraints: Iterable[KInner], *, heuristic: bool = False):
+    def __init__(self, constraints: Iterable[KInner], *, heuristic: bool = False, info: str = ''):
         object.__setattr__(self, 'constraints', tuple(constraints))
         object.__setattr__(self, 'heuristic', heuristic)
+        object.__setattr__(self, 'info', info)
 
 
 @final

pyk/src/pyk/kcfg/semantics.py

@@ -12,15 +12,28 @@ class KCFGSemantics(ABC):
     @abstractmethod
     def is_terminal(self, c: CTerm) -> bool: ...
 
+    """Check whether or not a given ``CTerm`` is terminal."""
+
     @abstractmethod
     def abstract_node(self, c: CTerm) -> CTerm: ...
 
+    """Implement an abstraction mechanism."""
+
     @abstractmethod
     def same_loop(self, c1: CTerm, c2: CTerm) -> bool: ...
 
+    """Check whether or not the two given ``CTerm``s represent the same loop head."""
+
+    @abstractmethod
+    def can_make_custom_step(self, c: CTerm) -> bool: ...
+
+    """Check whether or not the semantics can make a custom step from a given ``CTerm``."""
+
     @abstractmethod
     def custom_step(self, c: CTerm) -> KCFGExtendResult | None: ...
 
+    """Implement a custom semantic step."""
+
 
 class DefaultSemantics(KCFGSemantics):
     def is_terminal(self, c: CTerm) -> bool:
@@ -32,5 +45,8 @@ def abstract_node(self, c: CTerm) -> CTerm:
     def same_loop(self, c1: CTerm, c2: CTerm) -> bool:
         return False
 
+    def can_make_custom_step(self, c: CTerm) -> bool:
+        return False
+
     def custom_step(self, c: CTerm) -> KCFGExtendResult | None:
         return None