Encapsulate program-level options

package/version

@@ -1 +1 @@
-0.1.33
+0.1.34

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "kontrol"
-version = "0.1.33"
+version = "0.1.34"
 description = "Foundry integration for KEVM"
 authors = [
     "Runtime Verification, Inc. <contact@runtimeverification.com>",

src/kontrol/__init__.py

@@ -5,4 +5,4 @@
 if TYPE_CHECKING:
     from typing import Final
 
-VERSION: Final = '0.1.33'
+VERSION: Final = '0.1.34'

src/kontrol/__main__.py

@@ -28,6 +28,7 @@
     foundry_to_dot,
 )
 from .kompile import foundry_kompile
+from .options import ProveOptions
 from .prove import foundry_prove
 from .solc_to_k import solc_compile, solc_to_k
 
@@ -160,6 +161,7 @@ def exec_prove(
     trace_rewrites: bool = False,
     auto_abstract_gas: bool = False,
     run_constructor: bool = False,
+    fail_fast: bool = False,
     **kwargs: Any,
 ) -> None:
     _ignore_arg(kwargs, 'main_module', f'--main-module: {kwargs["main_module"]}')
@@ -175,26 +177,30 @@ def exec_prove(
     if isinstance(kore_rpc_command, str):
         kore_rpc_command = kore_rpc_command.split()
 
-    results = foundry_prove(
-        foundry_root=foundry_root,
-        max_depth=max_depth,
-        max_iterations=max_iterations,
+    options = ProveOptions(
+        auto_abstract_gas=auto_abstract_gas,
         reinit=reinit,
-        tests=tests,
-        workers=workers,
-        break_every_step=break_every_step,
-        break_on_jumpi=break_on_jumpi,
-        break_on_calls=break_on_calls,
-        bmc_depth=bmc_depth,
         bug_report=bug_report,
         kore_rpc_command=kore_rpc_command,
-        use_booster=use_booster,
-        counterexample_info=counterexample_info,
         smt_timeout=smt_timeout,
         smt_retry_limit=smt_retry_limit,
         trace_rewrites=trace_rewrites,
-        auto_abstract_gas=auto_abstract_gas,
+        bmc_depth=bmc_depth,
+        max_depth=max_depth,
+        break_every_step=break_every_step,
+        break_on_jumpi=break_on_jumpi,
+        break_on_calls=break_on_calls,
+        workers=workers,
+        counterexample_info=counterexample_info,
+        max_iterations=max_iterations,
         run_constructor=run_constructor,
+        fail_fast=fail_fast,
+    )
+
+    results = foundry_prove(
+        foundry_root=foundry_root,
+        options=options,
+        tests=tests,
     )
     failed = 0
     for pid, r in results.items():
@@ -526,6 +532,13 @@ def _parse_test_version_tuple(value: str) -> tuple[str, int | None]:
         action='store_true',
         help='Include the contract constructor in the test execution.',
     )
+    prove_args.add_argument(
+        '--fail-fast',
+        dest='fail_fast',
+        default=False,
+        action='store_true',
+        help='Stop execution on other branches if a failing node is detected.',
+    )
 
     show_args = command_parser.add_parser(
         'show',

src/kontrol/options.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
+    from pyk.utils import BugReport
+
+
+@dataclass(frozen=True)
+class ProveOptions:
+    auto_abstract_gas: bool
+    bug_report: BugReport | None
+    use_booster: bool
+    kore_rpc_command: tuple[str, ...]
+    smt_timeout: int | None
+    smt_retry_limit: int | None
+    trace_rewrites: bool
+    simplify_init: bool
+    bmc_depth: int | None
+    max_depth: int
+    break_every_step: bool
+    break_on_jumpi: bool
+    break_on_calls: bool
+    workers: int
+    counterexample_info: bool
+    max_iterations: int | None
+    run_constructor: bool
+    fail_fast: bool
+    reinit: bool
+    port: int | None
+
+    def __init__(
+        self,
+        auto_abstract_gas: bool = False,
+        bug_report: BugReport | None = None,
+        use_booster: bool = True,
+        kore_rpc_command: str | Iterable[str] | None = None,
+        smt_timeout: int | None = None,
+        smt_retry_limit: int | None = None,
+        trace_rewrites: bool = False,
+        bmc_depth: int | None = None,
+        max_depth: int = 1000,
+        break_every_step: bool = False,
+        break_on_jumpi: bool = False,
+        break_on_calls: bool = True,
+        workers: int = 1,
+        counterexample_info: bool = False,
+        max_iterations: int | None = None,
+        run_constructor: bool = False,
+        fail_fast: bool = True,
+        reinit: bool = False,
+        port: int | None = None,
+    ) -> None:
+        if kore_rpc_command is None:
+            kore_rpc_command = ('kore-rpc-booster',) if use_booster else ('kore-rpc',)
+        elif isinstance(kore_rpc_command, str):
+            kore_rpc_command = (kore_rpc_command,)
+        else:
+            kore_rpc_command = tuple(kore_rpc_command)
+        object.__setattr__(self, 'auto_abstract_gas', auto_abstract_gas)
+        object.__setattr__(self, 'bug_report', bug_report)
+        object.__setattr__(self, 'use_booster', use_booster)
+        object.__setattr__(self, 'kore_rpc_command', kore_rpc_command)
+        object.__setattr__(self, 'smt_timeout', smt_timeout)
+        object.__setattr__(self, 'smt_retry_limit', smt_retry_limit)
+        object.__setattr__(self, 'trace_rewrites', trace_rewrites)
+        object.__setattr__(self, 'bmc_depth', bmc_depth)
+        object.__setattr__(self, 'max_depth', max_depth)
+        object.__setattr__(self, 'break_every_step', break_every_step)
+        object.__setattr__(self, 'break_on_jumpi', break_on_jumpi)
+        object.__setattr__(self, 'break_on_calls', break_on_calls)
+        object.__setattr__(self, 'workers', workers)
+        object.__setattr__(self, 'counterexample_info', counterexample_info)
+        object.__setattr__(self, 'max_iterations', max_iterations)
+        object.__setattr__(self, 'run_constructor', run_constructor)
+        object.__setattr__(self, 'fail_fast', fail_fast)
+        object.__setattr__(self, 'reinit', reinit)
+        object.__setattr__(self, 'port', port)

src/kontrol/prove.py

@@ -35,62 +35,44 @@
 
     from pyk.kast.inner import KInner
     from pyk.kcfg import KCFGExplore
-    from pyk.utils import BugReport
+
+    from .options import ProveOptions
 
 
 _LOGGER: Final = logging.getLogger(__name__)
 
 
 def foundry_prove(
     foundry_root: Path,
-    max_depth: int = 1000,
-    max_iterations: int | None = None,
-    reinit: bool = False,
+    options: ProveOptions,
     tests: Iterable[tuple[str, int | None]] = (),
-    workers: int = 1,
-    break_every_step: bool = False,
-    break_on_jumpi: bool = False,
-    break_on_calls: bool = True,
-    bmc_depth: int | None = None,
-    bug_report: BugReport | None = None,
-    kore_rpc_command: str | Iterable[str] | None = None,
-    use_booster: bool = False,
-    smt_timeout: int | None = None,
-    smt_retry_limit: int | None = None,
-    failure_info: bool = True,
-    counterexample_info: bool = False,
-    trace_rewrites: bool = False,
-    auto_abstract_gas: bool = False,
-    port: int | None = None,
-    run_constructor: bool = False,
 ) -> dict[tuple[str, int], tuple[bool, list[str] | None]]:
-    if workers <= 0:
-        raise ValueError(f'Must have at least one worker, found: --workers {workers}')
-    if max_iterations is not None and max_iterations < 0:
-        raise ValueError(f'Must have a non-negative number of iterations, found: --max-iterations {max_iterations}')
+    if options.workers <= 0:
+        raise ValueError(f'Must have at least one worker, found: --workers {options.workers}')
+    if options.max_iterations is not None and options.max_iterations < 0:
+        raise ValueError(
+            f'Must have a non-negative number of iterations, found: --max-iterations {options.max_iterations}'
+        )
 
-    if use_booster:
+    if options.use_booster:
         try:
             run_process(('which', 'kore-rpc-booster'), pipe_stderr=True).stdout.strip()
         except CalledProcessError:
             raise RuntimeError(
                 "Couldn't locate the kore-rpc-booster RPC binary. Please put 'kore-rpc-booster' on PATH manually or using kup install/kup shell."
             ) from None
 
-    if kore_rpc_command is None:
-        kore_rpc_command = ('kore-rpc-booster',) if use_booster else ('kore-rpc',)
-
-    foundry = Foundry(foundry_root, bug_report=bug_report)
+    foundry = Foundry(foundry_root, bug_report=options.bug_report)
     foundry.mk_proofs_dir()
 
-    test_suite = collect_tests(foundry, tests, reinit=reinit)
+    test_suite = collect_tests(foundry, tests, reinit=options.reinit)
     test_names = [test.name for test in test_suite]
 
     contracts = [test.contract for test in test_suite]
-    setup_method_tests = collect_setup_methods(foundry, contracts, reinit=reinit)
+    setup_method_tests = collect_setup_methods(foundry, contracts, reinit=options.reinit)
     setup_method_names = [test.name for test in setup_method_tests]
 
-    constructor_tests = collect_constructors(foundry, contracts, reinit=reinit)
+    constructor_tests = collect_constructors(foundry, contracts, reinit=options.reinit)
     constructor_names = [test.name for test in constructor_tests]
 
     _LOGGER.info(f'Running tests: {test_names}')
@@ -109,28 +91,12 @@ def foundry_prove(
 
     def run_prover(test_suite: list[FoundryTest]) -> dict[tuple[str, int], tuple[bool, list[str] | None]]:
         return _run_cfg_group(
-            test_suite,
-            foundry,
-            max_depth=max_depth,
-            max_iterations=max_iterations,
-            workers=workers,
-            break_every_step=break_every_step,
-            break_on_jumpi=break_on_jumpi,
-            break_on_calls=break_on_calls,
-            bmc_depth=bmc_depth,
-            bug_report=bug_report,
-            kore_rpc_command=kore_rpc_command,
-            use_booster=use_booster,
-            smt_timeout=smt_timeout,
-            smt_retry_limit=smt_retry_limit,
-            counterexample_info=counterexample_info,
-            trace_rewrites=trace_rewrites,
-            auto_abstract_gas=auto_abstract_gas,
-            port=port,
-            run_constructor=run_constructor,
+            tests=test_suite,
+            foundry=foundry,
+            options=options,
         )
 
-    if run_constructor:
+    if options.run_constructor:
         _LOGGER.info(f'Running initialization code for contracts in parallel: {constructor_names}')
         results = run_prover(constructor_tests)
         failed = [init_cfg for init_cfg, passed in results.items() if not passed]
@@ -215,68 +181,50 @@ def collect_constructors(foundry: Foundry, contracts: Iterable[Contract] = (), *
 def _run_cfg_group(
     tests: list[FoundryTest],
     foundry: Foundry,
-    *,
-    max_depth: int,
-    max_iterations: int | None,
-    workers: int,
-    break_every_step: bool,
-    break_on_jumpi: bool,
-    break_on_calls: bool,
-    bmc_depth: int | None,
-    bug_report: BugReport | None,
-    kore_rpc_command: str | Iterable[str] | None,
-    use_booster: bool,
-    smt_timeout: int | None,
-    smt_retry_limit: int | None,
-    counterexample_info: bool,
-    trace_rewrites: bool,
-    auto_abstract_gas: bool,
-    port: int | None,
-    run_constructor: bool = False,
+    options: ProveOptions,
 ) -> dict[tuple[str, int], tuple[bool, list[str] | None]]:
     def init_and_run_proof(test: FoundryTest) -> tuple[bool, list[str] | None]:
-        llvm_definition_dir = foundry.llvm_library if use_booster else None
-        start_server = port is None
+        start_server = options.port is None
 
         with legacy_explore(
             foundry.kevm,
-            kcfg_semantics=KEVMSemantics(auto_abstract_gas=auto_abstract_gas),
+            kcfg_semantics=KEVMSemantics(auto_abstract_gas=options.auto_abstract_gas),
             id=test.id,
-            bug_report=bug_report,
-            kore_rpc_command=kore_rpc_command,
-            llvm_definition_dir=llvm_definition_dir,
-            smt_timeout=smt_timeout,
-            smt_retry_limit=smt_retry_limit,
-            trace_rewrites=trace_rewrites,
+            bug_report=options.bug_report,
+            kore_rpc_command=options.kore_rpc_command,
+            llvm_definition_dir=foundry.llvm_library if options.use_booster else None,
+            smt_timeout=options.smt_timeout,
+            smt_retry_limit=options.smt_retry_limit,
+            trace_rewrites=options.trace_rewrites,
             start_server=start_server,
-            port=port,
+            port=options.port,
         ) as kcfg_explore:
             proof = method_to_apr_proof(
-                foundry,
-                test,
-                kcfg_explore,
-                bmc_depth=bmc_depth,
-                run_constructor=run_constructor,
+                test=test,
+                foundry=foundry,
+                kcfg_explore=kcfg_explore,
+                bmc_depth=options.bmc_depth,
+                run_constructor=options.run_constructor,
             )
 
             passed = kevm_prove(
                 foundry.kevm,
                 proof,
                 kcfg_explore,
-                max_depth=max_depth,
-                max_iterations=max_iterations,
-                break_every_step=break_every_step,
-                break_on_jumpi=break_on_jumpi,
-                break_on_calls=break_on_calls,
+                max_depth=options.max_depth,
+                max_iterations=options.max_iterations,
+                break_every_step=options.break_every_step,
+                break_on_jumpi=options.break_on_jumpi,
+                break_on_calls=options.break_on_calls,
             )
             failure_log = None
             if not passed:
-                failure_log = print_failure_info(proof, kcfg_explore, counterexample_info)
+                failure_log = print_failure_info(proof, kcfg_explore, options.counterexample_info)
             return passed, failure_log
 
     _apr_proofs: list[tuple[bool, list[str] | None]]
-    if workers > 1:
-        with ProcessPool(ncpus=workers) as process_pool:
+    if options.workers > 1:
+        with ProcessPool(ncpus=options.workers) as process_pool:
             _apr_proofs = process_pool.map(init_and_run_proof, tests)
     else:
         _apr_proofs = []
@@ -289,8 +237,8 @@ def init_and_run_proof(test: FoundryTest) -> tuple[bool, list[str] | None]:
 
 
 def method_to_apr_proof(
-    foundry: Foundry,
     test: FoundryTest,
+    foundry: Foundry,
     kcfg_explore: KCFGExplore,
     bmc_depth: int | None = None,
     run_constructor: bool = False,
@@ -311,9 +259,9 @@ def method_to_apr_proof(
         setup_proof = _load_constructor_proof(foundry, test.contract)
 
     kcfg, init_node_id, target_node_id = _method_to_initialized_cfg(
-        foundry,
-        test,
-        kcfg_explore,
+        foundry=foundry,
+        test=test,
+        kcfg_explore=kcfg_explore,
         setup_proof=setup_proof,
     )