smtool: symbolic-model tool library checkpoint (wip)

certora_autosetup/utils/cloud_runner.py

@@ -58,6 +58,7 @@ def __init__(
         cloud_server: str | None = None,
         disable_cache: bool = False,
         cancel_jobs_on_cleanup: bool = True,
+        stop_on_first_violation: bool = False,
     ):
         """
         Initialize cloud job manager.
@@ -88,12 +89,21 @@ def __init__(
         )
         self.disable_cache = disable_cache
         self.cancel_jobs_on_cleanup = cancel_jobs_on_cleanup
+        # When True, poll a still-running job's PARTIAL rule results each cycle and cancel the
+        # job as soon as any rule is VIOLATED, instead of waiting for every rule to finish. The
+        # partial results (with the violation) are still parsed and returned. Off by default.
+        self.stop_on_first_violation = stop_on_first_violation
         self.job_wait_timeout = self.JOB_TIMEOUT_SECONDS
 
         # Progress tracking counters (read from spinner thread, written under asyncio lock)
         self._active_jobs = 0
         self._total_completed = 0
 
+        # id(job_spec) -> job_url for jobs currently submitted-and-waiting. Lets batch early
+        # termination cancel the REMOTE cloud jobs of the still-pending tasks (cancelling the local
+        # asyncio task alone leaves the submitted job running on the prover — orphaned/duplicate runs).
+        self._inflight_job_urls: dict[int, str] = {}
+
     @property
     def active_jobs_count(self) -> int:
         """Number of jobs currently being processed."""
@@ -369,14 +379,24 @@ async def _wait_for_tasks_with_early_termination(
                     f"Early termination triggered. Cancelling {len(pending_tasks)} remaining jobs."
                 )
 
-                # Cancel all remaining tasks and create cancelled results
+                # Cancel all remaining tasks and create cancelled results. Cancel the REMOTE cloud job
+                # FIRST (while its url is still in the in-flight registry) — cancelling only the local
+                # asyncio task leaves the submitted job running on the prover (orphaned/duplicate runs
+                # that later hit a wasteful server-side timeout). Read urls before .cancel() so the
+                # task's finally hasn't yet popped them.
                 for pending_task in list(pending_tasks):
-                    pending_task.cancel()
                     pending_job_spec = task_to_job_spec[pending_task]
+                    job_url = self._inflight_job_urls.get(id(pending_job_spec), "")
+                    if job_url:
+                        try:
+                            await self._cancel_cloud_job(job_url)
+                        except Exception as e:
+                            self.log(f"Failed to cancel remote job {job_url}: {e}", "WARNING")
+                    pending_task.cancel()
 
                     # Create cancelled result
                     cancelled_result = await self._create_cancelled_result(
-                        pending_job_spec, job_url=""
+                        pending_job_spec, job_url=job_url
                     )
                     completed_results.append(cancelled_result)
 
@@ -423,8 +443,13 @@ async def _submit_and_wait_single_job(
             f"✓ Successfully submitted {job_spec.contract_name} for {job_spec.phase} - job_url: {job_url}"
         )
 
-        # Wait for the job to complete
-        result = await self.wait_for_completion(job_url, job_spec, completion_callback)
+        # Track the live job so batch early-termination can cancel the REMOTE job (not just our task).
+        self._inflight_job_urls[id(job_spec)] = job_url
+        try:
+            # Wait for the job to complete
+            result = await self.wait_for_completion(job_url, job_spec, completion_callback)
+        finally:
+            self._inflight_job_urls.pop(id(job_spec), None)
 
         # Apply result transformer if provided
         if result_transformer:
@@ -821,7 +846,7 @@ async def _wait_and_parse_job_results(
 
             # Wait for job completion with configurable timeout
             job_wait_timeout = self.job_wait_timeout
-            success, prover_start_time, prover_finish_time = await self._wait_for_job_completion_with_api(
+            success, prover_start_time, prover_finish_time, early_stop_checks = await self._wait_for_job_completion_with_api(
                 prover_api, job_url, job_wait_timeout
             )
 
@@ -837,6 +862,37 @@ async def _wait_and_parse_job_results(
             except Exception as e:
                 self.log(f"Could not record prover runtime for usage ledger: {e}", "DEBUG")
 
+            if early_stop_checks is not None:
+                # stop_on_first_violation fired: the job was cancelled right after a rule VIOLATED.
+                # Reuse the partial checks captured at detection (robust to a gappy post-cancel refetch)
+                # and return a non-success result carrying the violation for the caller to act on.
+                job_handle.status = JobStatus.CANCELLED
+                rule_results = self._checks_to_rule_results(early_stop_checks, job_url)
+                log_with_contract(
+                    self.component,
+                    "info",
+                    job_spec.contract_name,
+                    f"Stopped on first violation after {duration:.1f}s "
+                    f"({len(rule_results)} partial rule result(s))",
+                )
+                return ProverResult(
+                    job_handle=job_handle,
+                    success=False,
+                    report_path=None,
+                    output_data={
+                        "job_url": job_url,
+                        "rule_count": len(rule_results),
+                        "stopped_on_first_violation": True,
+                        "prover_start_time": prover_start_time,
+                        "prover_finish_time": prover_finish_time,
+                    },
+                    job_spec=job_spec,
+                    rule_results=rule_results,
+                    alerts=[],
+                    duration=duration,
+                    transformed_result=None,
+                )
+
             if success:
                 # Job completed successfully, parse results
                 job_handle.status = JobStatus.COMPLETED
@@ -947,14 +1003,29 @@ async def _wait_and_parse_job_results(
                 transformed_result=None,
             )
 
+    def _partial_violated_checks(self, prover_api, job_url: str):
+        """Best-effort: fetch a (possibly still-running) job's partial checks and return the VIOLATED
+        ones. Returns (all_checks, violated_checks). Never raises — a fetch failure on an in-flight or
+        just-cancelled job (missing/partial files) yields ([], []) so polling simply continues."""
+        try:
+            all_checks = list(prover_api.get_all_checks(job_url) or [])
+        except Exception as e:
+            self.log(f"partial-check fetch failed for {job_url}: {e}", "DEBUG")
+            return [], []
+        violated = [c for c in all_checks if c.is_violated]
+        return all_checks, violated
+
     async def _wait_for_job_completion_with_api(
         self, prover_api: ProverOutputAPI, job_url: str, timeout_seconds: int
-    ) -> tuple[bool, Optional[float], Optional[float]]:
+    ) -> tuple[bool, Optional[float], Optional[float], Optional[list]]:
         """Wait for job completion using ProverOutputAPI.
 
         Returns:
