Fetch UnsatCoreTAC files via unsat_core_map.json instead of the whole archive

src/prover_output_utility/api/aws_data_fetcher.py

@@ -7,6 +7,7 @@
 Uses AWS SigV4 authentication to access Certora API through Lambda.
 """
 
+import json
 import os
 from typing import Any, Dict, List, Optional, cast
 
@@ -199,37 +200,44 @@ def fetch_tree_view_data(
             raise ProverAPIError(f"Failed to fetch tree-view data for {job_identifier}: {e}")
 
     def fetch_statsdata(self, job_identifier: str) -> Dict[str, Any]:
-        """
-        Fetch statsdata.json for a job via Lambda.
+        """Fetch statsdata.json for a job via Lambda."""
+        try:
+            return cast(Dict[str, Any], json.loads(self.fetch_output_file(job_identifier, "statsdata.json")))
+        except json.JSONDecodeError as e:
+            raise ProverAPIError(f"Failed to parse statsdata.json for {job_identifier}: {e}")
+
+    def fetch_output_file(self, job_identifier: str, rel_path: str) -> str:
+        """Fetch the raw text of a Reports/-relative output file (e.g. unsat_core_map.json).
 
         Args:
-            job_identifier: Job ID to fetch
+            job_identifier: Job ID
+            rel_path: Path of the file relative to the job's Reports/ directory
 
         Returns:
-            Stats data from statsdata.json
+            The file contents as text
 
         Raises:
             AuthenticationError: If authentication fails
-            JobNotFoundError: If job is not found
-            ProverAPIError: If API call fails
+            JobNotFoundError: If the job or file is not found
+            ProverAPIError: If the API call fails
         """
-        endpoint = f"{self.base_url}/v1/domain/jobs/{job_identifier}/f/statsdata.json"
+        endpoint = f"{self.base_url}/v1/domain/jobs/{job_identifier}/f/{rel_path}"
 
         try:
             response = self._make_signed_request("GET", endpoint)
 
             if response.status_code == 200:
-                return cast(Dict[str, Any], response.json())
+                return response.text
             elif response.status_code == 401:
                 raise AuthenticationError("Authentication failed - check AWS credentials")
             elif response.status_code == 404:
-                raise JobNotFoundError(f"Stats data not found for job {job_identifier}")
+                raise JobNotFoundError(f"Output file not found for job {job_identifier}: {rel_path}")
             else:
                 response.raise_for_status()
-                return cast(Dict[str, Any], response.json())
+                return response.text
 
         except requests.exceptions.RequestException as e:
-            raise ProverAPIError(f"Failed to fetch stats data for {job_identifier}: {e}")
+            raise ProverAPIError(f"Failed to fetch output file {rel_path} for {job_identifier}: {e}")
 
     def fetch_outputs(self, job_identifier: str) -> bytes:
         """

src/prover_output_utility/api/base_data_fetcher.py

@@ -159,6 +159,25 @@ def fetch_source_file_content(self, job_identifier: str, path: str) -> str:
         """
         pass
 
+    @abstractmethod
+    def fetch_output_file(self, job_identifier: str, rel_path: str) -> str:
+        """
+        Fetch the raw text of a file under the job's Reports/ output dir.
+
+        Args:
+            job_identifier: Job identifier (job ID for remote, emv path for local)
+            rel_path: Reports/-relative filename (e.g. "unsat_core_map.json", "UnsatCoreTAC-....txt")
+
+        Returns:
+            Raw file content as a string
+
+        Raises:
+            AuthenticationError: If authentication fails (remote fetchers)
+            JobNotFoundError: If the file is not found
+            ProverAPIError: If fetch fails
+        """
+        pass
+
     @abstractmethod
     def fetch_alert_report(self, job_identifier: str) -> List[Dict[str, Any]]:
         """

src/prover_output_utility/api/data_fetcher.py

@@ -6,6 +6,7 @@
 Low-level data fetching utilities for the Prover API.
 """
 import functools
+import json
 from typing import Any, Callable, Dict, List, Optional, TypeVar, cast
 
 import requests
