Bound RC10 persistence audit queries

backend/tests/test_rc10_data_operations.py

@@ -213,8 +213,13 @@ def test_source_family_report_covers_national_regional_and_maritime(tmp_path: Pa
             return self.values
 
     class Session:
-        def query(self, model):
+        def query(self, model, *_fields):
+            if model in rows:
                 return Query(rows[model])
+            owner = getattr(model, "class_", None)
+            if owner in rows:
+                return Query(rows[owner])
+            raise AssertionError(f"Unexpected query entity: {model!r}")
 
     state = audit.collect_database_state(Session(), tmp_path)
 

scripts/audit_data_operations.py

@@ -176,15 +176,40 @@ def _add_row_references(target: set[Path], row: Any, storage_root: Path, direct_
 
 
 def collect_database_state(db: Any, storage_root: Path) -> dict[str, Any]:
-    projects = db.query(Project).all()
+    projects = db.query(Project.id, Project.name, Project.status).all()
     project_names = {project.id: project.name for project in projects}
-    datasets = db.query(Dataset).all()
+    datasets = db.query(
-    versions = db.query(DatasetVersion).all()
+        Dataset.id,
-    exports = db.query(Export).all()
+        Dataset.project_id,
-    detections = db.query(Detection).all()
+        Dataset.name,
-    segmentations = db.query(Segmentation).all()
+        Dataset.source,
-    jobs = db.query(Job).all()
+        Dataset.source_name,
-    analysis_runs = db.query(AnalysisRun).all()
+        Dataset.storage_path,
+        Dataset.metadata_json,
+        Dataset.source_metadata,
+        Dataset.provenance_metadata,
+        Dataset.source_version,
+        Dataset.imported_at,
+        Dataset.status,
+    ).all()
+    versions = db.query(
+        DatasetVersion.storage_path,
+        DatasetVersion.source_metadata,
+        DatasetVersion.provenance_metadata,
+    ).all()
+    exports = db.query(Export.storage_path, Export.metadata_json).all()
+    # Geometry and feature properties can be very large. The persisted direct
+    # artifact columns are sufficient here and keep the operator audit bounded.
+    detections = db.query(Detection.source_tile_path).all()
+    segmentations = db.query(
+        Segmentation.mask_path,
+        Segmentation.source_tile_path,
+        Segmentation.provenance_json,
+    ).all()
+    # Job/run JSON is not authoritative artifact persistence. Dataset, Export,
+    # Detection and Segmentation rows above own every retained path.
+    jobs = db.query(Job.status, Job.created_at).all()
+    analysis_runs = db.query(AnalysisRun.status, AnalysisRun.created_at).all()
 
     references: set[Path] = set()
     for row in datasets:
@@ -197,11 +222,6 @@ def collect_database_state(db: Any, storage_root: Path) -> dict[str, Any]:
         _add_row_references(references, row, storage_root, ("source_tile_path",))
     for row in segmentations:
         _add_row_references(references, row, storage_root, ("mask_path", "source_tile_path"))
-    for row in jobs:
-        _add_row_references(references, row, storage_root, ())
-    for row in analysis_runs:
-        _add_row_references(references, row, storage_root, ())
-
     source_families: dict[str, dict[str, dict[str, Any]]] = {
         "national": {},
         "regional": {},
@@ -210,7 +230,7 @@ def collect_database_state(db: Any, storage_root: Path) -> dict[str, Any]:
     maritime_tokens = ("marine", "maritime", "north_sea", "bathymetry", "rbin", "mdk", "msp")
     for dataset in datasets:
         source_name = (dataset.source_name or dataset.source or "unknown").strip().lower()
-        metadata = dataset.source_metadata or {}
+        metadata = dataset.source_metadata if isinstance(dataset.source_metadata, dict) else {}
         zones = metadata.get("coverage_zones") or metadata.get("coverage_zone") or []
         if isinstance(zones, str):
             zones = [zones]