Cut on circularity rule application, better error message (#4445)

Related: #4359
Related:
runtimeverification/haskell-backend#3912

This PR is to support passing the final tests on the RPC prover in KEVM.
In particular:

- A bug we've been hitting in Haskell backend gets a better error
message, to hopefully be more informative off the bat.
- A duplicate log message in `KCFGExplore.section_edge` is removed.
- When doing `KCFGExlore.section_edge`, we were previously just dropping
the rule labels of the applied rules on the new edges. Now we save them
and store them appropriately on the KCFG.
- When computing the rule labels to store in the KCFG, if the unique-id
is not present in teh definition, we still use the supplied unique id
because the HB now returns the rule label instead if it's present
(runtimeverification/haskell-backend#3916). We
just report the location of the rule as `UNKNOWN`.
- The logic for how we apply circularity rules is refined a bit:
- When discharging a circularity proof obligation, we always consider
the circularity rule itself to be a cut-point rule. That's because we
need the state immediately following the application of the circularity
rule to compare it to subsumption into the initial target state.
- When discharging a circularity proof obligation, if we have not made
any progress yet on the proof, we first take an execution step of depth
1 before resuming normal execute depth. This is because we cannot inject
the circularity rule for consideration until we've made some progress,
but want to enable it for consideration as early as possible.
- A test of IMP that before had to manually massage the cut-point-rules
list does not have to do that anymore, and is passing directly.

---------

Co-authored-by: Petar Maksimović <[email protected]>

pyk/src/pyk/kcfg/explore.py

@@ -59,6 +59,19 @@ def _pretty_printer(self) -> PrettyPrinter:
     def pretty_print(self, kinner: KInner) -> str:
         return self._pretty_printer.print(kinner)
 
+    def _extract_rule_labels(self, _logs: tuple[LogEntry, ...]) -> list[str]:
+        _rule_lines = []
+        for node_log in _logs:
+            if isinstance(node_log, LogRewrite) and isinstance(node_log.result, RewriteSuccess):
+                if node_log.result.rule_id in self.cterm_symbolic._definition.sentence_by_unique_id:
+                    sent = self.cterm_symbolic._definition.sentence_by_unique_id[node_log.result.rule_id]
+                    _rule_lines.append(f'{sent.label}:{sent.source}')
+                else:
+                    if node_log.result.rule_id == 'UNKNOWN':
+                        _LOGGER.warning(f'Unknown unique id attached to rule log entry: {node_log}')
+                    _rule_lines.append(f'{node_log.result.rule_id}:UNKNOWN')
+        return _rule_lines
+
     def implication_failure_reason(self, antecedent: CTerm, consequent: CTerm) -> tuple[bool, str]:
         def _is_cell_subst(csubst: KInner) -> bool:
             if type(csubst) is KApply and csubst.label.name == '_==K_':
@@ -143,17 +156,18 @@ def step(
         _LOGGER.info(f'Found new node at depth {depth} {self.id}: {shorten_hashes((node.id, new_node.id))}')
         logs[new_node.id] = exec_res.logs
         out_edges = cfg.edges(source_id=node.id)
+        rule_logs = self._extract_rule_labels(exec_res.logs)
         if len(out_edges) == 0:
-            cfg.create_edge(node.id, new_node.id, depth=depth)
+            cfg.create_edge(node.id, new_node.id, depth=depth, rules=rule_logs)
         else:
             edge = out_edges[0]
             if depth > edge.depth:
                 raise ValueError(
                     f'Step depth {depth} greater than original edge depth {edge.depth} {self.id}: {shorten_hashes((edge.source.id, edge.target.id))}'
                 )
             cfg.remove_edge(edge.source.id, edge.target.id)
-            cfg.create_edge(edge.source.id, new_node.id, depth=depth)
-            cfg.create_edge(new_node.id, edge.target.id, depth=(edge.depth - depth))
+            cfg.create_edge(edge.source.id, new_node.id, depth=depth, rules=rule_logs)
+            cfg.create_edge(new_node.id, edge.target.id, depth=(edge.depth - depth), rules=edge.rules[depth:])
         return new_node.id
 
     def section_edge(
@@ -175,7 +189,6 @@ def section_edge(
         new_nodes = []
         curr_node_id = edge.source.id
         while new_depth < orig_depth:
-            _LOGGER.info(f'Taking {section_depth} steps from node {self.id}: {shorten_hashes(curr_node_id)}')
             curr_node_id = self.step(cfg, curr_node_id, logs, depth=section_depth)
             new_nodes.append(curr_node_id)
             new_depth += section_depth
@@ -205,18 +218,6 @@ def extend_cterm(
         def log(message: str, *, warning: bool = False) -> None:
             _LOGGER.log(logging.WARNING if warning else logging.INFO, f'Extend result for {self.id}: {message}')
 
-        def extract_rule_labels(_logs: tuple[LogEntry, ...]) -> list[str]:
-            _rule_lines = []
-            for node_log in _logs:
-                if isinstance(node_log, LogRewrite) and isinstance(node_log.result, RewriteSuccess):
-                    if node_log.result.rule_id in self.cterm_symbolic._definition.sentence_by_unique_id:
-                        sent = self.cterm_symbolic._definition.sentence_by_unique_id[node_log.result.rule_id]
-                        _rule_lines.append(f'{sent.label}:{sent.source}')
-                    else:
-                        _LOGGER.warning(f'Unknown unique id attached to rule log entry: {node_log}')
-                        _rule_lines.append('UNKNOWN')
-            return _rule_lines
-
         custom_step_result = self.kcfg_semantics.custom_step(_cterm)
         if custom_step_result is not None:
             log(f'custom step node: {node_id}')
@@ -238,7 +239,7 @@ def extract_rule_labels(_logs: tuple[LogEntry, ...]) -> list[str]:
         # Basic block
         if depth > 0:
             log(f'basic block at depth {depth}: {node_id}')
-            return Step(cterm, depth, next_node_logs, extract_rule_labels(next_node_logs))
+            return Step(cterm, depth, next_node_logs, self._extract_rule_labels(next_node_logs))
 
         # Stuck or vacuous
         if not next_states:
@@ -255,7 +256,7 @@ def extract_rule_labels(_logs: tuple[LogEntry, ...]) -> list[str]:
                 next_states[0].state,
                 1,
                 next_node_logs,
-                extract_rule_labels(next_node_logs),
+                self._extract_rule_labels(next_node_logs),
                 cut=True,
             )
 
@@ -283,4 +284,4 @@ def extract_rule_labels(_logs: tuple[LogEntry, ...]) -> list[str]:
             # 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, extract_rule_labels(next_node_logs))
+            return NDBranch(next_cterms, next_node_logs, self._extract_rule_labels(next_node_logs))

pyk/src/pyk/kore/rpc.py

@@ -519,7 +519,7 @@ def extract_entry(pattern: Pattern) -> tuple[EVar, Pattern]:
         for conjunct in manip.conjuncts(pattern):
             key, value = extract_entry(conjunct)
             if key in res:
-                raise ValueError(f'Duplicate substitution entry key: {key.text}')
+                raise ValueError(f'Duplicate substitution entry key: {key.text} -> {[res[key].text, value.text]}')
             res[key] = value
         return res
 

pyk/src/pyk/proof/reachability.py

@@ -70,6 +70,9 @@ class APRProofStep:
     module_name: str
     shortest_path_to_node: tuple[KCFG.Node, ...]
     prior_loops_cache: FrozenDict[int, tuple[int, ...]] = field(compare=False)
+    circularity: bool
+    nonzero_depth: bool
+    circularity_rule_id: str
 
 
 class APRProof(Proof[APRProofStep, APRProofResult], KCFGExploration):
@@ -162,10 +165,7 @@ def get_steps(self) -> list[APRProofStep]:
                     else:
                         shortest_path.append(succ.source)
 
-            def nonzero_depth(proof: APRProof, node: KCFG.Node) -> bool:
-                return not proof.kcfg.zero_depth_between(proof.init, node.id)
-
-            module_name = self.circularities_module_name if nonzero_depth(self, node) else self.dependencies_module_name
+            module_name = self.circularities_module_name if self.nonzero_depth(node) else self.dependencies_module_name
 
             steps.append(
                 APRProofStep(
@@ -176,6 +176,9 @@ def nonzero_depth(proof: APRProof, node: KCFG.Node) -> bool:
                     target=self.kcfg.node(self.target),
                     shortest_path_to_node=tuple(shortest_path),
                     prior_loops_cache=FrozenDict(self.prior_loops_cache),
+                    circularity=self.circularity,
+                    nonzero_depth=self.nonzero_depth(node),
+                    circularity_rule_id=f'{self.rule_id}-{self.init}-TO-{self.target}',
                 )
             )
         return steps
@@ -193,6 +196,13 @@ def commit(self, result: APRProofResult) -> None:
         else:
             raise ValueError(f'Incorrect result type, expected APRProofResult: {result}')
 
+    def nonzero_depth(self, node: KCFG.Node) -> bool:
+        return not self.kcfg.zero_depth_between(self.init, node.id)
+
+    @property
+    def rule_id(self) -> str:
+        return f'APRPROOF-{self.id.upper()}'
+
     @property
     def module_name(self) -> str:
         return self._make_module_name(self.id)
@@ -384,18 +394,22 @@ def from_claim(
         )
 
     def as_rules(self, priority: int = 20, direct_rule: bool = False) -> list[KRule]:
-        if (self.passed and direct_rule) or (self.admitted and not self.kcfg.predecessors(self.target)):
+        if (
+            self.circularity
+            or (self.passed and direct_rule)
+            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(f'APRPROOF-{self.id.upper()}-BASIC-BLOCK', priority=priority)
+            _rule = _edge.to_rule(self.rule_id, priority=priority)
             assert type(_rule) is KRule
             _rules.append(_rule)
         return _rules
 
     def as_rule(self, priority: int = 20) -> KRule:
         _edge = KCFG.Edge(self.kcfg.node(self.init), self.kcfg.node(self.target), depth=0, rules=())
-        _rule = _edge.to_rule(f'APRPROOF-{self.id.upper()}-BASIC-BLOCK', priority=priority)
+        _rule = _edge.to_rule(self.rule_id, priority=priority)
         assert type(_rule) is KRule
         return _rule
 
@@ -791,10 +805,20 @@ def step_proof(self, step: APRProofStep) -> list[APRProofResult]:
         if is_terminal:
             return terminal_result
 
+        # Ensure that we cut at applications of circularity, so that subsumption into target state will be checked
+        cut_rules = list(self.cut_point_rules)
+        if step.circularity and step.nonzero_depth:
+            cut_rules.append(step.circularity_rule_id)
+
+        # Ensure that we record progress ASAP for circularities, so the circularity rule will be included for execution as soon as possible
+        execute_depth = self.execute_depth
+        if step.circularity and not step.nonzero_depth and (execute_depth is None or execute_depth > 1):
+            execute_depth = 1
+
         extend_result = self.kcfg_explore.extend_cterm(
             step.node.cterm,
-            execute_depth=self.execute_depth,
-            cut_point_rules=self.cut_point_rules,
+            execute_depth=execute_depth,
+            cut_point_rules=cut_rules,
             terminal_rules=self.terminal_rules,
             module_name=step.module_name,
             node_id=step.node.id,

pyk/src/tests/integration/proof/test_imp.py

@@ -508,7 +508,7 @@ def custom_step(self, c: CTerm) -> KCFGExtendResult | None:
         'sum-loop',
         None,
         None,
-        ['IMP.while'],  # If we do not include `IMP.while` in this list, we get 4 branches instead of 2
+        [],
         True,
         ProofStatus.PASSED,
         2,