Code re-organizations prepping for KCFG refactors (#486)

Part of #471, #413

As I was working towards the goal of better BMC prover visualization,
it's clear we need:

- Proof specific viewers,
- Fixing the BMC checker off-by-one-error, and
- Trimming of the structure stored in the KCFG (removing the notion of
`target`, `expanded`).

This makes the first least controversial steps towards those changes,
including:

- Method `KCFG.replace_node` now replaces the node in place, so that the
node identifier doesn't change. This is possible now because the node
identifier is not tied to the contents of the node, and makes tracking
things like `init/target/terminal/bounded` sets externally in the
`*Proof` classes more feasible even if nodes are being simplified for
example.
- The notion of a unique `target` and `init` is added to the `APRProof`
class, in preparation for removing the `target` notion from `KCFG`.
- Add fields `target` and `init` to `APRProof` state, which explicitly
call out the unique init/target states.
- Move `KCFGExplore.target_subsume => APRProver._check_subsumed`, which
makes use of the explicitly called out `init/target` states.
- Moves `KCFG.path_constraints => APRProof.path_constraints`, because it
relies on a unique target node.
- The notion of `stuck` nodes is moved into `APRProof` state as
`terminal` nodes.
  - The field `_terminal_nodes` is added to `APRProof` state.
- `APRProver._check_terminal` directly manipulates this set instead of
`stuck/expanded` set.
- The notion of `APRProof.terminal` is added, which returns the set of
nodes in the `APRProof._terminal_nodes` set.
- The notion of `APRProof.pending` is added, which returns the set of
`KCFG.frontier` nodes that are _not_ in the proofs `terminal_nodes` set.
- The `APRProof.terminal` and `APRProof.pending` sets are used for
determining `APRProof.status`.
- The `APRBMCProof.bounded` set is added, which similarly tracks the
bounded set of nodes directly. This is used in computing
`APRBMCProof.pending` and `APRBMCProof.status`.
- Methods `APRProof.from_claim` and
`APRBMCProof.from_claim_with_bmc_depth` are added for reading these
proof types directly from claims (and tests are updated appropriately).
- Method `APRBMCProof.dict` is refactored to use `APRProof.dict`
directly first, then add the needed information to it.
- Methods `KCFG.node_attrs => KCFGShow.node_attrs` and
`KCFG.node_short_info => KCFGShow.node_short_info` are moved to localize
them to use site.
- Adds `Proof.json` as shortcut for `json.dumps(Proof.dict)`.
- For each of the `integration/kcfg` tests that do `APRProof`s, adds
logging at `info` level for the generated KCFG, to make debugging them
easier by just throwing a flag.

---------

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

Author: ehildenb

pyk/package/version

@@ -1 +1 @@
-0.1.324
+0.1.325

pyk/pyproject.toml

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

pyk/src/pyk/kcfg/explore.py

@@ -369,20 +369,6 @@ def section_edge(
             new_depth += section_depth
         return tuple(new_nodes)
 
-    def target_subsume(
-        self,
-        kcfg: KCFG,
-        node: KCFG.Node,
-    ) -> bool:
-        target_node = kcfg.get_unique_target()
-        _LOGGER.info(f'Checking subsumption into target state {self.id}: {shorten_hashes((node.id, target_node.id))}')
-        csubst = self.cterm_implies(node.cterm, target_node.cterm)
-        if csubst is not None:
-            kcfg.create_cover(node.id, target_node.id, csubst=csubst)
-            _LOGGER.info(f'Subsumed into target node {self.id}: {shorten_hashes((node.id, target_node.id))}')
-            return True
-        return False
-
     def extend(
         self,
         kcfg: KCFG,

pyk/src/pyk/kcfg/kcfg.py

@@ -16,11 +16,10 @@
     remove_source_attributes,
     rename_generated_vars,
 )
-from ..prelude.ml import mlAnd, mlTop
 from ..utils import single
 
 if TYPE_CHECKING:
-    from collections.abc import Callable, Iterable, Mapping
+    from collections.abc import Iterable, Mapping
     from types import TracebackType
     from typing import Any
 
@@ -253,21 +252,21 @@ def stuck(self) -> list[Node]:
         return [node for node in self.nodes if self.is_stuck(node.id)]
 
     @staticmethod
-    def from_claim(defn: KDefinition, claim: KClaim) -> KCFG:
+    def from_claim(defn: KDefinition, claim: KClaim) -> tuple[KCFG, NodeIdLike, NodeIdLike]:
         cfg = KCFG()
         claim_body = claim.body
         claim_body = defn.instantiate_cell_vars(claim_body)
         claim_body = rename_generated_vars(claim_body)
 
         claim_lhs = CTerm.from_kast(extract_lhs(claim_body)).add_constraint(bool_to_ml_pred(claim.requires))
-        init_state = cfg.create_node(claim_lhs)
-        cfg.add_init(init_state.id)
+        init_node = cfg.create_node(claim_lhs)
+        cfg.add_init(init_node.id)
 
         claim_rhs = CTerm.from_kast(extract_rhs(claim_body)).add_constraint(bool_to_ml_pred(claim.ensures))
-        target_state = cfg.create_node(claim_rhs)
-        cfg.add_target(target_state.id)
+        target_node = cfg.create_node(claim_rhs)
+        cfg.add_target(target_node.id)
 
-        return cfg
+        return cfg, init_node.id, target_node.id
 
     def to_dict(self) -> dict[str, Any]:
         nodes = [node.to_dict() for node in self.nodes]
@@ -359,15 +358,6 @@ def to_json(self) -> str:
     def from_json(s: str) -> KCFG:
         return KCFG.from_dict(json.loads(s))
 
-    def node_short_info(self, node: Node, node_printer: Callable[[CTerm], Iterable[str]] | None = None) -> list[str]:
-        attrs = self.node_attrs(node.id) + ['@' + alias for alias in sorted(self.aliases(node.id))]
-        attr_string = ' (' + ', '.join(attrs) + ')' if attrs else ''
-        node_header = str(node.id) + attr_string
-        node_strs = [node_header]
-        if node_printer:
-            node_strs.extend(f' {nl}' for nl in node_printer(node.cterm))
-        return node_strs
-
     def get_unique_init(self) -> Node:
         return single(self.init)
 
@@ -466,40 +456,13 @@ def remove_node(self, node_id: NodeIdLike) -> None:
         for alias in [alias for alias, id in self._aliases.items() if id == node_id]:
             self.remove_alias(alias)
 
-    def replace_node(self, node_id: NodeIdLike, new_cterm: CTerm) -> int:
-        # Remove old node, record data
-        node = self.node(node_id)
-        in_edges = self.edges(target_id=node.id)
-        out_edges = self.edges(source_id=node.id)
-        in_covers = self.covers(target_id=node.id)
-        out_covers = self.covers(source_id=node.id)
-        init = self.is_init(node.id)
-        target = self.is_target(node.id)
-        expanded = self.is_expanded(node.id)
-        in_expanded = {edge.source.id: self.is_expanded(edge.source.id) for edge in in_edges}
-        self.remove_node(node.id)
-
-        # Add the new, update data
-        new_node = self.create_node(new_cterm)
-        for in_edge in in_edges:
-            self.create_edge(in_edge.source.id, new_node.id, in_edge.depth)
-        for out_edge in out_edges:
-            self.create_edge(new_node.id, out_edge.target.id, out_edge.depth)
-        for in_cover in in_covers:
-            self.create_cover(in_cover.source.id, new_node.id, csubst=in_cover.csubst)
-        for out_cover in out_covers:
-            self.create_cover(new_node.id, out_cover.target.id, csubst=out_cover.csubst)
-        if init:
-            self.add_init(new_node.id)
-        if target:
-            self.add_target(new_node.id)
-        if expanded:
-            self.add_expanded(new_node.id)
-        for nid in in_expanded:
-            if in_expanded[nid]:
-                self.add_expanded(nid)
-
-        return new_node.id
+    def replace_node(self, node_id: NodeIdLike, cterm: CTerm) -> None:
+        term = cterm.kast
+        term = remove_source_attributes(term)
+        cterm = CTerm.from_kast(term)
+        node_id = self._resolve(node_id)
+        node = KCFG.Node(node_id, cterm)
+        self._nodes[node_id] = node
 
     def successors(
         self,
@@ -805,24 +768,6 @@ def is_stuck(self, node_id: NodeIdLike) -> bool:
         node_id = self._resolve(node_id)
         return self.is_expanded(node_id) and self.is_leaf(node_id)
 
-    def node_attrs(self, node_id: NodeIdLike) -> list[str]:
-        attrs = []
-        if self.is_init(node_id):
-            attrs.append('init')
-        if self.is_target(node_id):
-            attrs.append('target')
-        if self.is_expanded(node_id):
-            attrs.append('expanded')
-        if self.is_stuck(node_id):
-            attrs.append('stuck')
-        if self.is_frontier(node_id):
-            attrs.append('frontier')
-        if self.is_leaf(node_id):
-            attrs.append('leaf')
-        if self.is_split(node_id):
-            attrs.append('split')
-        return attrs
-
     def prune(self, node_id: NodeIdLike, keep_init: bool = True, keep_target: bool = True) -> list[int]:
         nodes = self.reachable_nodes(node_id)
         pruned_nodes = []
@@ -843,20 +788,6 @@ def shortest_path_between(
             return None
         return sorted(paths, key=(lambda path: path_length(path)))[0]
 
-    def path_constraints(self, final_node_id: NodeIdLike) -> KInner:
-        path = self.shortest_path_between(self.get_unique_init().id, final_node_id)
-        if path is None:
-            raise ValueError(f'No path found to specified node: {final_node_id}')
-        curr_constraint: KInner = mlTop()
-        for edge in reversed(path):
-            if type(edge) is KCFG.Split:
-                assert len(edge.targets) == 1
-                csubst = edge.splits[edge.targets[0].id]
-                curr_constraint = mlAnd([csubst.subst.ml_pred, csubst.constraint, curr_constraint])
-            if type(edge) is KCFG.Cover:
-                curr_constraint = mlAnd([edge.csubst.constraint, edge.csubst.subst.apply(curr_constraint)])
-        return mlAnd(flatten_label('#And', curr_constraint))
-
     def paths_between(self, source_id: NodeIdLike, target_id: NodeIdLike) -> list[tuple[Successor, ...]]:
         source_id = self._resolve(source_id)
         target_id = self._resolve(target_id)

pyk/src/pyk/kcfg/show.py

@@ -44,6 +44,37 @@ def __init__(
     ):
         self.kprint = kprint
 
+    @staticmethod
+    def node_attrs(kcfg: KCFG, node: KCFG.Node) -> list[str]:
+        attrs = []
+        if kcfg.is_init(node.id):
+            attrs.append('init')
+        if kcfg.is_target(node.id):
+            attrs.append('target')
+        if kcfg.is_expanded(node.id):
+            attrs.append('expanded')
+        if kcfg.is_stuck(node.id):
+            attrs.append('stuck')
+        if kcfg.is_frontier(node.id):
+            attrs.append('frontier')
+        if kcfg.is_leaf(node.id):
+            attrs.append('leaf')
+        if kcfg.is_split(node.id):
+            attrs.append('split')
+        return attrs
+
+    @staticmethod
+    def node_short_info(
+        kcfg: KCFG, node: KCFG.Node, node_printer: Callable[[CTerm], Iterable[str]] | None = None
+    ) -> list[str]:
+        attrs = KCFGShow.node_attrs(kcfg, node) + ['@' + alias for alias in sorted(kcfg.aliases(node.id))]
+        attr_string = ' (' + ', '.join(attrs) + ')' if attrs else ''
+        node_header = str(node.id) + attr_string
+        node_strs = [node_header]
+        if node_printer:
+            node_strs.extend(f' {nl}' for nl in node_printer(node.cterm))
+        return node_strs
+
     def pretty_segments(
         self,
         kcfg: KCFG,
@@ -67,7 +98,7 @@ def _green(text: str) -> str:
             return '\033[32m' + text + '\033[0m'
 
         def _print_node(node: KCFG.Node) -> list[str]:
-            short_info = kcfg.node_short_info(node, node_printer=node_printer)
+            short_info = KCFGShow.node_short_info(kcfg, node, node_printer=node_printer)
             if kcfg.is_frontier(node.id):
                 short_info[0] = _bold(short_info[0])
             return short_info
@@ -377,8 +408,8 @@ def _short_label(label: str) -> str:
         graph = Digraph()
 
         for node in kcfg.nodes:
-            label = '\n'.join(kcfg.node_short_info(node, node_printer=node_printer))
-            class_attrs = ' '.join(kcfg.node_attrs(node.id))
+            label = '\n'.join(KCFGShow.node_short_info(kcfg, node, node_printer=node_printer))
+            class_attrs = ' '.join(KCFGShow.node_attrs(kcfg, node))
             attrs = {'class': class_attrs} if class_attrs else {}
             graph.node(name=node.id, label=label, **attrs)
 

pyk/src/pyk/proof/proof.py

@@ -101,6 +101,10 @@ def read_proof(cls: type[Proof], id: str, proof_dir: Path) -> Proof:
 
         raise ValueError(f'Could not load Proof from file {id}: {proof_path}')
 
+    @property
+    def json(self) -> str:
+        return json.dumps(self.dict)
+
     @property
     def summary(self) -> Iterable[str]:
         return [