MergedEdge class for merging node functionality (#4603)

Here is the alternative implementation for [this
PR](#4600).

- commit 1: introduce the new class `MergedEdge`
- commit 2: add `to_rule` function to `MergedEdge` for CSE
- commit 3: add corresponding functions like `Edge` for `MergedEdge`
- commit 4: modify `Edge.to_rule` related functions for CSE
- commit 5: modify KCFGShow for MergedEdge
- commit 6: modify KCFGViewer for MergedEdge

Author: Stevengre

pyk/src/pyk/kcfg/kcfg.py

@@ -245,6 +245,41 @@ def replace_target(self, node: KCFG.Node) -> KCFG.Edge:
             assert node.id == self.target.id
             return KCFG.Edge(self.source, node, self.depth, self.rules)
 
+    @final
+    @dataclass(frozen=True)
+    class MergedEdge(EdgeLike):
+        """Merged edge is a collection of edges that have been merged into a single edge."""
+
+        source: KCFG.Node
+        target: KCFG.Node
+        edges: tuple[KCFG.Edge, ...]
+
+        def to_dict(self) -> dict[str, Any]:
+            return {
+                'source': self.source.id,
+                'target': self.target.id,
+                'edges': [edge.to_dict() for edge in self.edges],
+            }
+
+        @staticmethod
+        def from_dict(dct: dict[str, Any], nodes: Mapping[int, KCFG.Node]) -> KCFG.Successor:
+            return KCFG.MergedEdge(
+                nodes[dct['source']],
+                nodes[dct['target']],
+                tuple(KCFG.Edge.from_dict(edge, nodes) for edge in dct['edges']),
+            )
+
+        def replace_source(self, node: KCFG.Node) -> KCFG.Successor:
+            assert node.id == self.source.id
+            return KCFG.MergedEdge(node, self.target, self.edges)
+
+        def replace_target(self, node: KCFG.Node) -> KCFG.Successor:
+            assert node.id == self.target.id
+            return KCFG.MergedEdge(self.source, node, self.edges)
+
+        def to_rule(self, label: str, claim: bool = False, priority: int | None = None) -> KRuleLike:
+            return KCFG.Edge(self.source, self.target, 1, ()).to_rule(label, claim, priority)
+
     @final
     @dataclass(frozen=True)
     class Cover(EdgeLike):
@@ -389,6 +424,7 @@ def replace_target(self, node: KCFG.Node) -> KCFG.NDBranch:
     _deleted_nodes: set[int]
 
     _edges: dict[int, Edge]
+    _merged_edges: dict[int, MergedEdge]
     _covers: dict[int, Cover]
     _splits: dict[int, Split]
     _ndbranches: dict[int, NDBranch]
@@ -408,6 +444,7 @@ def __init__(self, cfg_dir: Path | None = None, optimize_memory: bool = True) ->
         self._created_nodes = set()
         self._deleted_nodes = set()
         self._edges = {}
+        self._merged_edges = {}
         self._covers = {}
         self._splits = {}
         self._ndbranches = {}
@@ -421,6 +458,8 @@ def __contains__(self, item: object) -> bool:
             return self.contains_node(item)
         if type(item) is KCFG.Edge:
             return self.contains_edge(item)
+        if type(item) is KCFG.MergedEdge:
+            return self.contains_merged_edge(item)
         if type(item) is KCFG.Cover:
             return self.contains_cover(item)
         if type(item) is KCFG.Split:
@@ -502,6 +541,8 @@ def path_length(_path: Iterable[KCFG.Successor]) -> int:
             return 1 + KCFG.path_length(_path[1:])
         elif type(_path[0]) is KCFG.Edge:
             return _path[0].depth + KCFG.path_length(_path[1:])
+        elif type(_path[0]) is KCFG.MergedEdge:
+            return min(edge.depth for edge in _path[0].edges) + KCFG.path_length(_path[1:])  # todo: check this
         raise ValueError(f'Cannot handle Successor type: {type(_path[0])}')
 
     def extend(
@@ -576,6 +617,7 @@ def to_dict_no_nodes(self) -> dict[str, Any]:
     def to_dict(self) -> dict[str, Any]:
         nodes = [node.to_dict() for node in self.nodes]
         edges = [edge.to_dict() for edge in self.edges()]
+        merged_edges = [merged_edge.to_dict() for merged_edge in self.merged_edges()]
         covers = [cover.to_dict() for cover in self.covers()]
         splits = [split.to_dict() for split in self.splits()]
         ndbranches = [ndbranch.to_dict() for ndbranch in self.ndbranches()]
@@ -586,6 +628,7 @@ def to_dict(self) -> dict[str, Any]:
             'next': self._node_id,
             'nodes': nodes,
             'edges': edges,
+            'merged_edges': merged_edges,
             'covers': covers,
             'splits': splits,
             'ndbranches': ndbranches,
@@ -608,6 +651,10 @@ def from_dict(dct: Mapping[str, Any], optimize_memory: bool = True) -> KCFG:
             edge = KCFG.Edge.from_dict(edge_dict, cfg._nodes)
             cfg.add_successor(edge)
 
+        for edge_dict in dct.get('merged_edges') or []:
+            merged_edge = KCFG.MergedEdge.from_dict(edge_dict, cfg._nodes)
+            cfg.add_successor(merged_edge)
+
         for cover_dict in dct.get('covers') or []:
             cover = KCFG.Cover.from_dict(cover_dict, cfg._nodes)
             cfg.add_successor(cover)
@@ -636,9 +683,11 @@ def to_json(self) -> str:
     def from_json(s: str, optimize_memory: bool = True) -> KCFG:
         return KCFG.from_dict(json.loads(s), optimize_memory=optimize_memory)
 
-    def to_rules(self, priority: int = 20, id: str | None = None) -> list[KRuleLike]:
-        id = '' if id is None else f'{id}-'
-        return [e.to_rule(f'{id}BASIC-BLOCK', priority=priority) for e in self.edges()]
+    def to_rules(self, _id: str | None = None, priority: int = 20) -> list[KRuleLike]:
+        _id = 'BASIC-BLOCK' if _id is None else _id
+        return [e.to_rule(_id, priority=priority) for e in self.edges()] + [
+            m.to_rule(_id, priority=priority) for m in self.merged_edges()
+        ]
 
     def to_module(
         self,
@@ -707,6 +756,9 @@ def remove_node(self, node_id: NodeIdLike) -> None:
         self._deleted_nodes.add(node.id)
 
         self._edges = {k: s for k, s in self._edges.items() if k != node_id and node_id not in s.target_ids}
+        self._merged_edges = {
+            k: s for k, s in self._merged_edges.items() if k != node_id and node_id not in s.target_ids
+        }
         self._covers = {k: s for k, s in self._covers.items() if k != node_id and node_id not in s.target_ids}
 
         self._splits = {k: s for k, s in self._splits.items() if k != node_id and node_id not in s.target_ids}
@@ -721,6 +773,8 @@ def _update_refs(self, node_id: int) -> None:
             new_succ = succ.replace_source(node)
             if type(new_succ) is KCFG.Edge:
                 self._edges[new_succ.source.id] = new_succ
+            if type(new_succ) is KCFG.MergedEdge:
+                self._merged_edges[new_succ.source.id] = new_succ
             if type(new_succ) is KCFG.Cover:
                 self._covers[new_succ.source.id] = new_succ
             if type(new_succ) is KCFG.Split:
@@ -732,6 +786,8 @@ def _update_refs(self, node_id: int) -> None:
             new_pred = pred.replace_target(node)
             if type(new_pred) is KCFG.Edge:
                 self._edges[new_pred.source.id] = new_pred
+            if type(new_pred) is KCFG.MergedEdge:
+                self._merged_edges[new_pred.source.id] = new_pred
             if type(new_pred) is KCFG.Cover:
                 self._covers[new_pred.source.id] = new_pred
             if type(new_pred) is KCFG.Split:
@@ -768,17 +824,19 @@ def replace_node(self, node: KCFG.Node) -> None:
 
     def successors(self, source_id: NodeIdLike) -> list[Successor]:
         out_edges: Iterable[KCFG.Successor] = self.edges(source_id=source_id)
+        out_merged_edges: Iterable[KCFG.Successor] = self.merged_edges(source_id=source_id)
         out_covers: Iterable[KCFG.Successor] = self.covers(source_id=source_id)
         out_splits: Iterable[KCFG.Successor] = self.splits(source_id=source_id)
         out_ndbranches: Iterable[KCFG.Successor] = self.ndbranches(source_id=source_id)
-        return list(out_edges) + list(out_covers) + list(out_splits) + list(out_ndbranches)
+        return list(out_edges) + list(out_merged_edges) + list(out_covers) + list(out_splits) + list(out_ndbranches)
 
     def predecessors(self, target_id: NodeIdLike) -> list[Successor]:
         in_edges: Iterable[KCFG.Successor] = self.edges(target_id=target_id)
+        in_merged_edges: Iterable[KCFG.Successor] = self.merged_edges(target_id=target_id)
         in_covers: Iterable[KCFG.Successor] = self.covers(target_id=target_id)
         in_splits: Iterable[KCFG.Successor] = self.splits(target_id=target_id)
         in_ndbranches: Iterable[KCFG.Successor] = self.ndbranches(target_id=target_id)
-        return list(in_edges) + list(in_covers) + list(in_splits) + list(in_ndbranches)
+        return list(in_edges) + list(in_merged_edges) + list(in_covers) + list(in_splits) + list(in_ndbranches)
 
     def _check_no_successors(self, source_id: NodeIdLike) -> None:
         if len(self.successors(source_id)) > 0:
@@ -797,6 +855,8 @@ def add_successor(self, succ: KCFG.Successor) -> None:
         self._check_no_zero_loops(succ.source.id, succ.target_ids)
         if type(succ) is KCFG.Edge:
             self._edges[succ.source.id] = succ
+        elif type(succ) is KCFG.MergedEdge:
+            self._merged_edges[succ.source.id] = succ
         elif type(succ) is KCFG.Cover:
             self._covers[succ.source.id] = succ
         else:
@@ -846,6 +906,46 @@ def remove_edge(self, source_id: NodeIdLike, target_id: NodeIdLike) -> None:
             raise ValueError(f'Edge does not exist: {source_id} -> {target_id}')
         self._edges.pop(source_id)
 
+    def merged_edge(self, source_id: NodeIdLike, target_id: NodeIdLike) -> MergedEdge | None:
+        source_id = self._resolve(source_id)
+        target_id = self._resolve(target_id)
+        merged_edge = self._merged_edges.get(source_id, None)
+        return merged_edge if merged_edge is not None and merged_edge.target.id == target_id else None
+
+    def merged_edges(
+        self, *, source_id: NodeIdLike | None = None, target_id: NodeIdLike | None = None
+    ) -> list[MergedEdge]:
+        source_id = self._resolve(source_id) if source_id is not None else None
+        target_id = self._resolve(target_id) if target_id is not None else None
+        return [
+            merged_edge
+            for merged_edge in self._merged_edges.values()
+            if (source_id is None or source_id == merged_edge.source.id)
+            and (target_id is None or target_id == merged_edge.target.id)
+        ]
+
+    def contains_merged_edge(self, edge: MergedEdge) -> bool:
+        if other := self.merged_edge(source_id=edge.source.id, target_id=edge.target.id):
+            return edge == other
+        return False
+
+    def create_merged_edge(self, source_id: NodeIdLike, target_id: NodeIdLike, edges: Iterable[Edge]) -> MergedEdge:
+        if len(list(edges)) == 0:
+            raise ValueError(f'Cannot build KCFG MergedEdge with no edges: {edges}')
+        source = self.node(source_id)
+        target = self.node(target_id)
+        merged_edge = KCFG.MergedEdge(source, target, tuple(edges))
+        self.add_successor(merged_edge)
+        return merged_edge
+
+    def remove_merged_edge(self, source_id: NodeIdLike, target_id: NodeIdLike) -> None:
+        source_id = self._resolve(source_id)
+        target_id = self._resolve(target_id)
+        merged_edge = self.merged_edge(source_id, target_id)
+        if not merged_edge:
+            raise ValueError(f'MergedEdge does not exist: {source_id} -> {target_id}')
+        self._merged_edges.pop(source_id)
+
     def cover(self, source_id: NodeIdLike, target_id: NodeIdLike) -> Cover | None:
         source_id = self._resolve(source_id)
         target_id = self._resolve(target_id)
@@ -887,8 +987,10 @@ def remove_cover(self, source_id: NodeIdLike, target_id: NodeIdLike) -> None:
         self._covers.pop(source_id)
 
     def edge_likes(self, *, source_id: NodeIdLike | None = None, target_id: NodeIdLike | None = None) -> list[EdgeLike]:
-        return cast('List[KCFG.EdgeLike]', self.edges(source_id=source_id, target_id=target_id)) + cast(
-            'List[KCFG.EdgeLike]', self.covers(source_id=source_id, target_id=target_id)
+        return (
+            cast('List[KCFG.EdgeLike]', self.edges(source_id=source_id, target_id=target_id))
+            + cast('List[KCFG.EdgeLike]', self.covers(source_id=source_id, target_id=target_id))
+            + cast('List[KCFG.EdgeLike]', self.merged_edges(source_id=source_id, target_id=target_id))
         )
 
     def add_vacuous(self, node_id: NodeIdLike) -> None:

pyk/src/pyk/kcfg/show.py

@@ -135,6 +135,13 @@ def _print_edge(edge: KCFG.Edge) -> list[str]:
             else:
                 return ['(' + str(edge.depth) + ' steps)']
 
+        def _print_merged_edge(merged_edge: KCFG.MergedEdge) -> list[str]:
+            res = '('
+            for edge in merged_edge.edges:
+                res += f'{edge.depth}|'
+            res = res[:-1] + ' steps)'
+            return [res] if len(res) < 78 else ['(merged edge)']
+
         def _print_cover(cover: KCFG.Cover) -> Iterable[str]:
             subst_strs = [f'{k} <- {self.kprint.pretty_print(v)}' for k, v in cover.csubst.subst.items()]
             subst_str = ''
@@ -249,6 +256,11 @@ def _print_subgraph(indent: str, curr_node: KCFG.Node, prior_on_trace: list[KCFG
                     ret_edge_lines.extend(add_indent(indent + '│  ', _print_edge(successor)))
                     ret_lines.append((f'edge_{successor.source.id}_{successor.target.id}', ret_edge_lines))
 
+                elif type(successor) is KCFG.MergedEdge:
+                    ret_edge_lines = []
+                    ret_edge_lines.extend(add_indent(indent + '│  ', _print_merged_edge(successor)))
+                    ret_lines.append((f'merged_edge_{successor.source.id}_{successor.target.id}', ret_edge_lines))
+
                 elif type(successor) is KCFG.Cover:
                     ret_edge_lines = []
                     ret_edge_lines.extend(add_indent(indent + '┊  ', _print_cover(successor)))

pyk/src/pyk/kcfg/tui.py

@@ -224,6 +224,10 @@ def _info_text(self) -> str:
             element_str = f'node({shorten_hashes(self._element.id)})'
         elif type(self._element) is KCFG.Edge:
             element_str = f'edge({shorten_hashes(self._element.source.id)},{shorten_hashes(self._element.target.id)})'
+        elif type(self._element) is KCFG.MergedEdge:
+            element_str = (
+                f'merged_edge({shorten_hashes(self._element.source.id)},{shorten_hashes(self._element.target.id)})'
+            )
         elif type(self._element) is KCFG.Cover:
             element_str = f'cover({shorten_hashes(self._element.source.id)},{shorten_hashes(self._element.target.id)})'
         return f'{element_str} selected. {minimize_str} Minimize Output. {term_str} Term View. {constraint_str} Constraint View. {status_str} Status View. {custom_str}'
@@ -296,6 +300,14 @@ def _cterm_text(cterm: CTerm) -> tuple[str, str]:
                 crewrite = CTerm(config, constraints_new)
                 term_str, constraint_str = _cterm_text(crewrite)
 
+            elif type(self._element) is KCFG.MergedEdge:
+                config_source, *constraints_source = self._element.source.cterm
+                config_target, *constraints_target = self._element.target.cterm
+                constraints_new = [c for c in constraints_target if c not in constraints_source]
+                config = push_down_rewrites(KRewrite(config_source, config_target))
+                crewrite = CTerm(config, constraints_new)
+                term_str, constraint_str = _cterm_text(crewrite)
+
             elif type(self._element) is KCFG.Cover:
                 subst_equalities = map(_boolify, flatten_label('#And', self._element.csubst.subst.ml_pred))
                 constraints = map(_boolify, flatten_label('#And', self._element.csubst.constraint))
@@ -421,6 +433,14 @@ def on_graph_chunk_selected(self, message: GraphChunk.Selected) -> None:
             edge = single(self._kcfg.edges(source_id=source_id, target_id=target_id))
             self.query_one('#node-view', NodeView).update(edge)
 
+        elif message.chunk_id.startswith('merged_edge_'):
+            self._selected_chunk = None
+            node_source, node_target, *_ = message.chunk_id[12:].split('_')
+            source_id = int(node_source)
+            target_id = int(node_target)
+            merged_edge = single(self._kcfg.merged_edges(source_id=source_id, target_id=target_id))
+            self.query_one('#node-view', NodeView).update(merged_edge)
+
         elif message.chunk_id.startswith('cover_'):
             self._selected_chunk = None
             node_source, node_target, *_ = message.chunk_id[6:].split('_')

pyk/src/pyk/proof/reachability.py

@@ -26,7 +26,7 @@
     from pathlib import Path
     from typing import Any, Final, TypeVar
 
-    from ..kast.outer import KClaim, KDefinition, KFlatModuleList
+    from ..kast.outer import KClaim, KDefinition, KFlatModuleList, KRuleLike
     from ..kcfg import KCFGExplore
     from ..kcfg.explore import KCFGExtendResult
     from ..kcfg.kcfg import CSubst, NodeIdLike
@@ -444,12 +444,12 @@ def as_rules(self, priority: int = 20, direct_rule: bool = False) -> list[KRule]
             or (self.admitted and not self.kcfg.predecessors(self.target))
         ):
             return [self.as_rule(priority=priority)]
-        _rules = []
-        for _edge in self.kcfg.edges():
-            _rule = _edge.to_rule(self.rule_id, priority=priority)
-            assert type(_rule) is KRule
-            _rules.append(_rule)
-        return _rules
+
+        def _return_rule(r: KRuleLike) -> KRule:
+            assert isinstance(r, KRule)
+            return r
+
+        return [_return_rule(rule) for rule in self.kcfg.to_rules(self.rule_id, priority=priority)]
 
     def as_rule(self, priority: int = 20) -> KRule:
         _edge = KCFG.Edge(self.kcfg.node(self.init), self.kcfg.node(self.target), depth=0, rules=())