@@ -255,39 +256,46 @@ def fetch_tree_view_data(
         except requests.exceptions.RequestException as e:
             raise ProverAPIError(f"Failed to fetch tree-view data for {job_identifier}: {e}")
 
-    @handle_token_expiration
     def fetch_statsdata(self, job_identifier: str) -> Dict[str, Any]:
-        """
-        Fetch statsdata.json for a job.
+        """Fetch statsdata.json for a job."""
+        try:
+            return cast(Dict[str, Any], json.loads(self.fetch_output_file(job_identifier, "statsdata.json")))
+        except json.JSONDecodeError as e:
+            raise ProverAPIError(f"Failed to parse statsdata.json for {job_identifier}: {e}")
+
+    @handle_token_expiration
+    def fetch_output_file(self, job_identifier: str, rel_path: str) -> str:
+        """Fetch the raw text of a Reports/-relative output file (e.g. unsat_core_map.json).
 
         Args:
-            job_identifier: Job ID to fetch
+            job_identifier: Job ID
+            rel_path: Path of the file relative to the job's Reports/ directory
 
         Returns:
-            Stats data from statsdata.json
+            The file contents as text
 
         Raises:
             AuthenticationError: If authentication fails
-            JobNotFoundError: If job is not found
-            ProverAPIError: If API call fails
+            JobNotFoundError: If the job or file is not found
+            ProverAPIError: If the API call fails
         """
-        endpoint = f"{self.api_base_url}/v1/domain/jobs/{job_identifier}/f/statsdata.json"
+        endpoint = f"{self.api_base_url}/v1/domain/jobs/{job_identifier}/f/{rel_path}"
 
         try:
             response = self.session.get(endpoint)
 
             if response.status_code == 200:
-                return cast(Dict[str, Any], response.json())
+                return response.text
             elif response.status_code == 401:
                 raise AuthenticationError("Authentication failed - check CERTORAKEY")
             elif response.status_code == 404:
-                raise JobNotFoundError(f"Stats data not found for job {job_identifier}")
+                raise JobNotFoundError(f"Output file not found for job {job_identifier}: {rel_path}")
             else:
                 response.raise_for_status()
-                return cast(Dict[str, Any], response.json())
+                return response.text
 
         except requests.exceptions.RequestException as e:
-            raise ProverAPIError(f"Failed to fetch stats data for {job_identifier}: {e}")
+            raise ProverAPIError(f"Failed to fetch output file {rel_path} for {job_identifier}: {e}")
 
     def fetch_outputs(self, job_identifier: str) -> bytes:
         """

src/prover_output_utility/api/local_data_fetcher.py

@@ -200,38 +200,11 @@ def cancel_jobs(self, job_ids: List[str]) -> Dict[str, Any]:
         raise ProverAPIError("cancel_jobs is not supported for local prover outputs")
 
     def fetch_statsdata(self, job_identifier: str) -> Dict[str, Any]:
-        """
-        Fetch statsdata.json from local emv-* folder.
-
-        Args:
-            job_identifier: Path to the emv-* folder
-
-        Returns:
-            Stats data from statsdata.json
-
-        Raises:
-            JobNotFoundError: If statsdata.json is not found
-            ProverAPIError: If local files cannot be read
-        """
-        emv_path = os.path.abspath(job_identifier)
-
-        if not os.path.exists(emv_path):
-            raise JobNotFoundError(f"Local prover output path not found: {job_identifier}")
-
+        """Fetch statsdata.json from local emv-* folder."""
         try:
-            # Look for statsdata.json in Reports directory
-            statsdata_path = os.path.join(emv_path, "Reports", "statsdata.json")
-
-            if not os.path.exists(statsdata_path):
-                raise JobNotFoundError(f"Stats data not found for local job {job_identifier}")
-
-            with open(statsdata_path, "r") as f:
-                return json.load(f)
-
+            return json.loads(self.fetch_output_file(job_identifier, "statsdata.json"))
         except json.JSONDecodeError as e:
             raise ProverAPIError(f"Failed to parse statsdata.json: {e}")
-        except Exception as e:
-            raise ProverAPIError(f"Failed to read stats data from {job_identifier}: {e}")
 
     def fetch_outputs(self, job_identifier: str) -> bytes:
         """
@@ -266,6 +239,32 @@ def fetch_source_file_content(self, job_identifier: str, path: str) -> str:
         """
         raise ProverAPIError("fetch_source_file_content is not supported for local prover outputs")
 
+    def fetch_output_file(self, job_identifier: str, rel_path: str) -> str:
+        """Read a Reports/-relative output file from a local emv-* folder.
+
+        Args:
+            job_identifier: Path to the emv-* folder
+            rel_path: Path of the file relative to the folder's Reports/ directory
+
+        Returns:
+            The file contents as text
+
+        Raises:
+            JobNotFoundError: If the folder or file is not found
+            ProverAPIError: If the file cannot be read
+        """
+        emv_path = os.path.abspath(job_identifier)
+        if not os.path.exists(emv_path):
+            raise JobNotFoundError(f"Local prover output path not found: {job_identifier}")
+        file_path = os.path.join(emv_path, "Reports", rel_path)
+        if not os.path.exists(file_path):
+            raise JobNotFoundError(f"Output file not found: {rel_path}")
+        try:
+            with open(file_path, "r") as f:
+                return f.read()
+        except OSError as e:
+            raise ProverAPIError(f"Failed to read output file {rel_path}: {e}")
+
     def fetch_alert_report(self, job_identifier: str) -> List[Dict[str, Any]]:
         """
         Fetch alert report (alertReport.json) from local emv-* folder.

src/prover_output_utility/api/main.py

@@ -22,7 +22,7 @@
 from ..auth import ProverAuth
 from ..aws_auth import AWSAuth
 from ..breadcrumb import BreadcrumbParser
-from ..exceptions import ProverAPIError
+from ..exceptions import JobNotFoundError, ProverAPIError
 from ..job_report import JobAnalyzer, JobReport
 from ..models import (
     BreadcrumbInfo,
@@ -1135,12 +1135,27 @@ def _fetch_outputs_cached(self, job_identifier: str) -> bytes:
         return tar_content
 
     def extract_unsat_core_files(self, job_input: str, dest_dir: Path) -> List[Path]:
-        """Extract UnsatCoreTAC*.txt files from the job output tar into dest_dir.
+        """Extract UnsatCoreTAC*.txt files into dest_dir.
 
-        No filtering is applied — the caller is responsible for any filtering.
+        Uses unsat_core_map.json to fetch only the referenced files; falls back to the
+        full output tar when the map is absent. No filtering is applied — the caller is
+        responsible for any filtering.
 
         Returns list of extracted file paths.
         """
+        dest_dir.mkdir(parents=True, exist_ok=True)
+        core_map = self.unsat_core_map(job_input)
+        if core_map:
+            filenames = sorted({f for files in core_map.values() for f in files})
+            extracted: List[Path] = []
+            for name in filenames:
+                file_dest = dest_dir / Path(name).name
+                file_dest.write_text(self.fetch_output_file(job_input, name), encoding="utf-8")
+                extracted.append(file_dest)
+            return extracted
+        return self._extract_unsat_core_files_from_tar(job_input, dest_dir)
+
+    def _extract_unsat_core_files_from_tar(self, job_input: str, dest_dir: Path) -> List[Path]:
         job_identifier = self._extract_job_identifier(job_input)
         dest_dir.mkdir(parents=True, exist_ok=True)
         tar_content = self._fetch_outputs_cached(job_identifier)
@@ -1159,6 +1174,30 @@ def extract_unsat_core_files(self, job_input: str, dest_dir: Path) -> List[Path]
                         extracted.append(file_dest)
         return extracted
 
+    def fetch_output_file(self, job_input: str, rel_path: str) -> str:
+        """Fetch the raw text of a Reports/-relative output file (e.g. 'unsat_core_map.json')."""
+        job_identifier = self._extract_job_identifier(job_input)
+        return self.data_fetcher.fetch_output_file(job_identifier, rel_path)
+
+    def unsat_core_map(self, job_input: str) -> Dict[str, List[str]]:
+        """The job's `{ ruleId -> [UnsatCoreTAC .txt filenames] }` map, or {} if the job has none."""
+        try:
+            content = self.fetch_output_file(job_input, "unsat_core_map.json")
+        except JobNotFoundError:
+            return {}
+        try:
+            return json.loads(content)
+        except json.JSONDecodeError as e:
+            raise ProverAPIError(f"Failed to parse unsat_core_map.json for {job_input}: {e}")
+
+    def unsat_core_filenames(self, job_input: str, rule_id: str) -> List[str]:
+        """UnsatCoreTAC .txt filenames for a rule (by its treeView ruleId); [] if none."""
+        return self.unsat_core_map(job_input).get(rule_id, [])
+
+    def read_unsat_cores(self, job_input: str, rule_id: str) -> List[str]:
+        """Contents of a rule's UnsatCoreTAC .txt dumps (by its treeView ruleId)."""
+        return [self.fetch_output_file(job_input, name) for name in self.unsat_core_filenames(job_input, rule_id)]
+
     def extract_certora_sources(self, job_input: str, dest_dir: Path) -> None:
         """Extract the .certora_sources tree from the job output tar into dest_dir.
 

src/prover_output_utility/api/tree_parser.py

@@ -285,6 +285,7 @@ def _create_violation_info(
         return CheckResult(
             rule_name=self._get_rule_name_from_context(context),
             method_name=method_name,  # Use the explicitly tracked method name
+            rule_id=node.get("ruleId"),
             contract_name=contract_name,
             method_only=method_only,
             assert_message=assert_message,

src/prover_output_utility/models.py

@@ -31,6 +31,7 @@ class NodeStatus(str, Enum):
 
     VIOLATED = "VIOLATED"
     VERIFIED = "VERIFIED"
+    SANITY_FAILED = "SANITY_FAILED"
     TIMEOUT = "TIMEOUT"
     ERROR = "ERROR"
     RUNNING = "RUNNING"
@@ -221,6 +222,7 @@ class CheckResult:
     node_type: NodeType
     contract_name: Optional[str] = None
     method_only: Optional[str] = None
+    rule_id: Optional[str] = None
     ui_id: Optional[str] = None
     output_files: List[str] = field(default_factory=list)
     debug_trace_file: Optional[str] = None
@@ -255,6 +257,7 @@ def to_dict(self) -> Dict[str, Any]:
         return {
             "rule_name": self.rule_name,
             "method_name": self.method_name,
+            "rule_id": self.rule_id,
             "contract_name": self.contract_name,
             "method_only": self.method_only,
             "assert_message": self.assert_message,