autosetup: preprocessing watchdog — cancel cloud jobs stuck in prover preprocessing

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)

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,18 +1012,30 @@ 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
 
         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,19 +1071,41 @@ 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(
                             f"❌ Job completed with unrecognized status '{job_info.status}' - "
                             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

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):

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

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."""