Merge remote-tracking branch 'origin/develop'

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