Bind failed import audits to Dolt state

Author: TonisOrmisson

src/openstatspec/sql/wide.py

@@ -89,10 +89,14 @@ def binary64_type() -> Float:
 def _record_failed_preflight(
     *, engine: Any, normative: Any, operation_id: str, source_name: str,
     source_format: str, variable_count: int, profile_name: str,
-    error: Exception,
+    active: Mapping[str, Any], error: Exception,
 ) -> None:
-    """Persist a failed preflight in the normative audit catalog."""
-    with engine.begin() as connection:
+    """Persist a failed preflight in the bound normative audit catalog."""
+    with _bound_catalog_transaction(
+        engine=engine, profile_name=profile_name, active=active,
+        audit_relations={normative.operation.name, normative.fidelity_event.name},
+        phase="import preflight failure",
+    ) as connection:
         _verify_normative_catalog(connection, normative)
         failed_at = datetime.now(UTC).replace(tzinfo=None)
         record_normative_operation(
@@ -832,7 +836,8 @@ def create_wide_dataset(
         _record_failed_preflight(
             engine=engine, normative=normative, operation_id=operation_id,
             source_name=source_name, source_format=source_format,
-            variable_count=len(variables), profile_name=profile.name, error=error,
+            variable_count=len(variables), profile_name=profile.name,
+            active=active_connection, error=error,
         )
         raise
 

tests/test_catalog_persistence_review.py

@@ -8,6 +8,7 @@
 
 from openstatspec.core import UnsupportedOperationError
 from openstatspec.sql import wide
+from openstatspec.sql.profiles import TargetCapabilityExceededError
 
 
 def _variables():
@@ -91,6 +92,38 @@ def observe(**kwargs):
     assert phases == ["catalog initialization", "import"]
 
 
+def test_failed_import_preflight_routes_audit_through_binding_guard(
+    tmp_path, monkeypatch,
+):
+    database_url = f"sqlite:///{tmp_path / 'failed-preflight.sqlite'}"
+    real_bound_transaction = wide._bound_catalog_transaction
+    phases = []
+
+    @contextmanager
+    def observe(**kwargs):
+        phases.append(kwargs["phase"])
+        with real_bound_transaction(**kwargs) as connection:
+            yield connection
+
+    monkeypatch.setattr(wide, "_bound_catalog_transaction", observe)
+    variables = _variables()
+    variables[0].update({
+        "storage_kind": "numeric", "string_width": None,
+        "format": "F8.2", "alignment": "right",
+    })
+    with pytest.raises(TargetCapabilityExceededError):
+        wide.create_wide_dataset(
+            database_url=database_url,
+            dataset_id="invalid",
+            source_name="invalid.sav",
+            source_format="SAV",
+            rows=[{"name": "not-a-number"}],
+            variables=variables,
+        )
+
+    assert phases == ["catalog initialization", "import preflight failure"]
+
+
 def test_export_audit_mutations_route_through_binding_guard(tmp_path, monkeypatch):
     database_url = f"sqlite:///{tmp_path / 'bound-export.sqlite'}"
     _create(database_url)