Refactor cterm_execute (#773)

* Extract class `CTermExecute` to represent the return value
* Assert some properties on the RPC response and simplify logic

---------

Co-authored-by: devops <[email protected]>

Author: tothtamas28

package/version

@@ -1 +1 @@
-0.1.542
+0.1.543

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "pyk"
-version = "0.1.542"
+version = "0.1.543"
 description = ""
 authors = [
     "Runtime Verification, Inc. <[email protected]>",

src/pyk/kcfg/explore.py

@@ -3,7 +3,7 @@
 import logging
 from abc import ABC
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, final
+from typing import TYPE_CHECKING, NamedTuple, final
 
 from ..cterm import CSubst, CTerm
 from ..kast.inner import KApply, KLabel, KRewrite, KVariable, Subst
@@ -47,6 +47,14 @@
 _LOGGER: Final = logging.getLogger(__name__)
 
 
+class CTermExecute(NamedTuple):
+    state: CTerm
+    next_states: tuple[CTerm, ...]
+    depth: int
+    vacuous: bool
+    logs: tuple[LogEntry, ...]
+
+
 class KCFGExplore:
     kprint: KPrint
     _kore_client: KoreClient
@@ -77,39 +85,39 @@ def cterm_execute(
         cut_point_rules: Iterable[str] | None = None,
         terminal_rules: Iterable[str] | None = None,
         module_name: str | None = None,
-    ) -> tuple[bool, int, CTerm, list[CTerm], tuple[LogEntry, ...]]:
+    ) -> CTermExecute:
         _LOGGER.debug(f'Executing: {cterm}')
         kore = self.kprint.kast_to_kore(cterm.kast, GENERATED_TOP_CELL)
-        er = self._kore_client.execute(
+        response = self._kore_client.execute(
             kore,
             max_depth=depth,
             cut_point_rules=cut_point_rules,
             terminal_rules=terminal_rules,
             module_name=module_name,
-            log_successful_rewrites=self._trace_rewrites if self._trace_rewrites else None,
-            log_failed_rewrites=self._trace_rewrites if self._trace_rewrites else None,
-            log_successful_simplifications=self._trace_rewrites if self._trace_rewrites else None,
-            log_failed_simplifications=self._trace_rewrites if self._trace_rewrites else None,
+            log_successful_rewrites=self._trace_rewrites,
+            log_failed_rewrites=self._trace_rewrites,
+            log_successful_simplifications=self._trace_rewrites,
+            log_failed_simplifications=self._trace_rewrites,
         )
 
-        if isinstance(er, AbortedResult):
-            unknown_predicate = er.unknown_predicate.text if er.unknown_predicate else None
+        if isinstance(response, AbortedResult):
+            unknown_predicate = response.unknown_predicate.text if response.unknown_predicate else None
             raise ValueError(f'Backend responded with aborted state. Unknown predicate: {unknown_predicate}')
 
-        _is_vacuous = er.reason is StopReason.VACUOUS
-        depth = er.depth
-        next_state = CTerm.from_kast(self.kprint.kore_to_kast(er.state.kore))
-        _next_states = er.next_states if er.next_states is not None else []
-        next_states = [CTerm.from_kast(self.kprint.kore_to_kast(ns.kore)) for ns in _next_states]
-        next_states = [cterm for cterm in next_states if not cterm.is_bottom]
-        if len(next_states) == 1 and len(next_states) < len(_next_states):
-            return _is_vacuous, depth + 1, next_states[0], [], er.logs
-        elif len(next_states) == 1:
-            if er.reason == StopReason.CUT_POINT_RULE:
-                return _is_vacuous, depth, next_state, next_states, er.logs
-            else:
-                next_states = []
-        return _is_vacuous, depth, next_state, next_states, er.logs
+        state = CTerm.from_kast(self.kprint.kore_to_kast(response.state.kore))
+        resp_next_states = response.next_states or ()
+        next_states = tuple(CTerm.from_kast(self.kprint.kore_to_kast(ns.kore)) for ns in resp_next_states)
+
+        assert all(not cterm.is_bottom for cterm in next_states)
+        assert len(next_states) != 1 or response.reason is StopReason.CUT_POINT_RULE
+
+        return CTermExecute(
+            state=state,
+            next_states=next_states,
+            depth=response.depth,
+            vacuous=response.reason is StopReason.VACUOUS,
+            logs=response.logs,
+        )
 
     def cterm_simplify(self, cterm: CTerm) -> tuple[CTerm, tuple[LogEntry, ...]]:
         _LOGGER.debug(f'Simplifying: {cterm}')
@@ -302,16 +310,14 @@ def step(
         if len(successors) != 0 and type(successors[0]) is KCFG.Split:
             raise ValueError(f'Cannot take step from split node {self.id}: {shorten_hashes(node.id)}')
         _LOGGER.info(f'Taking {depth} steps from node {self.id}: {shorten_hashes(node.id)}')
-        _, actual_depth, cterm, next_cterms, next_node_logs = self.cterm_execute(
-            node.cterm, depth=depth, module_name=module_name
-        )
-        if actual_depth != depth:
-            raise ValueError(f'Unable to take {depth} steps from node, got {actual_depth} steps {self.id}: {node.id}')
-        if len(next_cterms) > 0:
+        exec_res = self.cterm_execute(node.cterm, depth=depth, module_name=module_name)
+        if exec_res.depth != depth:
+            raise ValueError(f'Unable to take {depth} steps from node, got {exec_res.depth} steps {self.id}: {node.id}')
+        if len(exec_res.next_states) > 0:
             raise ValueError(f'Found branch within {depth} steps {self.id}: {node.id}')
-        new_node = cfg.create_node(cterm)
+        new_node = cfg.create_node(exec_res.state)
         _LOGGER.info(f'Found new node at depth {depth} {self.id}: {shorten_hashes((node.id, new_node.id))}')
-        logs[new_node.id] = next_node_logs
+        logs[new_node.id] = exec_res.logs
         out_edges = cfg.edges(source_id=node.id)
         if len(out_edges) == 0:
             cfg.create_edge(node.id, new_node.id, depth=depth)
@@ -405,7 +411,7 @@ def extend_cterm(
         if len(branches) > 1:
             return Branch(branches, heuristic=True)
 
-        _is_vacuous, depth, cterm, next_cterms, next_node_logs = self.cterm_execute(
+        cterm, next_cterms, depth, vacuous, next_node_logs = self.cterm_execute(
             _cterm,
             depth=execute_depth,
             cut_point_rules=cut_point_rules,
@@ -419,7 +425,7 @@ def extend_cterm(
 
         # Stuck or vacuous
         if not next_cterms:
-            if _is_vacuous:
+            if vacuous:
                 return Vacuous()
             return Stuck()
 

src/tests/integration/kcfg/test_multiple_definitions.py

@@ -44,27 +44,23 @@ def test_execute(
         kcfg_explore: KCFGExplore,
         test_id: str,
     ) -> None:
-        _, split_depth, split_post_term, split_next_terms, _logs = kcfg_explore.cterm_execute(self.config(), depth=1)
+        exec_res = kcfg_explore.cterm_execute(self.config(), depth=1)
+        split_next_terms = exec_res.next_states
+        split_k = kcfg_explore.kprint.pretty_print(exec_res.state.cell('K_CELL'))
+        split_next_k = [kcfg_explore.kprint.pretty_print(exec_res.state.cell('K_CELL')) for _ in split_next_terms]
 
-        split_k = kcfg_explore.kprint.pretty_print(split_post_term.cell('K_CELL'))
-        split_next_k = [kcfg_explore.kprint.pretty_print(split_post_term.cell('K_CELL')) for term in split_next_terms]
-
-        assert split_depth == 0
+        assert exec_res.depth == 0
         assert len(split_next_terms) == 2
         assert 'a ( X:KItem )' == split_k
         assert [
             'a ( X:KItem )',
             'a ( X:KItem )',
         ] == split_next_k
 
-        _, step_1_depth, step_1_post_term, step_1_next_terms, _logs = kcfg_explore.cterm_execute(
-            split_next_terms[0], depth=1
-        )
-        step_1_k = kcfg_explore.kprint.pretty_print(step_1_post_term.cell('K_CELL'))
+        step_1_res = kcfg_explore.cterm_execute(split_next_terms[0], depth=1)
+        step_1_k = kcfg_explore.kprint.pretty_print(step_1_res.state.cell('K_CELL'))
         assert 'c' == step_1_k
 
-        _, step_2_depth, step_2_post_term, step_2_next_terms, _logs = kcfg_explore.cterm_execute(
-            split_next_terms[1], depth=1
-        )
-        step_2_k = kcfg_explore.kprint.pretty_print(step_1_post_term.cell('K_CELL'))
+        step_2_res = kcfg_explore.cterm_execute(split_next_terms[1], depth=1)
+        step_2_k = kcfg_explore.kprint.pretty_print(step_2_res.state.cell('K_CELL'))
         assert 'c' == step_2_k

src/tests/integration/kcfg/test_simple.py

@@ -78,17 +78,16 @@ def test_execute(
         expected_k, expected_state, *_ = expected_post
 
         # When
-        _, actual_depth, actual_post_term, actual_next_terms, _logs = kcfg_explore.cterm_execute(
-            self.config(kcfg_explore.kprint, *pre), depth=depth
-        )
-        actual_k = kcfg_explore.kprint.pretty_print(actual_post_term.cell('K_CELL'))
-        actual_state = kcfg_explore.kprint.pretty_print(actual_post_term.cell('STATE_CELL'))
+        exec_res = kcfg_explore.cterm_execute(self.config(kcfg_explore.kprint, *pre), depth=depth)
+        actual_k = kcfg_explore.kprint.pretty_print(exec_res.state.cell('K_CELL'))
+        actual_state = kcfg_explore.kprint.pretty_print(exec_res.state.cell('STATE_CELL'))
+        actual_depth = exec_res.depth
         actual_next_states = [
             (
                 kcfg_explore.kprint.pretty_print(s.cell('K_CELL')),
                 kcfg_explore.kprint.pretty_print(s.cell('STATE_CELL')),
             )
-            for s in actual_next_terms
+            for s in exec_res.next_states
         ]
 
         # Then

src/tests/integration/proof/test_cell_map.py

@@ -104,10 +104,9 @@ def test_execute(
         expected_k, _, _ = expected_post
 
         # When
-        _, actual_depth, actual_post_term, _, _logs = kcfg_explore.cterm_execute(
-            self.config(kcfg_explore.kprint, k, aacounts, accounts), depth=depth
-        )
-        actual_k = kcfg_explore.kprint.pretty_print(actual_post_term.cell('K_CELL'))
+        exec_res = kcfg_explore.cterm_execute(self.config(kcfg_explore.kprint, k, aacounts, accounts), depth=depth)
+        actual_k = kcfg_explore.kprint.pretty_print(exec_res.state.cell('K_CELL'))
+        actual_depth = exec_res.depth
 
         # Then
         assert actual_depth == expected_depth

src/tests/integration/proof/test_imp.py

@@ -780,18 +780,17 @@ def test_execute(
         expected_k, expected_state = expected_post
 
         # When
-        _, actual_depth, actual_post_term, actual_next_terms, _logs = kcfg_explore.cterm_execute(
-            self.config(kcfg_explore.kprint, k, state), depth=depth
-        )
-        actual_k = kcfg_explore.kprint.pretty_print(actual_post_term.cell('K_CELL'))
-        actual_state = kcfg_explore.kprint.pretty_print(actual_post_term.cell('STATE_CELL'))
+        exec_res = kcfg_explore.cterm_execute(self.config(kcfg_explore.kprint, k, state), depth=depth)
+        actual_k = kcfg_explore.kprint.pretty_print(exec_res.state.cell('K_CELL'))
+        actual_state = kcfg_explore.kprint.pretty_print(exec_res.state.cell('STATE_CELL'))
+        actual_depth = exec_res.depth
 
         actual_next_states = [
             (
                 kcfg_explore.kprint.pretty_print(s.cell('K_CELL')),
                 kcfg_explore.kprint.pretty_print(s.cell('STATE_CELL')),
             )
-            for s in actual_next_terms
+            for s in exec_res.next_states
         ]
 
         # Then