Add support for vacuous nodes

package/version

@@ -1 +1 @@
-0.1.437
+0.1.438

pyproject.toml

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

src/pyk/kcfg/exploration.py

@@ -29,7 +29,12 @@ def is_terminal(self, node_id: NodeIdLike) -> bool:
 
     # Explorable node recogniser
     def is_explorable(self, node_id: NodeIdLike) -> bool:
-        return self.kcfg.is_leaf(node_id) and not self.is_terminal(node_id) and not self.kcfg.is_stuck(node_id)
+        return (
+            self.kcfg.is_leaf(node_id)
+            and not self.is_terminal(node_id)
+            and not self.kcfg.is_stuck(node_id)
+            and not self.kcfg.is_vacuous(node_id)
+        )
 
     #
     # Collectors

src/pyk/kcfg/explore.py

@@ -75,7 +75,7 @@ def cterm_execute(
         cut_point_rules: Iterable[str] | None = None,
         terminal_rules: Iterable[str] | None = None,
         module_name: str | None = None,
-    ) -> tuple[int, CTerm, list[CTerm], tuple[LogEntry, ...]]:
+    ) -> tuple[bool, int, CTerm, list[CTerm], tuple[LogEntry, ...]]:
         _LOGGER.debug(f'Executing: {cterm}')
         kore = self.kprint.kast_to_kore(cterm.kast, GENERATED_TOP_CELL)
         er = self._kore_client.execute(
@@ -89,19 +89,20 @@ def cterm_execute(
             log_successful_simplifications=self._trace_rewrites,
             log_failed_simplifications=self._trace_rewrites,
         )
+        _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 depth + 1, next_states[0], [], er.logs
+            return _is_vacuous, depth + 1, next_states[0], [], er.logs
         elif len(next_states) == 1:
             if er.reason == StopReason.CUT_POINT_RULE:
-                return depth, next_state, next_states, er.logs
+                return _is_vacuous, depth, next_state, next_states, er.logs
             else:
                 next_states = []
-        return depth, next_state, next_states, er.logs
+        return _is_vacuous, depth, next_state, next_states, er.logs
 
     def cterm_simplify(self, cterm: CTerm) -> tuple[CTerm, tuple[LogEntry, ...]]:
         _LOGGER.debug(f'Simplifying: {cterm}')
@@ -294,7 +295,7 @@ 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(
+        _, actual_depth, cterm, next_cterms, next_node_logs = self.cterm_execute(
             node.cterm, depth=depth, module_name=module_name
         )
         if actual_depth != depth:
@@ -368,6 +369,8 @@ def extend(
             raise ValueError(f'Cannot extend non-leaf node {self.id}: {node.id}')
         if kcfg.is_stuck(node.id):
             raise ValueError(f'Cannot extend stuck node {self.id}: {node.id}')
+        if kcfg.is_vacuous(node.id):
+            raise ValueError(f'Cannot extend vacuous node {self.id}: {node.id}')
         if kcfg_exploration.is_terminal(node.id):
             raise ValueError(f'Cannot extend terminal node {self.id}: {node.id}')
 
@@ -384,7 +387,7 @@ def extend(
                 return
 
         _LOGGER.info(f'Extending KCFG from node {self.id}: {shorten_hashes(node.id)}')
-        depth, cterm, next_cterms, next_node_logs = self.cterm_execute(
+        _is_vacuous, depth, cterm, next_cterms, next_node_logs = self.cterm_execute(
             node.cterm,
             depth=execute_depth,
             cut_point_rules=cut_point_rules,
@@ -403,8 +406,12 @@ def extend(
 
         # Stuck
         elif len(next_cterms) == 0:
-            kcfg.add_stuck(node.id)
-            _LOGGER.info(f'Found stuck node {self.id}: {shorten_hashes(node.id)}')
+            if _is_vacuous:
+                kcfg.add_vacuous(node.id)
+                _LOGGER.warning(f'Found vacuous node {self.id}: {shorten_hashes(node.id)}')
+            else:
+                kcfg.add_stuck(node.id)
+                _LOGGER.info(f'Found stuck node {self.id}: {shorten_hashes(node.id)}')
 
         # Cut Rule
         elif len(next_cterms) == 1:

src/pyk/kcfg/kcfg.py

@@ -205,6 +205,7 @@ def edges(self) -> tuple[KCFG.Edge, ...]:
     _covers: dict[int, dict[int, Cover]]
     _splits: dict[int, Split]
     _ndbranches: dict[int, NDBranch]
+    _vacuous: set[int]
     _stuck: set[int]
     _aliases: dict[str, int]
     _lock: RLock
@@ -219,6 +220,7 @@ def __init__(self, cfg_dir: Path | None = None) -> None:
         self._covers = {}
         self._splits = {}
         self._ndbranches = {}
+        self._vacuous = set()
         self._stuck = set()
         self._aliases = {}
         self._lock = RLock()
@@ -262,6 +264,10 @@ def nodes(self) -> list[Node]:
     def root(self) -> list[Node]:
         return [node for node in self.nodes if self.is_root(node.id)]
 
+    @property
+    def vacuous(self) -> list[Node]:
+        return [node for node in self.nodes if node.id in self._vacuous]
+
     @property
     def stuck(self) -> list[Node]:
         return [node for node in self.nodes if node.id in self._stuck]
@@ -315,6 +321,7 @@ def to_dict(self) -> dict[str, Any]:
         splits = [split.to_dict() for split in self.splits()]
         ndbranches = [ndbranch.to_dict() for ndbranch in self.ndbranches()]
 
+        vacuous = sorted(self._vacuous)
         stuck = sorted(self._stuck)
         aliases = dict(sorted(self._aliases.items()))
 
@@ -325,6 +332,7 @@ def to_dict(self) -> dict[str, Any]:
             'covers': covers,
             'splits': splits,
             'ndbranches': ndbranches,
+            'vacuous': vacuous,
             'stuck': stuck,
             'aliases': aliases,
         }
@@ -356,6 +364,9 @@ def from_dict(dct: Mapping[str, Any]) -> KCFG:
             csubst = CSubst.from_dict(cover_dict['csubst'])
             cfg.create_cover(source_id, target_id, csubst=csubst)
 
+        for vacuous_id in dct.get('vacuous') or []:
+            cfg.add_vacuous(vacuous_id)
+
         for stuck_id in dct.get('stuck') or []:
             cfg.add_stuck(stuck_id)
 
@@ -470,6 +481,7 @@ def remove_node(self, node_id: NodeIdLike) -> None:
             if not self._covers[source_id]:
                 self._covers.pop(source_id)
 
+        self._vacuous.discard(node_id)
         self._stuck.discard(node_id)
 
         self._splits = {k: s for k, s in self._splits.items() if k != node_id and node_id not in s.target_ids}
@@ -636,6 +648,19 @@ def edge_likes(self, *, source_id: NodeIdLike | None = None, target_id: NodeIdLi
             'List[KCFG.EdgeLike]', self.covers(source_id=source_id, target_id=target_id)
         )
 
+    def add_vacuous(self, node_id: NodeIdLike) -> None:
+        self._vacuous.add(self._resolve(node_id))
+
+    def remove_vacuous(self, node_id: NodeIdLike) -> None:
+        node_id = self._resolve(node_id)
+        if node_id not in self._vacuous:
+            raise ValueError(f'Node is not vacuous: {node_id}')
+        self._vacuous.remove(node_id)
+
+    def discard_vacuous(self, node_id: NodeIdLike) -> None:
+        node_id = self._resolve(node_id)
+        self._vacuous.discard(node_id)
+
     def add_stuck(self, node_id: NodeIdLike) -> None:
         self._stuck.add(self._resolve(node_id))
 
@@ -723,6 +748,10 @@ def is_root(self, node_id: NodeIdLike) -> bool:
         node_id = self._resolve(node_id)
         return len(self.predecessors(node_id)) == 0
 
+    def is_vacuous(self, node_id: NodeIdLike) -> bool:
+        node_id = self._resolve(node_id)
+        return node_id in self._vacuous
+
     def is_stuck(self, node_id: NodeIdLike) -> bool:
         node_id = self._resolve(node_id)
         return node_id in self._stuck
@@ -888,6 +917,7 @@ def write_cfg_data(self) -> None:
         splits = [split.to_dict() for split in self.splits()]
         ndbranches = [ndbranch.to_dict() for ndbranch in self.ndbranches()]
 
+        vacuous = sorted(self._vacuous)
         stuck = sorted(self._stuck)
         aliases = dict(sorted(self._aliases.items()))
         dct: dict[str, list[int] | int | dict[str, int] | list[dict[str, Any]]] = {}
@@ -897,6 +927,7 @@ def write_cfg_data(self) -> None:
         dct['covers'] = covers
         dct['splits'] = splits
         dct['ndbranches'] = ndbranches
+        dct['vacuous'] = vacuous
         dct['stuck'] = stuck
         dct['aliases'] = aliases
         cfg_json.write_text(json.dumps(dct))
@@ -964,6 +995,9 @@ def read_cfg_data(cfg_dir: Path, id: str) -> KCFG:
             csubst = CSubst.from_dict(cover_dict['csubst'])
             cfg.create_cover(source_id, target_id, csubst=csubst)
 
+        for vacuous_id in dct.get('vacuous') or []:
+            cfg.add_vacuous(vacuous_id)
+
         for stuck_id in dct.get('stuck') or []:
             cfg.add_stuck(stuck_id)
 

src/pyk/kcfg/show.py

@@ -60,6 +60,8 @@ def node_attrs(self, kcfg: KCFG, node: KCFG.Node) -> list[str]:
             attrs.append('root')
         if kcfg.is_stuck(node.id):
             attrs.append('stuck')
+        if kcfg.is_vacuous(node.id):
+            attrs.append('vacuous')
         if kcfg.is_leaf(node.id):
             attrs.append('leaf')
         if kcfg.is_split(node.id):

src/pyk/kore/rpc.py

@@ -343,6 +343,7 @@ class StopReason(str, Enum):
     BRANCHING = 'branching'
     CUT_POINT_RULE = 'cut-point-rule'
     TERMINAL_RULE = 'terminal-rule'
+    VACUOUS = 'vacuous'
 
 
 @final
@@ -493,6 +494,7 @@ class ExecuteResult(ABC):  # noqa: B024
         StopReason.BRANCHING: 'BranchingResult',
         StopReason.CUT_POINT_RULE: 'CutPointResult',
         StopReason.TERMINAL_RULE: 'TerminalResult',
+        StopReason.VACUOUS: 'VacuousResult',
     }
 
     reason: ClassVar[StopReason]
@@ -625,6 +627,28 @@ def from_dict(cls: type[TerminalResult], dct: Mapping[str, Any]) -> TerminalResu
         return TerminalResult(state=State.from_dict(dct['state']), depth=dct['depth'], rule=dct['rule'], logs=logs)
 
 
+@final
+@dataclass(frozen=True)
+class VacuousResult(ExecuteResult):
+    reason = StopReason.VACUOUS
+    next_states = None
+    rule = None
+
+    state: State
+    depth: int
+    logs: tuple[LogEntry, ...]
+
+    @classmethod
+    def from_dict(cls: type[VacuousResult], dct: Mapping[str, Any]) -> VacuousResult:
+        cls._check_reason(dct)
+        logs = tuple(LogEntry.from_dict(l) for l in dct['logs']) if 'logs' in dct else ()
+        return VacuousResult(
+            state=State.from_dict(dct['state']),
+            depth=dct['depth'],
+            logs=logs,
+        )
+
+
 @final
 @dataclass(frozen=True)
 class ImpliesResult:

src/pyk/proof/reachability.py

@@ -115,6 +115,7 @@ def is_failing(self, node_id: NodeIdLike) -> bool:
             and not self.is_explorable(node_id)
             and not self.is_target(node_id)
             and not self.is_refuted(node_id)
+            and not self.kcfg.is_vacuous(node_id)
         )
 
     def shortest_path_to(self, node_id: NodeIdLike) -> tuple[KCFG.Successor, ...]:
@@ -246,6 +247,7 @@ def summary(self) -> CompositeSummary:
                     len(self.kcfg.nodes),
                     len(self.pending),
                     len(self.failing),
+                    len(self.kcfg.vacuous),
                     len(self.kcfg.stuck),
                     len(self._terminal),
                     len(self.node_refutations),
@@ -514,6 +516,7 @@ def summary(self) -> CompositeSummary:
                     len(self.kcfg.nodes),
                     len(self.pending),
                     len(self.failing),
+                    len(self.kcfg.vacuous),
                     len(self.kcfg.stuck),
                     len(self._terminal),
                     len(self.node_refutations),
@@ -739,6 +742,7 @@ class APRSummary(ProofSummary):
     nodes: int
     pending: int
     failing: int
+    vacuous: int
     stuck: int
     terminal: int
     refuted: int
@@ -753,6 +757,7 @@ def lines(self) -> list[str]:
             f'    nodes: {self.nodes}',
             f'    pending: {self.pending}',
             f'    failing: {self.failing}',
+            f'    vacuous: {self.vacuous}',
             f'    stuck: {self.stuck}',
             f'    terminal: {self.terminal}',
             f'    refuted: {self.refuted}',
@@ -913,6 +918,7 @@ class APRBMCSummary(ProofSummary):
     nodes: int
     pending: int
     failing: int
+    vacuous: int
     stuck: int
     terminal: int
     refuted: int
@@ -927,6 +933,7 @@ def lines(self) -> list[str]:
             f'    nodes: {self.nodes}',
             f'    pending: {self.pending}',
             f'    failing: {self.failing}',
+            f'    vacuous: {self.vacuous}',
             f'    stuck: {self.stuck}',
             f'    terminal: {self.terminal}',
             f'    refuted: {self.refuted}',

src/tests/integration/k-files/kore-rpc-test.k

@@ -1,16 +1,19 @@
 module KORE-RPC-TEST-SYNTAX
     imports INT-SYNTAX
+    imports BOOL-SYNTAX
 endmodule
 
 
 module KORE-RPC-TEST
     imports KORE-RPC-TEST-SYNTAX
     imports INT
+    imports BOOL
 
     rule [r01]: 0 => 1
     rule [r12]: 1 => 2
     rule [r23]: 2 => 3
     rule [r24]: 2 => 4
+    rule [r34]: 3 => 4 ensures false
     rule [r45]: 4 => 5
     rule [r56]: 5 => 6
 

src/tests/integration/kcfg/test_cell_map.py

@@ -104,7 +104,7 @@ def test_execute(
         expected_k, _, _ = expected_post
 
         # When
-        actual_depth, actual_post_term, _, _logs = kcfg_explore.cterm_execute(
+        _, 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'))

src/tests/integration/kcfg/test_imp.py

@@ -780,7 +780,7 @@ def test_execute(
         expected_k, expected_state = expected_post
 
         # When
-        actual_depth, actual_post_term, actual_next_terms, _logs = kcfg_explore.cterm_execute(
+        _, 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'))

src/tests/integration/kcfg/test_multiple_definitions.py

@@ -44,7 +44,7 @@ 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)
+        _, split_depth, split_post_term, split_next_terms, _logs = kcfg_explore.cterm_execute(self.config(), depth=1)
 
         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]
@@ -57,13 +57,13 @@ def test_execute(
             'a ( X:KItem )',
         ] == split_next_k
 
-        step_1_depth, step_1_post_term, step_1_next_terms, _logs = kcfg_explore.cterm_execute(
+        _, 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'))
         assert 'c' == step_1_k
 
-        step_2_depth, step_2_post_term, step_2_next_terms, _logs = kcfg_explore.cterm_execute(
+        _, 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'))