-            Tuple of (success, prover_start_time, prover_finish_time).
-            Times may be None if unavailable.
+            Tuple of (success, prover_start_time, prover_finish_time, early_stop_checks).
+            Times may be None if unavailable. early_stop_checks is None on a normal
+            completion/failure; when stop_on_first_violation fired it holds the partial
+            checks captured just before the job was cancelled (so the caller can report
+            the violation). success is False in that case.
         """
         import asyncio
 
@@ -996,19 +1067,32 @@ async def _wait_for_job_completion_with_api(
                         # Note: HALTED jobs are treated as successful in PreAudit because they often contain
                         # partial results for some rules that can still be analyzed
                         self.log(f"Job completed successfully: {job_url}")
-                        return True, prover_start, prover_finish
+                        return True, prover_start, prover_finish, None
                     elif job_info.status in [ProverJobStatus.FAILED, ProverJobStatus.CANCELED, ProverJobStatus.SERVICE_UNAVAILABLE, ProverJobStatus.UPLOAD_FAILED]:
                         self.log(f"Job failed with status {job_info.status}: {job_url}")
-                        return False, prover_start, prover_finish
+                        return False, prover_start, prover_finish, None
                     # Check if job has completed but with an unrecognized status
                     elif hasattr(job_info, "is_completed") and job_info.is_completed:
                         self.log(
                             f"❌ Job completed with unrecognized status '{job_info.status}' - "
                             f"treating as failed: {job_url}",
                             "WARNING"
                         )
-                        return False, prover_start, prover_finish
-                    # If status is 'RUNNING', 'QUEUED', etc., continue waiting
+                        return False, prover_start, prover_finish, None
+                    # Status is 'RUNNING'/'QUEUED'/etc. Optionally short-circuit: if any rule has already
+                    # VIOLATED, cancel the job now instead of waiting for the remaining rules. The partial
+                    # checks captured here are returned so the caller reports the violation even if a
+                    # post-cancel refetch comes back gappy.
+                    if self.stop_on_first_violation:
+                        all_checks, violated = self._partial_violated_checks(prover_api, job_url)
+                        if violated:
+                            names = ", ".join(sorted({c.rule_name for c in violated})[:3])
+                            self.log(
+                                f"stop_on_first_violation: {len(violated)} check(s) VIOLATED ({names}) — "
+                                f"cancelling {job_url}"
+                            )
+                            await self._cancel_cloud_job(job_url)
+                            return False, prover_start, prover_finish, all_checks
                 else:
                     self.log(f"No job info returned for {job_url}")
 
@@ -1022,12 +1106,12 @@ async def _wait_for_job_completion_with_api(
                     self.log(
                         f"Authentication issue detected, assuming job completed: {job_url}"
                     )
-                    return True, None, None
+                    return True, None, None, None
                 await asyncio.sleep(poll_interval)
 
         # Timeout reached
         self.log(f"Job completion timeout after {timeout_seconds}s", "WARNING")
-        return False, None, None
+        return False, None, None, None
 
     def _create_failed_result(
         self, job_spec: ProverJobSpec, cache_key: str, error_msg: str

certora_autosetup/utils/prover_runner.py

@@ -529,7 +529,19 @@ def parse_rule_results_from_job(self, job_identifier: str) -> List[RuleResult]:
 
         try:
             all_checks = self.prover_api.get_all_checks(job_identifier)
+        except Exception as e:
+            self.log(
+                f"Failed to fetch checks for job {job_identifier}: {e}", "WARNING"
+            )
+            return []
+        return self._checks_to_rule_results(all_checks, job_identifier)
 
+    def _checks_to_rule_results(self, all_checks, job_identifier: str = "") -> List[RuleResult]:
+        """Convert already-fetched prover checks into RuleResult objects. Split out from
+        parse_rule_results_from_job so a caller that ALREADY holds checks (e.g. partial checks captured
+        while polling a still-running job for stop-on-first-violation) can reuse the same conversion
+        without a second network fetch. Never raises — a malformed check is skipped, not fatal."""
+        try:
             # Convert checks to RuleResult objects
             rule_results = []
             sanity_rule_count = 0
@@ -569,6 +581,9 @@ def parse_rule_results_from_job(self, job_identifier: str) -> List[RuleResult]:
                     notifications=notifications,
                     duration=duration,
                     assert_type=assert_type,
+                    node_type=check_result.node_type.value
+                    if hasattr(check_result.node_type, "value")
+                    else str(check_result.node_type),
                 )
                 rule_results.append(rule_result)
 

certora_autosetup/utils/runner_types.py

@@ -118,6 +118,7 @@ class RuleResult:
     is_sanity_rule: bool = False
     is_leaf: bool = True
     level: int = 0
+    node_type: Optional[str] = None  # POU NodeType, e.g. "ROOT" (per-rule top-level) vs sub-check leaves
     notifications: List[NotificationType] = field(default_factory=list)  # All rule notifications/warnings
     assert_type: Optional[AssertType] = None  # Type of assertion that was violated (for SAT results)
 

composer/cvl/pretty_print.py

@@ -353,6 +353,8 @@ def _print_cvl_type(self, cvl_type: CVLType) -> str:
                 return f"{self._print_cvl_type(cvl_type.base_type)}[{cvl_type.n}]"
             case "primitive":
                 return cvl_type.type_name
+            case "special":
+                return cvl_type.type_name
             case "storage_type":
                 return "storage"
             case "contract_type":
@@ -453,7 +455,7 @@ def _print_revert_cmd(self, cmd: RevertCmd) -> str:
         if cmd.message:
             return f"revert(\"{cmd.message}\");"
         else:
-            return "revert;"
+            return "revert();"
 
     def _print_elif(self, else_blk: ElseBlock, ident: LineBuilder):
         match else_blk.type:
@@ -527,6 +529,9 @@ def print_basic_block(self, block: BasicBlock) -> str:
                 self._print_hook_def(block, builder)
             case "methods_block":
                 self._print_methods_block(block, builder)
+            case "use_directive":
+                kw = "builtin rule" if block.use_kind == "builtin_rule" else block.use_kind
+                builder.line(f"use {kw} {block.name};")
         return "\n".join(builder.buffer)
 
     def _print_rule_block(self, rule: RuleBlock, ident: LineBuilder):
@@ -579,6 +584,7 @@ def _print_invariant(self, inv: Invariant, ident: LineBuilder):
             if inv.proofs:
                 lb.append(" {")
             else:
+                lb.append(";")   # CVL 2: an invariant with no preserved block must end with `;`
                 return
         with ident.indent() as nested:
             for proof in inv.proofs:
@@ -824,7 +830,8 @@ def _print_method_signature(self, meth_sig: MethodSignature) -> str:
             postfix=")"
         )
         res += f" {meth_sig.visibility}"
-        res += self.print_and_join(meth_sig.return_types, self._print_vm_param, prefix=" returns (", postfix=")")
+        if meth_sig.return_types:
+            res += self.print_and_join(meth_sig.return_types, self._print_vm_param, prefix=" returns (", postfix=")")
         if meth_sig.post_flags:
             res += " " + " ".join(meth_sig.post_flags)
         return res

composer/cvl/schema.py

@@ -46,6 +46,17 @@ class StorageType(BaseModel):
     """
     type: Literal["storage_type"]
 
