diff --git a/certora_autosetup/reporting/reporter.py b/certora_autosetup/reporting/reporter.py index a041220a..0ed6d742 100644 --- a/certora_autosetup/reporting/reporter.py +++ b/certora_autosetup/reporting/reporter.py @@ -602,7 +602,36 @@ def _collect_sanity_rows( except Exception as e: self.log(f"Failed to read config for {contract_name}: {e}", "WARNING") - jr = self._prover_api.get_job_report(result.job_url) + if result.is_preprocessing_timeout: + # The job was cancelled by the watchdog before producing any results — + # there is no job report to fetch; emit a dedicated row instead. + rows.append(_SanityRow( + contract_name=contract_name, + job_url=result.job_url, + sanity_status=( + "PREPROCESSING TIMEOUT — prover never finished preprocessing; " + "job cancelled by autosetup watchdog" + ), + method_failures={}, + unresolved_count=0, + storage_extension=config.get("storage_extension_annotation", False), + global_warnings=0, + runtime="N/A", + optimistic_loop=config.get("optimistic_loop", False), + loop_iter=config.get("loop_iter", "N/A"), + optimistic_hashing=config.get("optimistic_hashing", False), + hashing_bounds=config.get("hashing_length_bound", "N/A"), + )) + continue + + try: + jr = self._prover_api.get_job_report(result.job_url) + except Exception as e: + # A cancelled/failed job may have no report — skip the row rather than + # losing the whole summary. + self.log(f"Failed to fetch job report for {contract_name} " + f"({result.job_url}): {e}", "WARNING") + continue job_report_path = self.reports_dir / f"job_report_{contract_name}.json" with open(job_report_path, "w") as f: json.dump(jr.to_dict(), f, indent=2) diff --git a/certora_autosetup/utils/cloud_runner.py b/certora_autosetup/utils/cloud_runner.py index 8cd4d815..b976f5a6 100644 --- a/certora_autosetup/utils/cloud_runner.py +++ b/certora_autosetup/utils/cloud_runner.py @@ -10,6 +10,7 @@ import json import os import time +from enum import Enum from pathlib import Path from typing import Any, Dict, List, Optional, Sequence, cast @@ -25,6 +26,7 @@ ProverRunner, ResultTransformer, ) +from .preprocessing_watchdog import PreprocessingWatchdog, WatchdogVerdict from .runner_types import ( JobHandle, JobStatus, @@ -40,6 +42,14 @@ def extract_job_url(output: str) -> Optional[str]: return extract_job_url_from_text(output) +class _WaitOutcome(Enum): + """Terminal outcome of waiting on a cloud job.""" + + COMPLETED = "completed" + FAILED = "failed" # includes the overall wait timeout (unchanged semantics) + PREPROCESSING_TIMEOUT = "preprocessing_timeout" + + class CloudProverRunner(ProverRunner): """ Cloud job manager for prover execution. @@ -50,6 +60,15 @@ class CloudProverRunner(ProverRunner): # Timeout for job completion including job queue time - 150 minutes JOB_TIMEOUT_SECONDS = 7200 + 1800 + # Preprocessing watchdog defaults (see preprocessing_watchdog.py). The budget is + # time spent in RUNNING without the prover producing a treeview; healthy jobs + # produce one within minutes, while preprocessing blowups never do and would + # otherwise burn the whole JOB_TIMEOUT_SECONDS. Env-overridable; budget <= 0 + # disables the watchdog. + PREPROCESSING_BUDGET_SECONDS = 1800 + PREPROCESSING_GRACE_SECONDS = 300 + PREPROCESSING_PROBE_SECONDS = 90 + def __init__( self, project_root: Path, @@ -89,6 +108,12 @@ def __init__( self.disable_cache = disable_cache self.cancel_jobs_on_cleanup = cancel_jobs_on_cleanup self.job_wait_timeout = self.JOB_TIMEOUT_SECONDS + self.preprocessing_budget = int(os.environ.get( + "AUTOSETUP_PREPROCESSING_BUDGET_SECONDS", self.PREPROCESSING_BUDGET_SECONDS)) + self.preprocessing_grace = int(os.environ.get( + "AUTOSETUP_PREPROCESSING_GRACE_SECONDS", self.PREPROCESSING_GRACE_SECONDS)) + self.preprocessing_probe_interval = int(os.environ.get( + "AUTOSETUP_PREPROCESSING_PROBE_SECONDS", self.PREPROCESSING_PROBE_SECONDS)) # Progress tracking counters (read from spinner thread, written under asyncio lock) self._active_jobs = 0 @@ -821,9 +846,10 @@ 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( + outcome, prover_start_time, prover_finish_time = await self._wait_for_job_completion_with_api( prover_api, job_url, job_wait_timeout ) + success = outcome is _WaitOutcome.COMPLETED duration = time.time() - start_time @@ -895,6 +921,43 @@ async def _wait_and_parse_job_results( duration=duration, transformed_result=None, ) + elif outcome is _WaitOutcome.PREPROCESSING_TIMEOUT: + job_handle.status = JobStatus.PREPROCESSING_TIMEOUT + error_msg = ( + f"Preprocessing timeout: prover reported no rules in its treeview within " + f"{self.preprocessing_budget}s of running (stuck in preprocessing); " + f"job was cancelled after {duration:.1f}s" + ) + + log_with_contract( + self.component, + "error", + job_spec.contract_name, + error_msg, + ) + + # Drop the cached submission so a deliberate future run submits fresh + # instead of re-attaching to the cancelled job. + if not self.disable_cache: + await self._remove_submission_cache(cache_key) + + return ProverResult( + job_handle=job_handle, + success=False, + report_path=None, + output_data={ + "job_url": job_url, + "output": submission_result.output if submission_result else "", + "return_code": submission_result.return_code + if submission_result + else 0, + "preprocessing_timeout": True, + }, + job_spec=job_spec, + error_message=error_msg, + duration=duration, + transformed_result=None, + ) else: # Job failed or timed out job_handle.status = JobStatus.FAILED @@ -949,11 +1012,11 @@ async def _wait_and_parse_job_results( 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[_WaitOutcome, Optional[float], Optional[float]]: """Wait for job completion using ProverOutputAPI. Returns: - Tuple of (success, prover_start_time, prover_finish_time). + Tuple of (outcome, prover_start_time, prover_finish_time). Times may be None if unavailable. """ import asyncio @@ -961,6 +1024,18 @@ async def _wait_for_job_completion_with_api( start_time = time.time() poll_interval = 10 # Poll every 10 seconds for faster completion detection + # A job with no treeview after the budget is stuck in prover preprocessing and + # will never produce results — cancel it instead of waiting out the full timeout. + watchdog = None + if self.preprocessing_budget > 0: + watchdog = PreprocessingWatchdog( + budget_seconds=self.preprocessing_budget, + grace_seconds=self.preprocessing_grace, + probe_interval_seconds=self.preprocessing_probe_interval, + probe_treeview=lambda: prover_api.get_treeview_status(job_url), + log=self.log, + ) + while time.time() - start_time < timeout_seconds: try: # Check job status using ProverOutputAPI with timeout @@ -996,10 +1071,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 _WaitOutcome.COMPLETED, prover_start, prover_finish 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 _WaitOutcome.FAILED, prover_start, prover_finish # 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 +1082,30 @@ async def _wait_for_job_completion_with_api( f"treating as failed: {job_url}", "WARNING" ) - return False, prover_start, prover_finish + return _WaitOutcome.FAILED, prover_start, prover_finish # If status is 'RUNNING', 'QUEUED', etc., continue waiting + + if watchdog is not None: + is_running = job_info.status == ProverJobStatus.RUNNING + try: + verdict = await asyncio.wait_for( + asyncio.get_event_loop().run_in_executor( + None, watchdog.observe, is_running + ), + timeout=60, + ) + except asyncio.TimeoutError: + verdict = WatchdogVerdict.WAITING # probe hung; try again next tick + if verdict is WatchdogVerdict.PREPROCESSING_TIMEOUT: + self.log( + f"⏱ Preprocessing watchdog: treeview shows no rules after " + f"{self.preprocessing_budget}s of RUNNING — prover is stuck in " + f"preprocessing; cancelling {job_url}", + "WARNING", + ) + if not await self._cancel_cloud_job(job_url): + self.log(f"Could not cancel {job_url}; abandoning it anyway", "WARNING") + return _WaitOutcome.PREPROCESSING_TIMEOUT, prover_start, prover_finish else: self.log(f"No job info returned for {job_url}") @@ -1022,12 +1119,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 _WaitOutcome.COMPLETED, 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 _WaitOutcome.FAILED, None, None def _create_failed_result( self, job_spec: ProverJobSpec, cache_key: str, error_msg: str diff --git a/certora_autosetup/utils/job_problem_fixes.py b/certora_autosetup/utils/job_problem_fixes.py index 15d36215..a8d779de 100644 --- a/certora_autosetup/utils/job_problem_fixes.py +++ b/certora_autosetup/utils/job_problem_fixes.py @@ -140,6 +140,11 @@ def on_job_problem( """ if result.success: return False + if result.is_preprocessing_timeout: + # The prover never finished preprocessing — a prover-side scalability stall, + # not a conf problem; re-running an identical job would just burn another + # watchdog budget. + return False for workaround in _WORKAROUNDS: try: if workaround(result, config_manager, prover_api): diff --git a/certora_autosetup/utils/preprocessing_watchdog.py b/certora_autosetup/utils/preprocessing_watchdog.py new file mode 100644 index 00000000..6b74615c --- /dev/null +++ b/certora_autosetup/utils/preprocessing_watchdog.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python3 +""" +Preprocessing watchdog for cloud prover jobs. + +The prover's treeViewStatus.json carries an empty "rules" list for as long as the run is +in preprocessing (scene construction, storage/pointer analyses, CVL typechecking) — rules +appear in it only once rule checking begins. (Verified live: a stuck-preprocessing cloud +job serves a treeview with `rules: []`; the file itself exists early, so mere existence is +NOT the signal.) A job whose preprocessing blows up sits in RUNNING with an empty rule list +until the prover's global timeout (~2h), and autosetup would otherwise burn its whole +per-job wait budget polling it. + +The watchdog piggy-backs on the existing status-poll loop: it starts a clock on the first +RUNNING observation (queue time is free), and after a grace period probes the treeview at a +bounded cadence. First probe showing a non-empty "rules" list ⇒ preprocessing passed, +watchdog goes dormant. No rules within the budget ⇒ PREPROCESSING_TIMEOUT, and the caller +cancels the job. Probe transport errors self-disable the watchdog so the loop degrades to +the old behavior; a JobNotFoundError is NOT an error — it is the normal "treeview not +served yet" response early in a run. +""" + +import time +from enum import Enum +from typing import Callable, Optional + +from prover_output_utility.exceptions import JobNotFoundError # type: ignore[import-untyped] + + +class WatchdogVerdict(Enum): + WAITING = "waiting" # keep polling normally + PREPROCESSING_DONE = "preprocessing_done" # treeview seen; dormant forever + PREPROCESSING_TIMEOUT = "preprocessing_timeout" # budget exceeded; caller should cancel + DISABLED = "disabled" # probe endpoint unusable; old behavior + + +class PreprocessingWatchdog: + """Rate-limited treeview prober; synchronous so it can run inside the poll executor.""" + + def __init__( + self, + budget_seconds: float, + grace_seconds: float, + probe_interval_seconds: float, + probe_treeview: Callable[[], object], + log: Callable[..., None], + max_consecutive_probe_errors: int = 3, + clock: Callable[[], float] = time.monotonic, + ): + self.budget_seconds = budget_seconds + self.grace_seconds = grace_seconds + self.probe_interval_seconds = probe_interval_seconds + self._probe_treeview = probe_treeview + self._log = log + self._max_consecutive_probe_errors = max_consecutive_probe_errors + self._clock = clock + + self._running_since: Optional[float] = None + self._last_probe: Optional[float] = None + self._done = False + self._disabled = False + self._consecutive_probe_errors = 0 + + def observe(self, is_running: bool) -> WatchdogVerdict: + """Feed one status-poll tick. `is_running` is True only for RUNNING status — + queued/posted/starting ticks never start the preprocessing clock.""" + if self._done: + return WatchdogVerdict.PREPROCESSING_DONE + if self._disabled: + return WatchdogVerdict.DISABLED + if not is_running: + return WatchdogVerdict.WAITING + + now = self._clock() + if self._running_since is None: + self._running_since = now + + running_for = now - self._running_since + if running_for < self.grace_seconds: + return WatchdogVerdict.WAITING + if self._last_probe is not None and now - self._last_probe < self.probe_interval_seconds: + # Between probes; the budget check only fires on probe ticks so the final + # verdict is always backed by a fresh "still no treeview" observation. + return WatchdogVerdict.WAITING + + self._last_probe = now + try: + tree_data = self._probe_treeview() + except JobNotFoundError: + # Treeview not served yet — expected early in a run. + self._consecutive_probe_errors = 0 + tree_data = None + except Exception as e: + self._consecutive_probe_errors += 1 + self._log( + f"Preprocessing watchdog: treeview probe failed " + f"({self._consecutive_probe_errors}/{self._max_consecutive_probe_errors}): {e}", + "WARNING", + ) + if self._consecutive_probe_errors >= self._max_consecutive_probe_errors: + self._disabled = True + self._log( + "Preprocessing watchdog disabled after repeated probe failures; " + "falling back to plain status polling", + "WARNING", + ) + return WatchdogVerdict.DISABLED + return WatchdogVerdict.WAITING + + # The treeview exists (with rules: []) throughout preprocessing; only a + # non-empty rules list proves rule checking has started. + if isinstance(tree_data, dict) and tree_data.get("rules"): + self._done = True + return WatchdogVerdict.PREPROCESSING_DONE + + if running_for > self.budget_seconds: + return WatchdogVerdict.PREPROCESSING_TIMEOUT + return WatchdogVerdict.WAITING diff --git a/certora_autosetup/utils/runner_types.py b/certora_autosetup/utils/runner_types.py index d93f088a..58368ac6 100644 --- a/certora_autosetup/utils/runner_types.py +++ b/certora_autosetup/utils/runner_types.py @@ -46,6 +46,10 @@ class JobStatus(Enum): COMPLETED = "completed" FAILED = "failed" CANCELLED = "cancelled" + # Cancelled by the preprocessing watchdog: the prover never produced a treeview + # (never finished preprocessing) within the watchdog budget. Distinct from FAILED + # so retry logic can skip conf workarounds that cannot help. + PREPROCESSING_TIMEOUT = "preprocessing_timeout" @dataclass @@ -147,6 +151,11 @@ class ProverResult: alerts: List[ParsedAlert] = field(default_factory=list) transformed_result: Optional[Any] = None # Generic transformed result + @property + def is_preprocessing_timeout(self) -> bool: + """True when the job was cancelled by the preprocessing watchdog.""" + return self.job_handle.status == JobStatus.PREPROCESSING_TIMEOUT + @property def job_url(self) -> Optional[str]: """Get job URL from job_handle or output data.""" diff --git a/tests/test_preprocessing_watchdog.py b/tests/test_preprocessing_watchdog.py new file mode 100644 index 00000000..62b12b5d --- /dev/null +++ b/tests/test_preprocessing_watchdog.py @@ -0,0 +1,295 @@ +"""Tests for the cloud-job preprocessing watchdog. + +The watchdog probes the prover's treeview (written only once rule checking begins) to +detect jobs stuck in preprocessing, cancels them, and classifies them distinctly so no +conf workaround retries an identical doomed job. +""" + +import asyncio +import time +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from prover_output_utility.exceptions import JobNotFoundError +from prover_output_utility.models import JobStatus as ProverJobStatus + +from certora_autosetup.utils import job_problem_fixes +from certora_autosetup.utils.cloud_runner import CloudProverRunner, _WaitOutcome +from certora_autosetup.utils.job_problem_fixes import on_job_problem +from certora_autosetup.utils.preprocessing_watchdog import ( + PreprocessingWatchdog, + WatchdogVerdict, +) +from certora_autosetup.utils.runner_types import ( + JobHandle, + JobStatus, + ProverResult, + RunnerType, +) + + +class FakeClock: + def __init__(self): + self.now = 1000.0 + + def __call__(self) -> float: + return self.now + + def advance(self, seconds: float) -> None: + self.now += seconds + + +def make_watchdog(clock, probe, budget=1800, grace=300, interval=90, max_errors=3): + return PreprocessingWatchdog( + budget_seconds=budget, + grace_seconds=grace, + probe_interval_seconds=interval, + probe_treeview=probe, + log=lambda *a, **k: None, + max_consecutive_probe_errors=max_errors, + clock=clock, + ) + + +class TestWatchdogStateMachine: + def test_queue_time_never_starts_the_clock(self): + clock = FakeClock() + probe = MagicMock() + wd = make_watchdog(clock, probe) + for _ in range(10): + assert wd.observe(is_running=False) is WatchdogVerdict.WAITING + clock.advance(600) # far beyond budget while queued + probe.assert_not_called() + + def test_no_probes_during_grace(self): + clock = FakeClock() + probe = MagicMock() + wd = make_watchdog(clock, probe, grace=300) + assert wd.observe(is_running=True) is WatchdogVerdict.WAITING + clock.advance(299) + assert wd.observe(is_running=True) is WatchdogVerdict.WAITING + probe.assert_not_called() + + def test_probe_cadence_rate_limited(self): + clock = FakeClock() + probe = MagicMock(side_effect=JobNotFoundError("not yet")) + wd = make_watchdog(clock, probe, grace=0, interval=90) + for _ in range(9): # 9 ticks x 10s = 80s after the first probe + wd.observe(is_running=True) + clock.advance(10) + assert probe.call_count == 1 # first probe only; next allowed at +90s + clock.advance(10) + wd.observe(is_running=True) + assert probe.call_count == 2 + + def test_treeview_with_rules_makes_watchdog_dormant(self): + clock = FakeClock() + probe = MagicMock(return_value={"rules": [{"name": "sanity"}]}) + wd = make_watchdog(clock, probe, grace=0) + assert wd.observe(is_running=True) is WatchdogVerdict.PREPROCESSING_DONE + clock.advance(10_000) + assert wd.observe(is_running=True) is WatchdogVerdict.PREPROCESSING_DONE + assert probe.call_count == 1 # dormant: no further probes ever + + def test_empty_rules_treeview_is_still_preprocessing(self): + # Verified live: the cloud serves treeViewStatus.json with rules: [] for the + # whole preprocessing phase — existence alone must NOT count as done. + clock = FakeClock() + probe = MagicMock(return_value={"rules": [], "contract": "Vault"}) + wd = make_watchdog(clock, probe, budget=1800, grace=0, interval=90) + verdict = wd.observe(is_running=True) + while verdict is WatchdogVerdict.WAITING and clock.now < 1000 + 3600: + clock.advance(90) + verdict = wd.observe(is_running=True) + assert verdict is WatchdogVerdict.PREPROCESSING_TIMEOUT + + def test_budget_exceeded_without_treeview(self): + clock = FakeClock() + probe = MagicMock(side_effect=JobNotFoundError("not yet")) + wd = make_watchdog(clock, probe, budget=1800, grace=0, interval=90) + verdict = wd.observe(is_running=True) + while verdict is WatchdogVerdict.WAITING and clock.now < 1000 + 3600: + clock.advance(90) + verdict = wd.observe(is_running=True) + assert verdict is WatchdogVerdict.PREPROCESSING_TIMEOUT + + def test_not_found_is_not_an_error(self): + clock = FakeClock() + probe = MagicMock(side_effect=JobNotFoundError("not yet")) + wd = make_watchdog(clock, probe, grace=0, interval=0, max_errors=3) + for _ in range(10): + assert wd.observe(is_running=True) is not WatchdogVerdict.DISABLED + clock.advance(1) + + def test_disable_after_consecutive_errors_and_reset_on_success(self): + clock = FakeClock() + # two transport errors, then a not-found (resets the counter), then more errors + probe = MagicMock(side_effect=[ + RuntimeError("boom"), RuntimeError("boom"), JobNotFoundError("not yet"), + RuntimeError("boom"), RuntimeError("boom"), RuntimeError("boom"), + ]) + wd = make_watchdog(clock, probe, grace=0, interval=0, max_errors=3) + verdicts = [] + for _ in range(6): + verdicts.append(wd.observe(is_running=True)) + clock.advance(1) + assert verdicts[:5] == [WatchdogVerdict.WAITING] * 5 + assert verdicts[5] is WatchdogVerdict.DISABLED + # disabled is terminal + assert wd.observe(is_running=True) is WatchdogVerdict.DISABLED + assert probe.call_count == 6 + + +def _job_info(status): + return SimpleNamespace(status=status, start_time=None, finish_time=None, + is_completed=False) + + +def make_cloud_runner(tmp_path): + runner = CloudProverRunner( + project_root=tmp_path, + config_manager=MagicMock(), + cloud_server="production", + disable_cache=True, + ) + return runner + + +class TestWaitLoopIntegration: + def _run_wait(self, runner, prover_api, timeout=30): + async def go(): + with patch("asyncio.sleep", new=AsyncMock()): + return await runner._wait_for_job_completion_with_api( + prover_api, "https://prover.certora.com/output/1/abc", timeout + ) + return asyncio.run(go()) + + def test_stuck_preprocessing_is_cancelled_and_classified(self, tmp_path): + runner = make_cloud_runner(tmp_path) + runner.preprocessing_budget = 1 # 1s of RUNNING without treeview + runner.preprocessing_grace = 0 + runner.preprocessing_probe_interval = 0 + runner._cancel_cloud_job = AsyncMock(return_value=True) + + prover_api = MagicMock() + prover_api.get_job_info.return_value = _job_info(ProverJobStatus.RUNNING) + prover_api.get_treeview_status.side_effect = JobNotFoundError("not yet") + + t0 = time.monotonic() + outcome, _, _ = self._run_wait(runner, prover_api, timeout=60) + assert outcome is _WaitOutcome.PREPROCESSING_TIMEOUT + assert time.monotonic() - t0 < 30 # nowhere near the 60s overall timeout + runner._cancel_cloud_job.assert_awaited_once() + + def test_treeview_appearance_prevents_cancellation(self, tmp_path): + runner = make_cloud_runner(tmp_path) + runner.preprocessing_budget = 1 + runner.preprocessing_grace = 0 + runner.preprocessing_probe_interval = 0 + runner._cancel_cloud_job = AsyncMock(return_value=True) + + prover_api = MagicMock() + # RUNNING with a treeview available, then the job succeeds + prover_api.get_treeview_status.return_value = {"rules": [{"name": "r"}]} + prover_api.get_job_info.side_effect = ( + [_job_info(ProverJobStatus.RUNNING)] * 3 + + [_job_info(ProverJobStatus.SUCCEEDED)] + ) + + outcome, _, _ = self._run_wait(runner, prover_api, timeout=60) + assert outcome is _WaitOutcome.COMPLETED + runner._cancel_cloud_job.assert_not_awaited() + + def test_watchdog_disabled_by_zero_budget(self, tmp_path): + runner = make_cloud_runner(tmp_path) + runner.preprocessing_budget = 0 + prover_api = MagicMock() + prover_api.get_job_info.return_value = _job_info(ProverJobStatus.SUCCEEDED) + outcome, _, _ = self._run_wait(runner, prover_api) + assert outcome is _WaitOutcome.COMPLETED + prover_api.get_treeview_status.assert_not_called() + + +def make_preprocessing_timeout_result(tmp_path, contract="Vault"): + conf = tmp_path / "Vault.conf" + conf.write_text('{"optimistic_loop": true, "loop_iter": 3}') + job_spec = SimpleNamespace( + contract_name=contract, + phase="Sanity Test Run - warmup", + config_file=SimpleNamespace(path=conf, content_hash="deadbeef"), + ) + handle = JobHandle( + job_id="https://prover.certora.com/output/1/abc", + config_file=str(conf), + config_content_hash="deadbeef", + phase=job_spec.phase, + submitted_at=time.time(), + runner_type=RunnerType.CLOUD, + status=JobStatus.PREPROCESSING_TIMEOUT, + ) + return ProverResult( + job_handle=handle, + success=False, + report_path=None, + output_data={"job_url": handle.job_id, "preprocessing_timeout": True, + "return_code": 0}, + job_spec=job_spec, + error_message="Preprocessing timeout", + duration=100.0, + ) + + +class TestRetrySuppression: + def test_on_job_problem_skips_workarounds(self, tmp_path): + result = make_preprocessing_timeout_result(tmp_path) + sentinel = MagicMock(return_value=True) + with patch.object(job_problem_fixes, "_WORKAROUNDS", [sentinel]): + assert on_job_problem(result, MagicMock(), MagicMock()) is False + sentinel.assert_not_called() + + def test_status_round_trips_through_job_handle_serialization(self, tmp_path): + result = make_preprocessing_timeout_result(tmp_path) + restored = JobHandle.from_dict(result.job_handle.to_dict()) + assert restored.status is JobStatus.PREPROCESSING_TIMEOUT + + +class TestReporterRow: + def test_sanity_row_for_preprocessing_timeout(self, tmp_path): + from certora_autosetup.reporting.reporter import Reporter + + prover_api = MagicMock() + reporter = Reporter( + log=lambda *a, **k: None, + verbose=False, + skip_breadcrumbs=True, + reports_dir=tmp_path / "reports", + prover_api=prover_api, + ) + result = make_preprocessing_timeout_result(tmp_path) + rows = reporter._collect_sanity_rows([result], sanity_advanced=None) + assert len(rows) == 1 + assert "PREPROCESSING TIMEOUT" in rows[0].sanity_status + assert rows[0].job_url == result.job_url + prover_api.get_job_report.assert_not_called() + + def test_job_report_fetch_failure_skips_row_not_summary(self, tmp_path): + from certora_autosetup.reporting.reporter import Reporter + + prover_api = MagicMock() + prover_api.get_job_report.side_effect = RuntimeError("cancelled job, no report") + reporter = Reporter( + log=lambda *a, **k: None, + verbose=False, + skip_breadcrumbs=True, + reports_dir=tmp_path / "reports", + prover_api=prover_api, + ) + result = make_preprocessing_timeout_result(tmp_path) + result.job_handle.status = JobStatus.FAILED # generic failure, not watchdog + rows = reporter._collect_sanity_rows([result], sanity_advanced=None) + assert rows == [] + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])