Add extensible prover implementation (#103)

This PR introduces a module `kprovex` that defines an extensible prover
based on `APRPRover`. It has the the following submodules:
* `kprovex.api`: the plugin API definition. A plugin provides the K
definition to the prover, as well as functions for loading and printing
proofs of that definition.
* `kprovex._default`: semantics-agnostic defaults for loading and
printing proofs.
* `kprovex._loader`: plugin loader.
* `kprovex._kprovex`: prover implementation.

Furthermore, it adds a simple plugin implementation for
`riscv-semantics`.

The advantage of this architecture is that `kprovex` can be upstreamed
to `pyk` without breaking it or the `riscv-semantics` prover.
(Naturally, imports still need to be adjusted.)

---------

Co-authored-by: devops <devops@runtimeverification.com>

Author: tothtamas28

package/version

@@ -1 +1 @@
-0.1.100
+0.1.101

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "kriscv"
-version = "0.1.100"
+version = "0.1.101"
 description = "K tooling for the RISC-V architecture"
 readme = "README.md"
 requires-python = "~=3.10"
@@ -28,6 +28,9 @@ kriscv-asm = "kriscv.devtools:kriscv_asm"
 [project.entry-points.kdist]
 riscv-semantics = "kriscv.kdist.plugin"
 
+[project.entry-points.kprovex]
+riscv = "kriscv.symtools:KRiscVPlugin"
+
 [dependency-groups]
 dev = [
     "autoflake",

src/kriscv/kprovex/__init__.py

@@ -0,0 +1 @@
+from ._kprovex import KProveX, create_prover

src/kriscv/kprovex/_default.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from pathlib import Path
+    from typing import Final
+
+    from pyk.kast import KInner
+    from pyk.proof.reachability import APRProof
+
+    from .api import Config, Init, Show
+
+
+def init_from_claims(config: Config, spec_file: Path, claim_id: str) -> APRProof:
+    from pyk.ktool.claim_loader import ClaimLoader
+    from pyk.proof.reachability import APRProof
+
+    spec_module, claim_label = claim_id.split('.', 1)
+    include_dirs = config.dist.source_dirs
+
+    claims = ClaimLoader(config.kprove).load_claims(
+        spec_file=spec_file,
+        spec_module_name=spec_module,
+        claim_labels=[claim_label],
+        include_dirs=include_dirs,
+    )
+    (claim,) = claims
+
+    proof = APRProof.from_claim(
+        config.kprove.definition,
+        claim=claim,
+        logs={},
+        proof_dir=config.proof_dir,
+    )
+    return proof
+
+
+def show_pretty_term(config: Config, term: KInner) -> str:
+    from pyk.konvert import kast_to_kore
+    from pyk.kore.tools import kore_print
+
+    kore = kast_to_kore(config.definition, term)
+    text = kore_print(kore, definition_dir=config.dist.haskell_dir)
+    return text
+
+
+# Check signatures
+_default_init: Final[Init] = init_from_claims
+_default_show: Final[Show] = show_pretty_term

src/kriscv/kprovex/_kprovex.py

@@ -0,0 +1,214 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from functools import cached_property
+from pathlib import Path
+from typing import TYPE_CHECKING, final
+
+from pyk.proof.reachability import APRProof, APRProver
+
+from .api import Config
+
+if TYPE_CHECKING:
+    from pyk.proof import ProofStatus
+    from pyk.proof.show import APRProofNodePrinter
+    from pyk.utils import BugReport
+
+    from .api import Init, Plugin, Show
+
+
+def create_prover(plugin_id: str, proof_dir: str | Path, *, bug_report: BugReport | None = None) -> KProveX:
+    from ._loader import PLUGINS
+
+    if plugin_id not in PLUGINS:
+        raise ValueError(f'Unknown plugin: {plugin_id}')
+
+    plugin = PLUGINS[plugin_id]
+    proof_dir = Path(proof_dir)
+
+    return KProveX(
+        plugin=plugin,
+        proof_dir=proof_dir,
+        bug_report=bug_report,
+    )
+
+
+@final
+@dataclass
+class KProveX:
+    plugin: Plugin
+    proof_dir: Path
+    bug_report: BugReport | None
+
+    def __init__(
+        self,
+        plugin: Plugin,
+        proof_dir: Path,
+        *,
+        bug_report: BugReport | None = None,
+    ):
+        self.plugin = plugin
+        self.proof_dir = proof_dir
+        self.bug_report = bug_report
+
+        proof_dir.mkdir(parents=True, exist_ok=True)
+
+    @cached_property
+    def config(self) -> Config:
+        return Config(
+            dist=self.plugin.dist(),
+            proof_dir=self.proof_dir,
+            bug_report=self.bug_report,
+        )
+
+    def init_proof(
+        self,
+        spec_file: str | Path,
+        claim_id: str,
+        *,
+        init_id: str | None = None,
+        exist_ok: bool = False,
+    ) -> str:
+        spec_file = Path(spec_file)
+        init = self._load_init(init_id=init_id)
+        proof = init(config=self.config, spec_file=spec_file, claim_id=claim_id)
+        if not exist_ok and APRProof.proof_data_exists(proof.id, self.proof_dir):
+            raise ValueError(f'Proof with id already exists: {proof.id}')
+
+        proof.write_proof_data()
+        return proof.id
+
+    def list_proofs(self) -> list[str]:
+        raise ValueError('TODO')
+
+    def list_nodes(self, proof_id: str) -> list[int]:
+        proof = self._load_proof(proof_id)
+        return [node.id for node in proof.kcfg.nodes]
+
+    def advance_proof(
+        self,
+        proof_id: str,
+        *,
+        max_depth: int | None = None,
+        max_iterations: int | None = None,
+    ) -> ProofStatus:
+        proof = self._load_proof(proof_id)
+
+        with self.config.explore(id=proof_id) as kcfg_explore:
+            prover = APRProver(
+                kcfg_explore=kcfg_explore,
+                execute_depth=max_depth,
+            )
+            prover.advance_proof(proof, max_iterations=max_iterations)
+
+        return proof.status
+
+    def show_proof(
+        self,
+        proof_id: str,
+        *,
+        show_id: str | None = None,
+        truncate: bool = False,
+    ) -> str:
+        from pyk.proof.show import APRProofShow
+
+        proof = self._load_proof(proof_id)
+        node_printer = self._proof_node_printer(proof, show_id=show_id, full_printer=False)
+        proof_show = APRProofShow(self.config.definition, node_printer=node_printer)
+        lines = proof_show.show(proof)
+        if truncate:
+            lines = [_truncate(line, 120) for line in lines]
+        return '\n'.join(lines)
+
+    def view_proof(
+        self,
+        proof_id: str,
+        *,
+        show_id: str | None = None,
+    ) -> None:
+        from pyk.proof.tui import APRProofViewer
+
+        proof = self._load_proof(proof_id)
+        node_printer = self._proof_node_printer(proof, show_id=show_id, full_printer=False)
+        viewer = APRProofViewer(proof, self.config.kprove, node_printer=node_printer)
+        viewer.run()
+
+    def prune_node(self, proof_id: str, node_id: str) -> list[int]:
+        proof = self._load_proof(proof_id)
+        res = proof.prune(node_id)
+        proof.write_proof_data()
+        return res
+
+    def show_node(
+        self,
+        proof_id: str,
+        node_id: str,
+        *,
+        show_id: str | None = None,
+        truncate: bool = False,
+    ) -> str:
+        proof = self._load_proof(proof_id)
+        node_printer = self._proof_node_printer(proof, show_id=show_id, full_printer=True)
+        kcfg = proof.kcfg
+        node = kcfg.node(node_id)
+        lines = node_printer.print_node(kcfg, node)
+        if truncate:
+            lines = [_truncate(line, 120) for line in lines]
+        return '\n'.join(lines)
+
+    # Private helpers
+
+    def _load_proof(self, proof_id: str) -> APRProof:
+        return APRProof.read_proof_data(proof_dir=self.proof_dir, id=proof_id)
+
+    def _load_init(self, *, init_id: str | None) -> Init:
+        if init_id is None:
+            from . import _default
+
+            return _default.init_from_claims
+
+        inits = self.plugin.inits()
+        if init_id not in inits:
+            raise ValueError(f'Unknown init function: {init_id}')
+
+        return inits[init_id]
+
+    def _load_show(self, *, show_id: str | None) -> Show:
+        if show_id is None:
+            from . import _default
+
+            return _default.show_pretty_term
+
+        shows = self.plugin.shows()
+        if show_id not in shows:
+            raise ValueError(f'Unknown show function: {show_id}')
+
+        return shows[show_id]
+
+    def _proof_node_printer(
+        self,
+        proof: APRProof,
+        *,
+        show_id: str | None = None,
+        full_printer: bool = False,
+    ) -> APRProofNodePrinter:
+        from pyk.cterm.show import CTermShow
+        from pyk.proof.show import APRProofNodePrinter
+
+        show = self._load_show(show_id=show_id)
+        printer = lambda kast: show(self.config, kast)
+        return APRProofNodePrinter(
+            proof=proof,
+            cterm_show=CTermShow(
+                printer=printer,
+                minimize=False,
+                break_cell_collections=False,
+            ),
+            full_printer=full_printer,
+        )
+
+
+def _truncate(line: str, n: int) -> str:
+    if len(line) <= n:
+        return line
+    return line[: n - 3] + '...'

src/kriscv/kprovex/_loader.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+import importlib
+import logging
+import re
+from typing import TYPE_CHECKING
+
+from pyk.utils import FrozenDict
+
+from .api import Plugin
+
+if TYPE_CHECKING:
+    from importlib.metadata import EntryPoint
+    from typing import Final
+
+
+_LOGGER: Final = logging.getLogger(__name__)
+
+
+def _load_plugins() -> FrozenDict[str, Plugin]:
+    entry_points = importlib.metadata.entry_points(group='kprovex')
+    plugins: FrozenDict[str, Plugin] = FrozenDict(
+        (entry_point.name, plugin) for entry_point in entry_points if (plugin := _load_plugin(entry_point)) is not None
+    )
+    return plugins
+
+
+def _load_plugin(entry_point: EntryPoint) -> Plugin | None:
+    if not _valid_id(entry_point.name):
+        _LOGGER.warning(f'Invalid entry point name, skipping: {entry_point.name}')
+        return None
+
+    _LOGGER.info(f'Loading entry point: {entry_point.name}')
+    try:
+        module_name, class_name = entry_point.value.split(':')
+    except ValueError:
+        _LOGGER.error(f'Invalid entry point value: {entry_point.value}', exc_info=True)
+        return None
+
+    try:
+        _LOGGER.info(f'Importing module: {module_name}')
+        module = importlib.import_module(module_name)
+    except Exception:
+        _LOGGER.error(f'Module {module_name} cannot be imported', exc_info=True)
+        return None
+
+    try:
+        _LOGGER.info(f'Loading plugin: {class_name}')
+        cls = getattr(module, class_name)
+    except AttributeError:
+        _LOGGER.error(f'Class {class_name} not found in module {module_name}', exc_info=True)
+        return None
+
+    if not issubclass(cls, Plugin):
+        _LOGGER.error(f'Class {class_name} is not a Plugin', exc_info=True)
+        return None
+
+    try:
+        _LOGGER.info(f'Instantiating plugin: {class_name}')
+        plugin = cls()
+    except TypeError:
+        _LOGGER.error(f'Cannot instantiate plugin {class_name}', exc_info=True)
+        return None
+
+    return plugin
+
+
+_ID_PATTERN = re.compile('[a-z0-9]+(-[a-z0-9]+)*')
+
+
+def _valid_id(s: str) -> bool:
+    return _ID_PATTERN.fullmatch(s) is not None
+
+
+PLUGINS: Final = _load_plugins()