+class SpecialType(BaseModel):
+    """
+    A CVL-only leaf type with no Solidity counterpart: the mathematical integer
+    `mathint`, the transaction environment `env`, an abstract `method`, and the
+    generic-argument placeholder `calldataarg`. These are legal as CVL variable,
+    parameter, and function return types but never cross the VM boundary.
+    """
+    type: Literal["special"]
+    type_name: Literal["mathint", "env", "method", "calldataarg"] = Field(
+        description="The CVL-only type, e.g. `mathint` or `env`")
+
 class PrimitiveType(BaseModel):
     """
     One of the built-in primitive types allowed by solidity. This includes all of the primitives
@@ -69,7 +80,7 @@ class PrimitiveType(BaseModel):
     ] = Field(description="The name of the type, e.g., `uint256` or `bool`")
 
 CVLType = Annotated[
-    MappingType | ArrayType | StaticArrayType | ContractType | StorageType | PrimitiveType,
+    MappingType | ArrayType | StaticArrayType | ContractType | StorageType | SpecialType | PrimitiveType,
     Discriminator("type")
 ]
 
@@ -509,7 +520,11 @@ class HookDef(BaseModel):
     block: CodeBlock = Field(description="The CVL code block to execute when the hook is triggered. Can access the identifiers bound in the pattern")
 
 class UseDirective(BaseModel):
-    pass
+    # `use invariant X;` / `use rule X;` / `use builtin rule X;` — include an imported invariant/rule
+    # in this spec's verification. (Needed to prove an invariant that is declared in an imported spec.)
+    type: Literal["use_directive"] = "use_directive"
+    use_kind: Literal["invariant", "rule", "builtin_rule"] = "invariant"
+    name: str
 
 class OverrideDirective(BaseModel):
     pass
@@ -652,7 +667,8 @@ class MethodsBlock(BaseModel):
     method_entries: list[MethodEntry] = Field(description="List of method declarations and summaries")
 
 BasicBlock = Annotated[
-    Invariant | FunctionDef | RuleBlock | SortDef | GhostDef | MacroDef | HookDef | MethodsBlock,
+    Invariant | FunctionDef | RuleBlock | SortDef | GhostDef | MacroDef | HookDef | MethodsBlock
+    | UseDirective,
     Discriminator("type")
 ]
 

pyproject.toml

@@ -134,6 +134,7 @@ tui-autoprove = "composer.cli.tui_autoprove:main"
 console-foundry = "composer.cli.console_foundry:main"
 tui-foundry = "composer.cli.tui_foundry:main"
 autoprove-report-render = "composer.spec.source.report.render:main"
+detect-summaries = "summarization_detector.cli:main"
 tui-natspec = "composer.cli.tui_pipeline:main"
 cache-natspec = "composer.cli.cache_natspec:main"
 cache-autoprove = "composer.cli.cache_autoprove:main"

pyrightconfig.json

@@ -1,5 +1,5 @@
 {
-  "include": ["composer", "analyzer", "certora_autosetup", "sanity_analyzer"],
+  "include": ["composer", "analyzer", "certora_autosetup", "sanity_analyzer", "summarization_detector"],
   "extraPaths": ["graphcore"],
   "stubPath": "stubs",
   "typeCheckingMode": "standard",

smtool/.gitignore

@@ -0,0 +1,4 @@
+generated/
+__pycache__/
+*.pyc
+.certora_internal/

smtool/agent/__init__.py

@@ -0,0 +1 @@
+