From a920732091ad76a8e70f495d2f066074d2356db3 Mon Sep 17 00:00:00 2001 From: jar-ben Date: Thu, 16 Jul 2026 14:53:39 +0200 Subject: [PATCH 1/3] Fetch UnsatCoreTAC files via unsat_core_map.json instead of the whole output tar --- .../api/aws_data_fetcher.py | 28 +++++------- .../api/base_data_fetcher.py | 18 ++++++++ src/prover_output_utility/api/data_fetcher.py | 30 +++++-------- .../api/local_data_fetcher.py | 45 ++++++------------- src/prover_output_utility/api/main.py | 42 +++++++++++++++-- src/prover_output_utility/api/tree_parser.py | 1 + src/prover_output_utility/models.py | 3 ++ 7 files changed, 95 insertions(+), 72 deletions(-) diff --git a/src/prover_output_utility/api/aws_data_fetcher.py b/src/prover_output_utility/api/aws_data_fetcher.py index 6653233..0110e3d 100644 --- a/src/prover_output_utility/api/aws_data_fetcher.py +++ b/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,28 @@ 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. - - Args: - job_identifier: Job ID to fetch + """Fetch statsdata.json for a job via Lambda.""" + return cast(Dict[str, Any], json.loads(self.fetch_output_file(job_identifier, "statsdata.json"))) - Returns: - Stats data from statsdata.json - - Raises: - AuthenticationError: If authentication fails - JobNotFoundError: If job is not found - ProverAPIError: If API call fails - """ - endpoint = f"{self.base_url}/v1/domain/jobs/{job_identifier}/f/statsdata.json" + 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).""" + 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: """ diff --git a/src/prover_output_utility/api/base_data_fetcher.py b/src/prover_output_utility/api/base_data_fetcher.py index 920d206..ef755e9 100644 --- a/src/prover_output_utility/api/base_data_fetcher.py +++ b/src/prover_output_utility/api/base_data_fetcher.py @@ -159,6 +159,24 @@ 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: + 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]]: """ diff --git a/src/prover_output_utility/api/data_fetcher.py b/src/prover_output_utility/api/data_fetcher.py index 2049cea..cea2e54 100644 --- a/src/prover_output_utility/api/data_fetcher.py +++ b/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,30 @@ 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. - - Args: - job_identifier: Job ID to fetch + """Fetch statsdata.json for a job.""" + return cast(Dict[str, Any], json.loads(self.fetch_output_file(job_identifier, "statsdata.json"))) - Returns: - Stats data from statsdata.json - - Raises: - AuthenticationError: If authentication fails - JobNotFoundError: If job is not found - ProverAPIError: If API call fails - """ - endpoint = f"{self.api_base_url}/v1/domain/jobs/{job_identifier}/f/statsdata.json" + @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).""" + 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: """ diff --git a/src/prover_output_utility/api/local_data_fetcher.py b/src/prover_output_utility/api/local_data_fetcher.py index da3e419..02c1f3e 100644 --- a/src/prover_output_utility/api/local_data_fetcher.py +++ b/src/prover_output_utility/api/local_data_fetcher.py @@ -200,38 +200,8 @@ 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}") - - 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) - - 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}") + """Fetch statsdata.json from local emv-* folder.""" + return json.loads(self.fetch_output_file(job_identifier, "statsdata.json")) def fetch_outputs(self, job_identifier: str) -> bytes: """ @@ -266,6 +236,17 @@ 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.""" + 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}") + with open(file_path, "r") as f: + return f.read() + def fetch_alert_report(self, job_identifier: str) -> List[Dict[str, Any]]: """ Fetch alert report (alertReport.json) from local emv-* folder. diff --git a/src/prover_output_utility/api/main.py b/src/prover_output_utility/api/main.py index 3fbedb0..1f1467b 100644 --- a/src/prover_output_utility/api/main.py +++ b/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,27 @@ 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 {} + return json.loads(content) + + 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. diff --git a/src/prover_output_utility/api/tree_parser.py b/src/prover_output_utility/api/tree_parser.py index 90f31bf..49d38a4 100644 --- a/src/prover_output_utility/api/tree_parser.py +++ b/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, diff --git a/src/prover_output_utility/models.py b/src/prover_output_utility/models.py index 443cd7d..4b90dbc 100644 --- a/src/prover_output_utility/models.py +++ b/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, From 9dbcd6cb8e5c4c3d8da26fefb7223be19e548a76 Mon Sep 17 00:00:00 2001 From: jar-ben Date: Thu, 16 Jul 2026 21:23:03 +0200 Subject: [PATCH 2/3] Restore exception contract dropped in the statsdata refactor --- .../api/aws_data_fetcher.py | 20 ++++++++++++-- .../api/base_data_fetcher.py | 1 + src/prover_output_utility/api/data_fetcher.py | 20 ++++++++++++-- .../api/local_data_fetcher.py | 26 ++++++++++++++++--- 4 files changed, 59 insertions(+), 8 deletions(-) diff --git a/src/prover_output_utility/api/aws_data_fetcher.py b/src/prover_output_utility/api/aws_data_fetcher.py index 0110e3d..8343aaa 100644 --- a/src/prover_output_utility/api/aws_data_fetcher.py +++ b/src/prover_output_utility/api/aws_data_fetcher.py @@ -201,10 +201,26 @@ def fetch_tree_view_data( def fetch_statsdata(self, job_identifier: str) -> Dict[str, Any]: """Fetch statsdata.json for a job via Lambda.""" - return cast(Dict[str, Any], json.loads(self.fetch_output_file(job_identifier, "statsdata.json"))) + 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).""" + """Fetch the raw text of a Reports/-relative output file (e.g. unsat_core_map.json). + + Args: + job_identifier: Job ID + rel_path: Path of the file relative to the job's Reports/ directory + + Returns: + The file contents as text + + Raises: + AuthenticationError: If authentication 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/{rel_path}" try: diff --git a/src/prover_output_utility/api/base_data_fetcher.py b/src/prover_output_utility/api/base_data_fetcher.py index ef755e9..6e95b3c 100644 --- a/src/prover_output_utility/api/base_data_fetcher.py +++ b/src/prover_output_utility/api/base_data_fetcher.py @@ -172,6 +172,7 @@ def fetch_output_file(self, job_identifier: str, rel_path: str) -> str: Raw file content as a string Raises: + AuthenticationError: If authentication fails (remote fetchers) JobNotFoundError: If the file is not found ProverAPIError: If fetch fails """ diff --git a/src/prover_output_utility/api/data_fetcher.py b/src/prover_output_utility/api/data_fetcher.py index cea2e54..4fa6aca 100644 --- a/src/prover_output_utility/api/data_fetcher.py +++ b/src/prover_output_utility/api/data_fetcher.py @@ -258,11 +258,27 @@ def fetch_tree_view_data( def fetch_statsdata(self, job_identifier: str) -> Dict[str, Any]: """Fetch statsdata.json for a job.""" - return cast(Dict[str, Any], json.loads(self.fetch_output_file(job_identifier, "statsdata.json"))) + 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).""" + """Fetch the raw text of a Reports/-relative output file (e.g. unsat_core_map.json). + + Args: + job_identifier: Job ID + rel_path: Path of the file relative to the job's Reports/ directory + + Returns: + The file contents as text + + Raises: + AuthenticationError: If authentication 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/{rel_path}" try: diff --git a/src/prover_output_utility/api/local_data_fetcher.py b/src/prover_output_utility/api/local_data_fetcher.py index 02c1f3e..0e0d2e4 100644 --- a/src/prover_output_utility/api/local_data_fetcher.py +++ b/src/prover_output_utility/api/local_data_fetcher.py @@ -201,7 +201,10 @@ def cancel_jobs(self, job_ids: List[str]) -> Dict[str, Any]: def fetch_statsdata(self, job_identifier: str) -> Dict[str, Any]: """Fetch statsdata.json from local emv-* folder.""" - return json.loads(self.fetch_output_file(job_identifier, "statsdata.json")) + try: + 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}") def fetch_outputs(self, job_identifier: str) -> bytes: """ @@ -237,15 +240,30 @@ 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.""" + """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}") - with open(file_path, "r") as f: - return f.read() + 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]]: """ From 0ae3909328e64c3d3785b7ff82f8276286e4c33a Mon Sep 17 00:00:00 2001 From: jar-ben Date: Fri, 17 Jul 2026 09:41:58 +0200 Subject: [PATCH 3/3] adding JSONDecodeError guard --- src/prover_output_utility/api/main.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/prover_output_utility/api/main.py b/src/prover_output_utility/api/main.py index 1f1467b..1c06476 100644 --- a/src/prover_output_utility/api/main.py +++ b/src/prover_output_utility/api/main.py @@ -1185,7 +1185,10 @@ def unsat_core_map(self, job_input: str) -> Dict[str, List[str]]: content = self.fetch_output_file(job_input, "unsat_core_map.json") except JobNotFoundError: return {} - return json.loads(content) + 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."""