Add extensible prover implementation

tothtamas28 · tothtamas28 · commit 1cd62437786a · 2025-05-21T12:52:23.000Z
diff --git a/pyproject.toml b/pyproject.toml
@@ -17,6 +17,9 @@ kriscv-asm = "kriscv.devtools:kriscv_asm"
 [tool.poetry.plugins.kdist]
 riscv-semantics = "kriscv.kdist.plugin"
 
+[tool.poetry.plugins.kprovex]
+riscv = "kriscv.symtools:KRiscVPlugin"
+
 [tool.poetry.dependencies]
 python = "^3.10"
 kframework = "7.1.257"
diff --git a/src/kriscv/kprovex/__init__.py b/src/kriscv/kprovex/__init__.py
@@ -0,0 +1 @@
+from ._kprovex import KProveX, create_prover
diff --git a/src/kriscv/kprovex/_default.py b/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
diff --git a/src/kriscv/kprovex/_kprovex.py b/src/kriscv/kprovex/_kprovex.py
@@ -0,0 +1,227 @@
+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.kcfg import KCFG
+    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=True)
+        proof_show = APRProofShow(self.config.kprove, 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,
+        minimize: bool = False,
+    ) -> APRProofNodePrinter:
+        from pyk.kast.manip import minimize_term
+        from pyk.proof.show import APRProofNodePrinter
+
+        show = self._load_show(show_id=show_id)
+        config = self.config
+
+        class _NodePrinter(APRProofNodePrinter):
+            def print_node(self, kcfg: KCFG, node: KCFG.Node) -> list[str]:
+                attrs = self.node_attrs(kcfg, node)
+                attr_str = ' (' + ', '.join(attrs) + ')' if attrs else ''
+                node_strs = [f'{node.id}{attr_str}']
+                if self.full_printer:
+                    kast = node.cterm.kast
+                    if self.minimize:
+                        kast = minimize_term(kast)
+                    show_res = show(config, kast)
+                    node_strs.extend('  ' + line for line in show_res.split('\n'))
+                return node_strs
+
+        return _NodePrinter(
+            proof=proof,
+            kprint=None,  # type: ignore [arg-type]
+            full_printer=full_printer,
+            minimize=minimize,
+        )
+
+
+def _truncate(line: str, n: int) -> str:
+    if len(line) <= n:
+        return line
+    return line[: n - 3] + '...'
diff --git a/src/kriscv/kprovex/_loader.py b/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()
diff --git a/src/kriscv/kprovex/api.py b/src/kriscv/kprovex/api.py
diff --git a/src/kriscv/symtools.py b/src/kriscv/symtools.py

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+from ._kprovex import KProveX, create_prover`