Verify Dolt diffs and workflow triggers

Author: TonisOrmisson

src/openstatspec/sql/catalog_verification.py

@@ -205,7 +205,11 @@ def verify_catalog_relations(
         if name
     )
 
-    from .workflow import PROFILE_ID, PROFILE_SCHEMA_VERSION, workflow_catalog
+    from .workflow import (
+        PROFILE_ID, PROFILE_SCHEMA_VERSION, TransformationError,
+        _assert_trigger_definitions, _derived_trigger_sql,
+        _validate_workflow_schema, workflow_catalog,
+    )
 
     workflow = workflow_catalog(MetaData())
     workflow_tables = {table.name: table for table in workflow.all()}
@@ -216,6 +220,10 @@ def verify_catalog_relations(
             for table in workflow_tables.values()
         ):
             _reject(workflow_present)
+        try:
+            _validate_workflow_schema(connection, workflow)
+        except TransformationError:
+            _reject(workflow_present)
         identity_rows = connection.execute(
             select(workflow.transformation_profile_identity)
         ).mappings().all()
@@ -228,6 +236,7 @@ def verify_catalog_relations(
             _reject({workflow.transformation_profile_identity.name})
         owned_tables.update(workflow_tables)
         for row in connection.execute(select(
+            workflow.derived_dataset.c.derived_dataset_id,
             workflow.derived_dataset.c.physical_relation_name,
             workflow.derived_dataset.c.output_mode,
         )).mappings():
@@ -236,6 +245,16 @@ def verify_catalog_relations(
                 owned_views.add(name)
             else:
                 owned_tables.add(name)
+                try:
+                    _assert_trigger_definitions(
+                        connection,
+                        _derived_trigger_sql(
+                            connection, str(row["derived_dataset_id"]), name,
+                        ),
+                        code="derived_corrupt",
+                    )
+                except TransformationError:
+                    _reject({name})
 
     from .inplace_transform import apply_audit_catalog
 

src/openstatspec/sql/wide.py

@@ -474,23 +474,33 @@ def _dolt_evidence_block(
 def _capture_dolt_state(
     connection: Any, *, profile_name: str, audit_relations: set[str],
 ) -> dict[str, Any] | None:
-    """Capture the Dolt working-set identity without mutating it."""
-    del audit_relations
+    """Capture and classify the Dolt working set without mutating it."""
+    audit_relations = {str(name) for name in audit_relations}
     if profile_name != "dolt":
         return None
     identity = connection.exec_driver_sql(
         "SELECT DATABASE() AS database_name, ACTIVE_BRANCH() AS active_branch, "
         "DOLT_HASHOF('HEAD') AS head_hash"
     ).mappings().one()
+    status_rows = connection.exec_driver_sql(
+        "SELECT table_name, staged, status FROM dolt_status "
+        "ORDER BY table_name, staged, status"
+    ).mappings().all()
     state = {
         "database": str(identity["database_name"]).strip(),
         "active_branch": str(identity["active_branch"]).strip(),
         "head": str(identity["head_hash"]).strip(),
         "status": _dolt_evidence_block(
-            connection.exec_driver_sql(
-                "SELECT table_name, staged, status FROM dolt_status "
-                "ORDER BY table_name, staged, status"
-            ).mappings().all(),
+            status_rows, expected_keys=("table_name", "staged", "status"),
+        ),
+        "diff_summaries": {},
+    }
+    unrelated = {
+        "status": _dolt_evidence_block(
+            (
+                row for row in status_rows
+                if str(dict(row).get("table_name") or "") not in audit_relations
+            ),
             expected_keys=("table_name", "staged", "status"),
         ),
         "diff_summaries": {},
@@ -500,18 +510,33 @@ def _capture_dolt_state(
         ("head_to_staged", "HEAD", "STAGED"),
         ("staged_to_working", "STAGED", "WORKING"),
     ):
+        rows = connection.exec_driver_sql(
+            "SELECT from_table_name, to_table_name, diff_type, "
+            "data_change, schema_change "
+            f"FROM DOLT_DIFF_SUMMARY('{left}', '{right}') "
+            "ORDER BY from_table_name, to_table_name, diff_type"
+        ).mappings().all()
+        expected_keys = (
+            "from_table_name", "to_table_name", "diff_type",
+            "data_change", "schema_change",
+        )
         state["diff_summaries"][label] = _dolt_evidence_block(
-            connection.exec_driver_sql(
-                "SELECT from_table_name, to_table_name, diff_type, "
-                "data_change, schema_change "
-                f"FROM DOLT_DIFF_SUMMARY('{left}', '{right}') "
-                "ORDER BY from_table_name, to_table_name, diff_type"
-            ).mappings().all(),
-            expected_keys=(
-                "from_table_name", "to_table_name", "diff_type",
-                "data_change", "schema_change",
+            rows, expected_keys=expected_keys,
+        )
+        unrelated["diff_summaries"][label] = _dolt_evidence_block(
+            (
+                row for row in rows
+                if not {
+                    str(value) for value in (
+                        dict(row).get("from_table_name"),
+                        dict(row).get("to_table_name"),
+                    ) if value
+                } <= audit_relations
             ),
+            expected_keys=expected_keys,
         )
+    state["unrelated_working_set"] = unrelated
+    state["unrelated_sha256"] = _canonical_sha256(unrelated)
     state["snapshot_sha256"] = _canonical_sha256(state)
     return state
 
@@ -545,9 +570,11 @@ def _require_dolt_success_identity(
         return
     if before is None or after is None or any(
         before[key] != after[key] for key in ("database", "active_branch", "head")
-    ):
+    ) or (
+        "unrelated_sha256" in before or "unrelated_sha256" in after
+    ) and before.get("unrelated_sha256") != after.get("unrelated_sha256"):
         raise UnsupportedOperationError(
-            f"Dolt database/branch/HEAD changed during {phase}."
+            f"Dolt identity or unrelated working-set state changed during {phase}."
         )
 
 
@@ -676,10 +703,7 @@ def initialize_wide_catalog(
     )
     engine = create_engine(database_url)
     normative = normative_catalog(MetaData())
-    audit_relations = {
-        normative.fidelity_event.name,
-        normative.operation.name,
-    }
+    audit_relations = {table.name for table in normative.all()}
     with _bound_catalog_transaction(
         engine=engine, profile_name=profile.name, active=active,
         audit_relations=audit_relations, phase="catalog initialization",
@@ -780,10 +804,7 @@ def create_wide_dataset(
     )
     engine = create_engine(database_url)
     normative = normative_catalog(MetaData())
-    audit_relations = {
-        normative.fidelity_event.name,
-        normative.operation.name,
-    }
+    audit_relations = {table.name for table in normative.all()}
     with _bound_catalog_transaction(
         engine=engine, profile_name=profile.name, active=active_connection,
         audit_relations=audit_relations, phase="catalog initialization",
@@ -851,7 +872,7 @@ def create_wide_dataset(
         namespace_owned = True
         with _bound_catalog_transaction(
             engine=engine, profile_name=profile.name, active=active_connection,
-            audit_relations=audit_relations, phase="import",
+            audit_relations={*audit_relations, data_table.name}, phase="import",
         ) as connection:
             if connection.execute(
                 select(normative.dataset.c.dataset_id).where(or_(

tests/test_review_transaction_boundary.py

@@ -57,6 +57,20 @@ def capture(connection, **_kwargs):
     assert transaction_states == [False, True]
 
 
+def test_dolt_completion_rejects_unrelated_working_set_changes():
+    before = {
+        "database": "catalog", "active_branch": "main", "head": "abc123",
+        "unrelated_sha256": "before",
+    }
+    after = {**before, "unrelated_sha256": "after"}
+
+    import pytest
+    from openstatspec.core import UnsupportedOperationError
+
+    with pytest.raises(UnsupportedOperationError, match="unrelated working-set"):
+        wide._require_dolt_success_identity(before, after, phase="export audit")
+
+
 def test_export_lifecycle_events_remain_linked_to_the_dataset(tmp_path):
     path = tmp_path / "export-events.sqlite"
     database_url = f"sqlite:///{path}"

tests/test_strict_catalog_review.py

@@ -95,6 +95,23 @@ def test_read_and_export_reject_obsolete_catalog_relations(tmp_path):
         )
 
 
+def test_missing_workflow_trigger_is_rejected_by_core_catalog_verification(
+    tmp_path,
+):
+    path = tmp_path / "workflow-trigger-drift.sqlite"
+    database = f"sqlite:///{path}"
+    openstatspec.initialize_catalog(database_url=database)
+    engine = create_engine(database)
+    with engine.begin() as connection:
+        create_workflow_catalog(connection, workflow_catalog(MetaData()))
+        connection.exec_driver_sql(
+            "drop trigger oss_transformation_run_no_delete"
+        )
+
+    with pytest.raises(UnsupportedOperationError, match="incompatible"):
+        openstatspec.initialize_catalog(database_url=database)
+
+
 def test_verified_workflow_profile_remains_catalog_owned(tmp_path):
     path = tmp_path / "workflow-owned.sqlite"
     database = f"sqlite:///{path}"