pyk: booster-only recover mode for the APR prover, isolating and recording Kore handoffs

pyk/src/pyk/cli/pyk.py

@@ -319,6 +319,7 @@ class ProveOptions(LoggingOptions, SpecOptions, SaveDirOptions):
     assume_defined: bool
     show_kcfg: bool
     haskell_logging: bool
+    booster_recover_mode: bool
 
     @staticmethod
     def default() -> dict[str, Any]:
@@ -332,6 +333,7 @@ def default() -> dict[str, Any]:
             'assume_defined': False,
             'show_kcfg': False,
             'haskell_logging': False,
+            'booster_recover_mode': False,
         }
 
     @staticmethod
@@ -533,6 +535,17 @@ def create_argument_parser() -> ArgumentParser:
             ' Requires --save-directory.'
         ),
     )
+    prove_args.add_argument(
+        '--booster-recover-mode',
+        dest='booster_recover_mode',
+        default=None,
+        action='store_true',
+        help=(
+            'Run the reachability prover booster-only, escalating to Kore (re-simplification then'
+            ' the Kore operation) only where booster cannot make progress, and record in the KCFG'
+            ' exactly where Kore did work (variants + kore_handoffs) for diagnostic attribution.'
+        ),
+    )
 
     show_args = pyk_args_command.add_parser(
         'show',

pyk/src/pyk/cterm/symbolic.py

@@ -14,6 +14,7 @@
 from ..kast.pretty import PrettyPrinter
 from ..konvert import kast_to_kore, kflatmodule_to_kore, kore_to_kast
 from ..kore.rpc import (
+    AbortedError,
     AbortedResult,
     KoreClient,
     KoreExecLogFormat,
@@ -78,6 +79,9 @@ class CTermImplies(NamedTuple):
     failing_cells: tuple[tuple[str, KInner], ...]
     remaining_implication: KInner | None
     logs: tuple[LogEntry, ...]
+    # True when the backend aborted the implies (it could not decide it) — a couldn't-determine,
+    # not a decisive verdict.  Recover-mode escalates on it instead of trusting `csubst is None`.
+    indeterminate: bool = False
 
 
 @final
@@ -153,6 +157,16 @@ def _capture_haskell_log(self, entries: tuple[Any, ...] | None) -> None:
         log_file = self._haskell_log_dir / f'{request_id}.jsonl'
         log_file.write_text('\n'.join(json.dumps(entry) for entry in entries) + '\n')
 
+    @property
+    def last_request_id(self) -> str | None:
+        """JSON-RPC id of the most recent RPC issued by the backing client (see `KoreClient.last_request_id`).
+
+        Recover-mode snapshots this immediately after a simplify to key the `KoreHandoff` to that
+        request, so a handoff can be correlated with the per-request log file that the configured
+        `haskell_log_dir` writer (see `_capture_haskell_log`) emits for the same id.
+        """
+        return self._kore_client.last_request_id
+
     def execute(
         self,
         cterm: CTerm,
@@ -162,6 +176,7 @@ def execute(
         module_name: str | None = None,
         booster_only_simplify: bool | None = None,
         haskell_logging: bool | None = None,
+        raise_on_aborted: bool = True,
     ) -> CTermExecute:
 
         _LOGGER.debug(f'Executing: {cterm}')
@@ -184,7 +199,12 @@ def execute(
             raise self._smt_solver_error(err) from err
         self._capture_haskell_log(response.haskell_log_entries)
 
-        if isinstance(response, AbortedResult):
+        # Under booster-only execution, an `AbortedResult` is the expected signal that the booster
+        # could make no further progress (it is what drives the recover-mode ladder).  Callers that
+        # expect aborts pass `raise_on_aborted=False`; the `AbortedResult.reason`/`next_states`
+        # class attributes let the rest of this body proceed unchanged, and a no-progress abort
+        # surfaces as a depth-0 result.
+        if raise_on_aborted and 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}')
 
@@ -316,6 +336,14 @@ def implies(
             )
         except SmtSolverError as err:
             raise self._smt_solver_error(err) from err
+        except AbortedError as err:
+            # The backend aborted because it could not decide the implication (e.g. constraints
+            # the engine cannot discharge).  Surface it as an indeterminate, not-subsumed result,
+            # so callers (notably recover-mode's TRY_KORE escalation, which calls kore-implies
+            # directly and so sees the raw abort the proxy would otherwise absorb) treat it as
+            # "unknown" instead of crashing.
+            _LOGGER.warning(f'implies aborted, treating as indeterminate: {err.data}')
+            return CTermImplies(None, (), None, (), indeterminate=True)
         self._capture_haskell_log(result.haskell_log_entries)
 
         if not result.valid:

pyk/src/pyk/kcfg/explore.py

@@ -2,7 +2,7 @@
 
 import logging
 from functools import cached_property
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, NamedTuple
 
 from ..kast.inner import KApply, KVariable
 from ..kast.manip import (
@@ -17,7 +17,7 @@
 from ..kast.pretty import PrettyPrinter
 from ..kore.rpc import LogRewrite, RewriteSuccess
 from ..utils import not_none, shorten_hashes, single, unique
-from .kcfg import KCFG, Abstract, Branch, NDBranch, Step, Stuck, Vacuous
+from .kcfg import KCFG, Abstract, Branch, NDBranch, NoProgress, Producer, Step, Vacuous
 from .semantics import DefaultSemantics
 
 if TYPE_CHECKING:
@@ -35,6 +35,20 @@
 _LOGGER: Final = logging.getLogger(__name__)
 
 
+class SimplifyVariant(NamedTuple):
+    """Result of a recover-mode variant-producing simplification.
+
+    Carries what the coordinator needs to land the variant (`KCFG.add_variant`) and key its
+    `KoreHandoff`: the `producer`, the simplified `cterm`, and the simplify RPC's `request_id`.
+    The per-request haskell log bundle, when wanted, is written by the configured `haskell_log_dir`
+    (see `CTermSymbolic._capture_haskell_log`) under the same `request_id`.
+    """
+
+    producer: Producer
+    cterm: CTerm
+    request_id: str | None
+
+
 class KCFGExplore:
     cterm_symbolic: CTermSymbolic
 
@@ -136,6 +150,22 @@ def simplify(self, cfg: KCFG, logs: dict[int, tuple[LogEntry, ...]]) -> None:
                 else:
                     logs[node.id] = next_node_logs
 
+    def simplify_variant(self, cterm: CTerm, *, booster_only: bool) -> SimplifyVariant:
+        """Re-simplify ``cterm`` with one backend, capturing the request id of the call.
+
+        Non-destructive: returns a `SimplifyVariant` for the coordinator to land via
+        `KCFG.add_variant`; the recover-mode ladder uses `booster_only=True` (rung 0→1) then
+        `booster_only=False` (rung 1→2).  Per-request haskell logging is governed by the
+        `CTermSymbolic`'s configured `haskell_log_dir`, not forced here.
+        """
+        producer = Producer.BOOSTER_SIMPLIFY if booster_only else Producer.KORE_SIMPLIFY
+        simplified, _logs = self.cterm_symbolic.simplify(cterm, booster_only_simplify=booster_only)
+        return SimplifyVariant(
+            producer=producer,
+            cterm=simplified,
+            request_id=self.cterm_symbolic.last_request_id,
+        )
+
     def step(
         self,
         cfg: KCFG,
@@ -218,6 +248,8 @@ def extend_cterm(
         cut_point_rules: Iterable[str] = (),
         terminal_rules: Iterable[str] = (),
         module_name: str | None = None,
+        booster_only_simplify: bool | None = None,
+        raise_on_aborted: bool = True,
         haskell_logging: bool | None = None,
     ) -> list[KCFGExtendResult]:
 
@@ -229,27 +261,36 @@ def extend_cterm(
         if _cterm != abstract_cterm:
             return [Abstract(abstract_cterm)]
 
-        cterm, next_states, depth, vacuous, next_node_logs = self.cterm_symbolic.execute(
+        exec_result = self.cterm_symbolic.execute(
             _cterm,
             depth=execute_depth,
             cut_point_rules=cut_point_rules,
             terminal_rules=terminal_rules,
             module_name=module_name,
+            booster_only_simplify=booster_only_simplify,
+            raise_on_aborted=raise_on_aborted,
             haskell_logging=haskell_logging,
         )
+        cterm = exec_result.state
+        next_states = exec_result.next_states
+        depth = exec_result.depth
+        vacuous = exec_result.vacuous
+        next_node_logs = exec_result.logs
 
         extend_results: list[KCFGExtendResult] = []
 
         # Basic block
         if depth > 0:
             extend_results.append(Step(cterm, depth, next_node_logs, self._extract_rule_labels(next_node_logs)))
 
-        # Stuck or vacuous
+        # No-progress or vacuous.  `Vacuous` is a context-free semantic verdict and stays here; a
+        # depth-0 no-progress result is reported as the neutral `NoProgress` rather than `Stuck`, so
+        # the coordinator (`APRProof.commit`) decides whether the node is terminal.
         if not next_states:
             if vacuous:
                 extend_results.append(Vacuous())
             elif depth == 0:
-                extend_results.append(Stuck())
+                extend_results.append(NoProgress())
         # Cut rule
         elif len(next_states) == 1:
             if not self.kcfg_semantics.can_make_custom_step(cterm):