added data verification on canonify

app/mapper.py

@@ -422,6 +422,8 @@ def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd
     # 7️⃣ SCHEMA VERSIONING & TRANSACTIONAL INSERT
     os.makedirs("./db", exist_ok=True)
     
+    rows_inserted = 0  # Track insertion count
+    
     with transactional_conn(org_id) as duck:
         ensure_schema_versions_table(duck)
         
@@ -468,7 +470,8 @@ def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd
                     f"INSERT INTO {table_name} ({cols_str}) VALUES ({placeholders})",
                     df_to_insert.values.tolist()
                 )
-                print(f"[canonify] 💾 Inserted {len(df_to_insert)} rows into {table_name}")
+                rows_inserted = len(df_to_insert)
+                print(f"[canonify] 💾 Inserted {rows_inserted} rows into {table_name}")
             
             # Mark schema as applied
             if is_new_schema and version_id:
@@ -482,12 +485,45 @@ def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd
                 except Exception as e:
                     print(f"[canonify] ⚠️ Schema update warning: {e}")
     
+    # ⬇️⬇️⬇️ NEW VERIFICATION STEP ⬇️⬇️⬇️
+    # CRITICAL: Ensure data is committed and visible before spawning worker
+    
+    print(f"[canonify] 🔍 Verifying data visibility...")
+    
+    # Use a NEW connection to verify data is visible (ensures isolation level is correct)
+    verification_conn = None
+    try:
+        verification_conn = get_conn(org_id)
+        verification_conn.execute("CHECKPOINT")  # Force any pending writes to flush
+        
+        # Query the table we just inserted into
+        verify_table = f"main.{entity_type}_canonical"
+        verify_count = verification_conn.execute(f"SELECT COUNT(*) FROM {verify_table}").fetchone()[0]
+        
+        if verify_count >= rows_inserted:
+            print(f"[canonify] ✅ Verification passed: {verify_count} rows visible in {verify_table}")
+        else:
+            print(f"[canonify] ❌ Verification failed: Expected {rows_inserted}, found {verify_count} rows")
+            # Don't proceed if verification fails
+            return df, industry, industry_confidence
+            
+    except Exception as e:
+        print(f"[canonify] ❌ Verification error: {e}")
+        return df, industry, industry_confidence
+    finally:
+        if verification_conn:
+            try:
+                verification_conn.close()
+                print("[canonify] 🔒 Verification connection closed")
+            except:
+                pass
+    # ⬆️⬆️⬆️ END VERIFICATION ⬆️⬆️⬆️
     # 8️⃣ FINAL: Clean DataFrame for response
     df = df.replace([np.inf, -np.inf, np.nan], None)
     duration_ms = (datetime.now() - start_time).total_seconds() * 1000
     print(f"[canonify] ✅ Pipeline complete in {duration_ms:.2f}ms for {org_id}")
     
-    # 9️⃣ SINGLE, SAFE WORKER TRIGGER (idempotent)
+    # 9️⃣ SINGLE, SAFE WORKER TRIGGER (idempotent) - NOW ONLY AFTER VERIFICATION
     try:
         # Defensive: ensure keys exist (they should from poll_for_entity)
         e_key = f"entity:{org_id}:{source_id}"
@@ -502,7 +538,7 @@ def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd
             "type": "kpi_compute",
             "entity_type": entity_type,
             "industry": industry,
-            "rows_inserted": len(df),
+            "rows_inserted": rows_inserted,  # Use actual count
             "timestamp": datetime.now().isoformat()
         })
         print(f"[canonify] 🚀 Triggered analytics for {source_id}")