CloudRunner: stop_on_first_violation — cancel a running job on the first violated rule

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,6 +89,10 @@ 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)
@@ -821,7 +826,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 +842,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 +983,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 +1047,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 +1086,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

tests/test_cloudrunner_stop_on_violation.py

@@ -0,0 +1,101 @@
+"""CloudProverRunner.stop_on_first_violation: cancel a still-running job the moment a rule VIOLATES,
+and fetch partial results robustly (a cancelled/gappy job must degrade, never crash).
+"""
+import asyncio
+from types import SimpleNamespace
+
+from prover_output_utility.models import JobStatus as ProverJobStatus
+
+from certora_autosetup.utils.cloud_runner import CloudProverRunner
+from certora_autosetup.utils.prover_runner import ProverRunner
+
+
+def _check(rule, violated):
+    # mirrors prover_output_utility.models.CheckResult: the helper reads the `is_violated` property.
+    return SimpleNamespace(rule_name=rule, is_violated=violated)
+
+
+def _stub(**attrs):
+    s = SimpleNamespace(**attrs)
+    s.log = lambda msg, level="INFO": None
+    return s
+
+
+# ---- _partial_violated_checks: filter + never-raise ------------------------------------------------
+
+def test_partial_violated_checks_filters_violated():
+    api = SimpleNamespace(get_all_checks=lambda url: [
+        _check("a", False), _check("b", True), _check("c", False), _check("b", True),
+    ])
+    stub = _stub()
+    allc, viol = CloudProverRunner._partial_violated_checks(stub, api, "u")
+    assert len(allc) == 4
+    assert sorted({c.rule_name for c in viol}) == ["b"]
+    assert len(viol) == 2
+
+
+def test_partial_violated_checks_never_raises_on_fetch_error():
+    def boom(url):
+        raise RuntimeError("job cancelled, tree missing")
+    stub = _stub()
+    allc, viol = CloudProverRunner._partial_violated_checks(stub, SimpleNamespace(get_all_checks=boom), "u")
+    assert allc == [] and viol == []
+
+
+# ---- parse robustness on a cancelled/gappy job -----------------------------------------------------
+
+def test_parse_rule_results_empty_on_fetch_error():
+    def boom(job):
+        raise RuntimeError("cannot fetch checks for cancelled job")
+    stub = _stub(prover_api=SimpleNamespace(get_all_checks=boom))
+    out = ProverRunner.parse_rule_results_from_job(stub, "cancelled-job-id")
+    assert out == []
+
+
+# ---- the poll-loop hook: cancel + return partial checks on first violation --------------------------
+
+def test_wait_cancels_and_returns_partial_on_first_violation():
+    cancelled = {"called": False}
+    async def fake_cancel(url):
+        cancelled["called"] = True
+        return True
+    checks = [_check("ok", False), _check("bad", True)]
+    api = SimpleNamespace(
+        get_job_info=lambda url: SimpleNamespace(status=ProverJobStatus.RUNNING, start_time=1.0, finish_time=None),
+        get_all_checks=lambda url: checks,
+    )
+    stub = _stub(stop_on_first_violation=True, _cancel_cloud_job=fake_cancel)
+    stub._partial_violated_checks = CloudProverRunner._partial_violated_checks.__get__(stub)
+    success, _s, _f, early = asyncio.run(
+        CloudProverRunner._wait_for_job_completion_with_api(stub, api, "u", 60)
+    )
+    assert success is False
+    assert cancelled["called"] is True
+    assert early is not None and [c.rule_name for c in early] == ["ok", "bad"]
+
+
+def test_wait_ignores_violation_when_flag_off():
+    # flag off -> a RUNNING poll with a violated check must NOT cancel; a subsequent SUCCEEDED ends it.
+    seq = [ProverJobStatus.RUNNING, ProverJobStatus.SUCCEEDED]
+    def get_job_info(url):
+        return SimpleNamespace(status=seq.pop(0), start_time=1.0, finish_time=2.0)
+    cancelled = {"called": False}
+    async def fake_cancel(url):
+        cancelled["called"] = True
+        return True
+    api = SimpleNamespace(get_job_info=get_job_info, get_all_checks=lambda url: [_check("bad", True)])
+    stub = _stub(stop_on_first_violation=False, _cancel_cloud_job=fake_cancel)
+    stub._partial_violated_checks = CloudProverRunner._partial_violated_checks.__get__(stub)
+    # poll_interval is 10s; shrink the wait by making the 2nd poll SUCCEED. asyncio.sleep(10) would stall
+    # the test, so patch asyncio.sleep to a no-op for this call.
+    import certora_autosetup.utils.cloud_runner as cr
+    orig_sleep = asyncio.sleep
+    async def nosleep(_): return None
+    asyncio.sleep = nosleep
+    try:
+        success, _s, _f, early = asyncio.run(
+            CloudProverRunner._wait_for_job_completion_with_api(stub, api, "u", 60)
+        )
+    finally:
+        asyncio.sleep = orig_sleep
+    assert success is True and early is None and cancelled["called"] is False