diff --git a/certora_autosetup/utils/cloud_runner.py b/certora_autosetup/utils/cloud_runner.py index 8cd4d815..7eb88313 100644 --- a/certora_autosetup/utils/cloud_runner.py +++ b/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,10 +1047,10 @@ 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( @@ -1007,8 +1058,21 @@ async def _wait_for_job_completion_with_api( 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 diff --git a/certora_autosetup/utils/prover_runner.py b/certora_autosetup/utils/prover_runner.py index 2bc10678..b4dce09b 100644 --- a/certora_autosetup/utils/prover_runner.py +++ b/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 diff --git a/tests/test_cloudrunner_stop_on_violation.py b/tests/test_cloudrunner_stop_on_violation.py new file mode 100644 index 00000000..f8d55e4f --- /dev/null +++ b/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