fix(check-collections): report runner scan FAILED with logs on abort re-raise (SAS-13001)

soda-core/src/soda_core/check_collections/session.py

@@ -201,6 +201,7 @@ def execute_check_collections(
             constructed.append((impl, impl_class, None, yaml_source))
         except Exception as exc:
             if abort_on_first_error:
+                _report_runner_scan_failed_before_reraise(soda_cloud_impl, publish_results, logs, exc)
                 raise
             constructed.append((None, impl_class, exc, yaml_source))
 
@@ -238,6 +239,7 @@ def execute_check_collections(
                 results.append(impl.verify())
             except Exception as exc:
                 if abort_on_first_error:
+                    _report_runner_scan_failed_before_reraise(soda_cloud_impl, publish_results, logs, exc)
                     raise
                 results.append(impl_class.build_error_result(yaml_source, exc))
 
@@ -398,6 +400,43 @@ def execute_check_collections(
     return CheckCollectionSessionResult(results)
 
 
+def _report_runner_scan_failed_before_reraise(
+    soda_cloud_impl: Optional[SodaCloud],
+    publish_results: bool,
+    logs: Optional[Logs],
+    exc: BaseException,
+) -> None:
+    """Best-effort: mark the runner's (still-PENDING) Cloud scan FAILED with the captured logs
+    just before an ``abort_on_first_error`` re-raise.
+
+    A single-contract (runner) scan aborts on the first construction/verify exception, which
+    re-raises out of ``execute_check_collections`` before phase 3's combined upload runs — the only
+    place that otherwise ships logs / marks the scan FAILED. Without this the exception propagates to
+    the CLI (exit 3) and the launcher only writes it to pod logs, so the Cloud scan record shows
+    FAILED with no logs and the failure is undiagnosable (SAS-13001). Reporting here keeps the
+    historical re-raise contract intact while making the failure visible in Cloud.
+
+    No-op when there is no cloud client, no publish opt-in, or no runner scan id (ad-hoc runs).
+    Never masks the original exception: any error while reporting is swallowed so the caller still
+    sees the real failure. The shared ``logs`` gatherer holds every construction/verify record (each
+    impl is built with ``logs=logs``), so its records are exactly what should reach Cloud.
+    """
+    if soda_cloud_impl is None or not publish_results:
+        return
+    soda_scan_id: Optional[str] = EnvConfigHelper().soda_scan_id
+    if not soda_scan_id:
+        return
+    try:
+        log_records = logs.get_log_records() if logs is not None else None
+        soda_cloud_impl.mark_scan_as_failed(
+            scan_id=soda_scan_id,
+            logs=log_records,
+            exc=exc if isinstance(exc, Exception) else None,
+        )
+    except Exception:
+        logger.debug("Could not report runner scan as failed before re-raising", exc_info=True)
+
+
 def _parse_data_timestamp(value: Optional[Union[str, datetime]]) -> Optional[datetime]:
     """Coerce a ``data_timestamp`` value to ``Optional[datetime]``.
 

soda-tests/tests/unit/test_contract_marks_scan_failed_on_connection_error.py

@@ -12,8 +12,10 @@
 
 from unittest.mock import patch
 
+import pytest
 from helpers.mock_soda_cloud import MockSodaCloud
 from soda_core.common.data_source_impl import DataSourceImpl
+from soda_core.common.logging_constants import soda_logger
 from soda_core.common.yaml import ContractYamlSource, DataSourceYamlSource
 from soda_core.contracts.contract_verification import ContractVerificationSession
 from soda_core.contracts.impl.contract_verification_impl import ContractImpl
@@ -145,3 +147,44 @@
         "Ad-hoc run has no scan id to mark failed; mark_scan_as_failed must not be used. "
         f"Requests seen: {request_types}"
     )
+
+
+def test_uncaught_exception_during_verify_marks_scan_failed_with_logs(monkeypatch):
+    """A single-contract (runner) scan that raises an *uncaught* exception during verify aborts
+    and re-raises before phase 3's combined upload runs. The engine must still report the runner
+    scan as FAILED with the captured logs, otherwise the Cloud scan record shows no logs and the
+    failure is undiagnosable (SAS-13001)."""
+    monkeypatch.setenv("SODA_SCAN_ID", "scan-under-test")
+
+    def _boom(self, *args, **kwargs):
+        soda_logger.error("Boom: could not build check collection")
+        raise RuntimeError("verify exploded")
+
+    monkeypatch.setattr(ContractImpl, "verify", _boom)
+
+    data_source_impl = DataSourceImpl.from_yaml_source(DataSourceYamlSource.from_str(_DATA_SOURCE_YAML))
+    mock_cloud = MockSodaCloud()
+    mock_cloud._upload_contract_yaml_file = lambda *args, **kwargs: "contract-file-id"
+
+    # The abort-on-first-error contract path re-raises the exception verbatim.
+    with pytest.raises(RuntimeError, match="verify exploded"):
+        ContractVerificationSession.execute(
+            contract_yaml_sources=[ContractYamlSource.from_str(_CONTRACT_YAML)],
+            data_source_impls=[data_source_impl],
+            soda_cloud_impl=mock_cloud,
+            soda_cloud_publish_results=True,
+        )
+
+    request_types = [r.json.get("type") for r in mock_cloud.requests if isinstance(r.json, dict)]
+    mark_requests = [
+        r.json
+        for r in mock_cloud.requests
+        if isinstance(r.json, dict) and r.json.get("type") == "sodaCoreMarkScanFailed"
+    ]
+    assert mark_requests, (
+        "A scan that raised an uncaught exception during verify must be marked FAILED so the "
+        f"failure is visible in Cloud. Requests seen: {request_types}"
+    )
+    assert mark_requests[0].get("scanId") == "scan-under-test"
+    # The captured engine logs must be shipped, not an empty payload — that is the whole point.
+    assert mark_requests[0].get("logs"), "mark-scan-failed must carry the captured engine logs, not an empty payload"