feat: progressive multi-round face matching + pending person API

- Identity agent: per-face max matching, multi-round with derived seeds from high-confidence faces, angle diversity filter (cosine sim < 0.90) - Pending person API: POST /file/:file_uuid/pending-person + GET /file/:file_uuid/pending-persons with status=pending, source=manual - Update API docs (07_identity.md)
2026-06-24 03:42:04 +08:00
parent 766a1d9a6d
commit 14e886cc08
31 changed files with 5882 additions and 742 deletions
@@ -23,6 +23,14 @@ pub struct FaceEmbeddingPayload {
    pub yaw: f64,
    pub pitch: f64,
    pub roll: f64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub identity_uuid: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub identity_ref: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub stranger_ref: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none", rename = "type")]
+    pub r#type: Option<String>,
 }

 #[derive(Debug, Clone, Deserialize)]
@@ -166,13 +174,117 @@ impl FaceEmbeddingDb {
            .context("Failed to batch upsert face embeddings")?;

        if !response.status().is_success() {
+            let status = response.status();
            let text = response.text().await.unwrap_or_default();
-            anyhow::bail!("Qdrant batch upsert failed: {}", text);
+            anyhow::bail!("Qdrant batch upsert failed (HTTP {}): {}", status, text);
        }

        Ok(points.len())
    }

+    pub async fn update_identity_by_trace(
+        &self,
+        file_uuid: &str,
+        trace_id: i32,
+        identity_uuid: &str,
+    ) -> Result<usize> {
+        let url = format!(
+            "{}/collections/{}/points",
+            self.base_url, self.collection_name
+        );
+
+        let body = serde_json::json!({
+            "filter": {
+                "must": [
+                    {
+                        "key": "file_uuid",
+                        "match": { "value": file_uuid }
+                    },
+                    {
+                        "key": "trace_id",
+                        "match": { "value": trace_id }
+                    }
+                ]
+            },
+            "payload": {
+                "identity_uuid": identity_uuid
+            }
+        });
+
+        let response = self
+            .client
+            .post(&url)
+            .header("api-key", &self.api_key)
+            .header("Content-Type", "application/json")
+            .json(&body)
+            .send()
+            .await
+            .context("Failed to update identity_uuid in Qdrant")?;
+
+        if !response.status().is_success() {
+            let text = response.text().await.unwrap_or_default();
+            anyhow::bail!("Qdrant identity update failed: {}", text);
+        }
+
+        tracing::info!(
+            "[FaceEmbedding] Updated identity_uuid={} for file={}, trace={}",
+            identity_uuid, file_uuid, trace_id
+        );
+
+        Ok(1)
+    }
+
+    pub async fn clear_identity_by_trace(
+        &self,
+        file_uuid: &str,
+        trace_id: i32,
+    ) -> Result<usize> {
+        let url = format!(
+            "{}/collections/{}/points",
+            self.base_url, self.collection_name
+        );
+
+        let body = serde_json::json!({
+            "filter": {
+                "must": [
+                    {
+                        "key": "file_uuid",
+                        "match": { "value": file_uuid }
+                    },
+                    {
+                        "key": "trace_id",
+                        "match": { "value": trace_id }
+                    }
+                ]
+            },
+            "payload": {
+                "identity_uuid": null
+            }
+        });
+
+        let response = self
+            .client
+            .post(&url)
+            .header("api-key", &self.api_key)
+            .header("Content-Type", "application/json")
+            .json(&body)
+            .send()
+            .await
+            .context("Failed to clear identity_uuid in Qdrant")?;
+
+        if !response.status().is_success() {
+            let text = response.text().await.unwrap_or_default();
+            anyhow::bail!("Qdrant identity clear failed: {}", text);
+        }
+
+        tracing::info!(
+            "[FaceEmbedding] Cleared identity_uuid for file={}, trace={}",
+            file_uuid, trace_id
+        );
+
+        Ok(1)
+    }
+
    pub async fn search_similar(
        &self,
        query_embedding: &[f32],
@@ -294,6 +406,26 @@ impl FaceEmbeddingDb {
                        .get("roll")
                        .and_then(|v| v.as_f64())
                        .unwrap_or(0.0),
+                    identity_uuid: r
+                        .payload
+                        .get("identity_uuid")
+                        .and_then(|v| v.as_str())
+                        .map(|s| s.to_string()),
+                    identity_ref: r
+                        .payload
+                        .get("identity_ref")
+                        .and_then(|v| v.as_str())
+                        .map(|s| s.to_string()),
+                    stranger_ref: r
+                        .payload
+                        .get("stranger_ref")
+                        .and_then(|v| v.as_str())
+                        .map(|s| s.to_string()),
+                    r#type: r
+                        .payload
+                        .get("type")
+                        .and_then(|v| v.as_str())
+                        .map(|s| s.to_string()),
                };
                FaceEmbeddingPoint {
                    id,
@@ -498,6 +630,26 @@ impl FaceEmbeddingDb {
                        .get("roll")
                        .and_then(|v| v.as_f64())
                        .unwrap_or(0.0),
+                    identity_uuid: r
+                        .payload
+                        .get("identity_uuid")
+                        .and_then(|v| v.as_str())
+                        .map(|s| s.to_string()),
+                    identity_ref: r
+                        .payload
+                        .get("identity_ref")
+                        .and_then(|v| v.as_str())
+                        .map(|s| s.to_string()),
+                    stranger_ref: r
+                        .payload
+                        .get("stranger_ref")
+                        .and_then(|v| v.as_str())
+                        .map(|s| s.to_string()),
+                    r#type: r
+                        .payload
+                        .get("type")
+                        .and_then(|v| v.as_str())
+                        .map(|s| s.to_string()),
                };
                (id, r.vector, payload)
            })
@@ -537,6 +689,258 @@ impl FaceEmbeddingDb {

        Ok(0)
    }
+
+    pub async fn upsert_seed_embedding(
+        &self,
+        identity_uuid: &str,
+        identity_name: &str,
+        tmdb_id: i32,
+        embedding: &[f32],
+    ) -> Result<()> {
+        let url = format!(
+            "{}/collections/{}/points?wait=true",
+            self.base_url, self.collection_name
+        );
+
+        let point_id = identity_uuid.to_string();
+        let payload = serde_json::json!({
+            "file_uuid": "",
+            "trace_id": 0,
+            "frame": 0,
+            "bbox_x": 0.0,
+            "bbox_y": 0.0,
+            "bbox_w": 0.0,
+            "bbox_h": 0.0,
+            "confidence": 0.0,
+            "yaw": 0.0,
+            "pitch": 0.0,
+            "roll": 0.0,
+            "identity_uuid": identity_uuid,
+            "identity_ref": serde_json::Value::Null,
+            "stranger_ref": serde_json::Value::Null,
+            "identity_name": identity_name,
+            "tmdb_id": tmdb_id,
+            "type": "identity_seed",
+        });
+
+        let body = serde_json::json!({
+            "points": [{
+                "id": point_id,
+                "vector": embedding,
+                "payload": payload
+            }]
+        });
+
+        let response = self
+            .client
+            .put(&url)
+            .header("api-key", &self.api_key)
+            .header("Content-Type", "application/json")
+            .json(&body)
+            .send()
+            .await
+            .context("Failed to upsert seed embedding")?;
+
+        if !response.status().is_success() {
+            let text = response.text().await.unwrap_or_default();
+            anyhow::bail!("Qdrant seed upsert failed: {}", text);
+        }
+
+        tracing::info!(
+            "[SeedEmbedding] Stored seed for identity_uuid={}, name={}",
+            identity_uuid, identity_name
+        );
+
+        Ok(())
+    }
+
+    pub async fn get_seed_embeddings(
+        &self,
+    ) -> Result<Vec<(String, String, Vec<f32>)>> {
+        let url = format!(
+            "{}/collections/{}/points/scroll",
+            self.base_url, self.collection_name
+        );
+
+        let body = serde_json::json!({
+            "limit": 10000,
+            "with_payload": true,
+            "with_vector": true,
+            "filter": {
+                "must": [
+                    {"key": "type", "match": { "value": "identity_seed" }}
+                ]
+            }
+        });
+
+        let response = self
+            .client
+            .post(&url)
+            .header("api-key", &self.api_key)
+            .header("Content-Type", "application/json")
+            .json(&body)
+            .send()
+            .await
+            .context("Failed to scroll seed embeddings")?;
+
+        let status = response.status();
+        let text = response.text().await.unwrap_or_default();
+
+        if !status.is_success() {
+            anyhow::bail!("Qdrant scroll failed: {} - {}", status, text);
+        }
+
+        #[derive(Deserialize)]
+        struct ScrollResult {
+            result: ScrollPoints,
+        }
+
+        #[derive(Deserialize)]
+        struct ScrollPoints {
+            points: Vec<PointResult>,
+        }
+
+        #[derive(Deserialize)]
+        struct PointResult {
+            id: serde_json::Value,
+            vector: Vec<f32>,
+            payload: HashMap<String, serde_json::Value>,
+        }
+
+        let parsed: ScrollResult =
+            serde_json::from_str(&text).context("Failed to parse Qdrant scroll response")?;
+
+        let results: Vec<(String, String, Vec<f32>)> = parsed
+            .result
+            .points
+            .into_iter()
+            .filter_map(|r| {
+                let identity_uuid = r
+                    .payload
+                    .get("identity_uuid")
+                    .and_then(|v| v.as_str())
+                    .unwrap_or("")
+                    .to_string();
+                let identity_name = r
+                    .payload
+                    .get("identity_name")
+                    .and_then(|v| v.as_str())
+                    .unwrap_or("")
+                    .to_string();
+                if identity_uuid.is_empty() {
+                    None
+                } else {
+                    Some((identity_uuid, identity_name, r.vector))
+                }
+            })
+            .collect();
+
+        Ok(results)
+    }
+
+    pub async fn update_identity_ref_by_trace(
+        &self,
+        file_uuid: &str,
+        trace_id: i32,
+        identity_ref: &str,
+    ) -> Result<usize> {
+        let url = format!(
+            "{}/collections/{}/points/payload",
+            self.base_url, self.collection_name
+        );
+
+        let body = serde_json::json!({
+            "filter": {
+                "must": [
+                    {
+                        "key": "file_uuid",
+                        "match": { "value": file_uuid }
+                    },
+                    {
+                        "key": "trace_id",
+                        "match": { "value": trace_id }
+                    }
+                ]
+            },
+            "payload": {
+                "identity_ref": identity_ref
+            }
+        });
+
+        let response = self
+            .client
+            .post(&url)
+            .header("api-key", &self.api_key)
+            .header("Content-Type", "application/json")
+            .json(&body)
+            .send()
+            .await
+            .context("Failed to update identity_ref in Qdrant")?;
+
+        if !response.status().is_success() {
+            let text = response.text().await.unwrap_or_default();
+            anyhow::bail!("Qdrant identity_ref update failed: {}", text);
+        }
+
+        tracing::info!(
+            "[FaceEmbedding] Updated identity_ref={} for file={}, trace={}",
+            identity_ref, file_uuid, trace_id
+        );
+
+        Ok(1)
+    }
+
+    pub async fn update_stranger_ref_by_trace(
+        &self,
+        file_uuid: &str,
+        trace_id: i32,
+        stranger_ref: &str,
+    ) -> Result<usize> {
+        let url = format!(
+            "{}/collections/{}/points/payload",
+            self.base_url, self.collection_name
+        );
+
+        let body = serde_json::json!({
+            "filter": {
+                "must": [
+                    {
+                        "key": "file_uuid",
+                        "match": { "value": file_uuid }
+                    },
+                    {
+                        "key": "trace_id",
+                        "match": { "value": trace_id }
+                    }
+                ]
+            },
+            "payload": {
+                "stranger_ref": stranger_ref
+            }
+        });
+
+        let response = self
+            .client
+            .post(&url)
+            .header("api-key", &self.api_key)
+            .header("Content-Type", "application/json")
+            .json(&body)
+            .send()
+            .await
+            .context("Failed to update stranger_ref in Qdrant")?;
+
+        if !response.status().is_success() {
+            let text = response.text().await.unwrap_or_default();
+            anyhow::bail!("Qdrant stranger_ref update failed: {}", text);
+        }
+
+        tracing::info!(
+            "[FaceEmbedding] Updated stranger_ref={} for file={}, trace={}",
+            stranger_ref, file_uuid, trace_id
+        );
+
+        Ok(1)
+    }
 }

 impl Default for FaceEmbeddingDb {
@@ -5,6 +5,7 @@ use serde_json;
 use sqlx::{postgres::PgPoolOptions, PgPool, Row};
 use std::sync::Arc;
 use tokio::sync::RwLock;
+use tracing::{info, warn, error};
 use uuid::Uuid;

 use super::{schema, Database, QdrantDb};
@@ -448,6 +449,7 @@ pub enum ProcessorType {
    FiveW1H,
    Appearance,
    MediaPipe,
+    FaceCluster,
 }

 impl sqlx::Type<sqlx::Postgres> for ProcessorType {
@@ -487,6 +489,7 @@ impl ProcessorType {
            ProcessorType::FiveW1H => "5w1h",
            ProcessorType::Appearance => "appearance",
            ProcessorType::MediaPipe => "mediapipe",
+            ProcessorType::FaceCluster => "face_cluster",
        }
    }

@@ -505,6 +508,7 @@ impl ProcessorType {
            "5w1h" => Some(ProcessorType::FiveW1H),
            "appearance" => Some(ProcessorType::Appearance),
            "mediapipe" => Some(ProcessorType::MediaPipe),
+            "face_cluster" => Some(ProcessorType::FaceCluster),
            _ => None,
        }
    }
@@ -524,13 +528,14 @@ impl ProcessorType {
            ProcessorType::FiveW1H => 0.1,
            ProcessorType::Appearance => 0.3,
            ProcessorType::MediaPipe => 0.3,
+            ProcessorType::FaceCluster => 0.7,
        }
    }

    pub fn uses_gpu(&self) -> bool {
        match self {
            ProcessorType::Yolo | ProcessorType::Face | ProcessorType::Pose | ProcessorType::Hand => true,
-            ProcessorType::MediaPipe => false,
+            ProcessorType::MediaPipe | ProcessorType::FaceCluster => false,
            _ => false,
        }
    }
@@ -550,6 +555,7 @@ impl ProcessorType {
            ProcessorType::FiveW1H => 256,
            ProcessorType::Appearance => 512,
            ProcessorType::MediaPipe => 1024,
+            ProcessorType::FaceCluster => 1024,
        }
    }

@@ -568,6 +574,7 @@ impl ProcessorType {
            ProcessorType::FiveW1H => Some("gemma4"),
            ProcessorType::Appearance => None,
            ProcessorType::MediaPipe => Some("mediapipe/holistic"),
+            ProcessorType::FaceCluster => Some("sklearn/agglomerative"),
        }
    }

@@ -583,6 +590,7 @@ impl ProcessorType {
            ],
            ProcessorType::FiveW1H => vec![ProcessorType::Story],
            ProcessorType::Appearance => vec![ProcessorType::Pose],
+            ProcessorType::FaceCluster => vec![ProcessorType::Face],
            ProcessorType::Hand => vec![],
            ProcessorType::MediaPipe => vec![],
            _ => vec![],
@@ -597,6 +605,7 @@ impl ProcessorType {
            ProcessorType::Yolo,
            ProcessorType::Ocr,
            ProcessorType::Face,
+            ProcessorType::FaceCluster,
            ProcessorType::Pose,
            ProcessorType::Hand,
            ProcessorType::Appearance,
@@ -611,7 +620,8 @@ impl ProcessorType {
            | ProcessorType::Pose
            | ProcessorType::Hand
            | ProcessorType::Appearance
-            | ProcessorType::MediaPipe => PipelineType::Frame,
+            | ProcessorType::MediaPipe
+            | ProcessorType::FaceCluster => PipelineType::Frame,

            ProcessorType::Cut
            | ProcessorType::Asr
@@ -1074,9 +1084,9 @@ impl PostgresDb {
        let mj_cols = [
            "video_id BIGINT",
            "user_id BIGINT",
-            "processors TEXT[]",
-            "completed_processors TEXT[]",
-            "failed_processors TEXT[]",
+            "processors TEXT[] DEFAULT '{\"asr\",\"cut\",\"yolo\",\"ocr\",\"face\",\"pose\",\"asrx\"}'",
+            "completed_processors TEXT[] DEFAULT '{}'",
+            "failed_processors TEXT[] DEFAULT '{}'",
        ];
        for col in &mj_cols {
            let (col_name, col_def) = col.split_once(' ').unwrap_or((col, ""));
@@ -1087,6 +1097,10 @@ impl PostgresDb {
            .execute(pool)
            .await?;
        }
+        // Update existing rows to have default processors array
+        sqlx::query("UPDATE monitor_jobs SET processors = '{\"asr\",\"cut\",\"yolo\",\"ocr\",\"face\",\"pose\",\"asrx\"}' WHERE processors IS NULL OR processors = '{}'")
+            .execute(pool)
+            .await?;
        sqlx::query("CREATE INDEX IF NOT EXISTS idx_monitor_jobs_status ON monitor_jobs(status)")
            .execute(pool)
            .await?;
@@ -1869,16 +1883,16 @@ impl PostgresDb {
            .await?
        } else {
            // Insert new job
-            sqlx::query(
-                &format!(
-                    r#"
-                    INSERT INTO {} (uuid, video_path, status, video_id)
-                    VALUES ($1, $2, 'pending', $3)
+sqlx::query(
+                    &format!(
+                        r#"
+                    INSERT INTO {} (uuid, video_path, status, video_id, processors)
+                    VALUES ($1, $2, 'pending', $3, ARRAY['asr','cut','yolo','ocr','face','face_cluster','pose','asrx'])
                    RETURNING id, uuid, video_path, status, current_processor, progress_total, progress_current, error_count, last_error, started_at::TEXT, updated_at::TEXT, created_at::TEXT, processors, completed_processors, failed_processors, video_id
                    "#,
-                    jobs_table
+                        jobs_table
+                    )
                )
-            )
            .bind(uuid)
            .bind(video_path)
            .bind(video_id_i64)
@@ -3176,6 +3190,40 @@ impl PostgresDb {
        Ok(r.rows_affected())
    }

+    pub async fn retry_failed_processor(
+        &self,
+        result_id: i32,
+        max_retries: i32,
+    ) -> Result<bool> {
+        let table = schema::table_name("processor_results");
+        use sqlx::Row;
+        
+        let current_retry: i32 = sqlx::query_scalar(&format!(
+            "SELECT COALESCE(retry_count, 0) FROM {} WHERE id = $1",
+            table
+        ))
+        .bind(result_id)
+        .fetch_one(&self.pool)
+        .await?;
+
+        if current_retry < max_retries {
+            sqlx::query(&format!(
+                "UPDATE {} SET status = 'pending', error_message = NULL, retry_count = $1, updated_at = CURRENT_TIMESTAMP WHERE id = $2",
+                table
+            ))
+            .bind(current_retry + 1)
+            .bind(result_id)
+            .execute(&self.pool)
+            .await?;
+            
+            info!("🔄 Retrying processor (result_id={}, retry_count={}/{})", result_id, current_retry + 1, max_retries);
+            Ok(true)
+        } else {
+            info!("⚠️ Processor exceeded max retries (result_id={}, retry_count={})", result_id, current_retry);
+            Ok(false)
+        }
+    }
+
    pub async fn search_bm25(
        &self,
        query: &str,
@@ -69,7 +69,8 @@ pub struct IdentityBinding {
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct BindIdentityRequest {
    pub file_uuid: String,
-    pub face_id: String,
+    pub face_id: Option<String>,
+    pub id: Option<i64>,
 }

 #[derive(Debug, Clone, Deserialize, Serialize)]
@@ -81,7 +82,8 @@ pub struct BindIdentityTraceRequest {
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct UnbindIdentityRequest {
    pub file_uuid: String,
-    pub face_id: String,
+    pub face_id: Option<String>,
+    pub id: Option<i64>,
 }

 #[derive(Debug, Clone, Deserialize, Serialize)]
@@ -0,0 +1,75 @@
+use anyhow::{Context, Result};
+use serde::{Deserialize, Serialize};
+use std::time::Duration;
+
+use super::executor::PythonExecutor;
+
+const FACE_CLUSTER_TIMEOUT: Duration = Duration::from_secs(3600);
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct FaceClusterResult {
+    pub clusters: Vec<FaceClusterInfo>,
+    pub frames: Vec<FaceClusterFrame>,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct FaceClusterInfo {
+    pub cluster_id: String,
+    pub face_count: usize,
+    pub representative_face: Option<String>,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct FaceClusterFrame {
+    pub frame: u64,
+    pub timestamp: f64,
+    pub faces: Vec<ClusteredFace>,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct ClusteredFace {
+    pub face_id: String,
+    pub cluster_id: String,
+    pub confidence: f32,
+}
+
+pub async fn process_face_cluster(
+    video_path: &str,
+    output_path: &str,
+    uuid: Option<&str>,
+    frames: Option<&[i64]>,
+) -> Result<FaceClusterResult> {
+    let executor = PythonExecutor::new()?;
+    let script_path = executor.script_path("fast_face_clustering_processor.py");
+
+    tracing::info!("[FACE_CLUSTER] Starting face clustering: {}", video_path);
+
+    if !script_path.exists() {
+        tracing::warn!("[FACE_CLUSTER] Script not found, returning empty result");
+        return Ok(FaceClusterResult {
+            clusters: vec![],
+            frames: vec![],
+        });
+    }
+
+    executor
+        .run_with_frames(
+            "fast_face_clustering_processor.py",
+            &[video_path, output_path],
+            uuid,
+            "FACE_CLUSTER",
+            Some(FACE_CLUSTER_TIMEOUT),
+            frames,
+        )
+        .await
+        .with_context(|| format!("Failed to run face clustering script"))?;
+
+    let json_str = std::fs::read_to_string(output_path).context("Failed to read FACE_CLUSTER output")?;
+
+    let result: FaceClusterResult =
+        serde_json::from_str(&json_str).context("Failed to parse FACE_CLUSTER output")?;
+
+    tracing::info!("[FACE_CLUSTER] Result: {} clusters, {} frames", result.clusters.len(), result.frames.len());
+
+    Ok(result)
+}
@@ -7,6 +7,7 @@ pub mod clip;
 pub mod cut;
 pub mod executor;
 pub mod face;
+pub mod face_clustering;
 pub mod face_recognition;
 pub mod hand;
 pub mod heuristic_scene;
@@ -32,6 +33,9 @@ pub use clip::{
 pub use cut::{process_cut, CutResult, CutScene};
 pub use executor::{validate_python_env, PythonExecutor, RetryConfig};
 pub use face::{process_face, Face, FaceFrame, FaceResult};
+pub use face_clustering::{
+    process_face_cluster, ClusteredFace, FaceClusterFrame, FaceClusterInfo, FaceClusterResult,
+};
 pub use face_recognition::{
    process_face_recognition, register_face, FaceAttributes, FaceCluster, FaceIdentity, FacePose,
    FaceRecognitionFrame, FaceRecognitionResult, FaceRegistrationResult, RecognizedFace,
@@ -129,7 +129,7 @@ async fn populate_face_embeddings_to_qdrant(
    // Load from face_detections table
    let fd_table = t("face_detections");
    let rows: Vec<(i32, i64, f64, f64, f64, f64, f64, Option<Vec<f32>>)> = sqlx::query_as(&format!(
-        "SELECT trace_id::int, frame_number::bigint, x::float8, y::float8, width::float8, height::float8, confidence::float8, embedding \
+        "SELECT trace_id::int, frame_number::bigint, x::float8, y::float8, width::float8, height::float8, confidence::float8, embedding::float4[] \
         FROM {} WHERE file_uuid = $1 AND trace_id IS NOT NULL AND embedding IS NOT NULL",
        fd_table
    ))
@@ -165,11 +165,20 @@ async fn populate_face_embeddings_to_qdrant(
                yaw,
                pitch,
                roll,
+                identity_uuid: None,
+                identity_ref: None,
+                stranger_ref: None,
+                r#type: None,
            };
            points.push((point_id, emb.clone(), payload));
        }
    }

+    info!(
+        "[TKG-Phase1] Attempting to store {} face embeddings in Qdrant for {}",
+        points.len(),
+        file_uuid
+    );
    let count = face_db.batch_upsert(points).await?;
    info!(
        "[TKG-Phase1] Stored {} face embeddings in Qdrant for {}",
@@ -401,19 +410,7 @@ fn detect_mutual_gaze(
 #[derive(Debug, Deserialize)]
 struct YoloJson {
 #[serde(default)]
-frames: Vec<YoloFrameData>,
-}
-
-#[derive(Debug, Deserialize)]
-struct YoloFrameData {
-#[serde(default)]
-frame: u32,
-#[serde(default)]
-timestamp: f64,
-#[serde(default)]
-detections: Vec<YoloDetEntry>,
-#[serde(default)]
-objects: Vec<YoloDetEntry>,
+frames: HashMap<String, YoloFrameEntry>,
 }

 #[derive(Debug, Deserialize)]
@@ -1033,7 +1030,7 @@ async fn build_yolo_object_nodes(
        .with_context(|| format!("Failed to parse {:?}", yolo_path))?;

 let mut class_counts: HashMap<String, i64> = HashMap::new();
-for fdata in &yolo.frames {
+for fdata in yolo.frames.values() {
 let dets = if !fdata.detections.is_empty() {
 &fdata.detections
 } else {
@@ -1277,9 +1274,9 @@ async fn build_co_occurrence_edges_from_qdrant(

 let mut edge_count = 0;
 for (frame, faces) in frame_faces.iter() {
-let yolo_frame = match yolo.frames.iter().find(|f| f.frame == *frame as u32) {
-Some(f) => f,
-None => continue,
+let yolo_frame = match yolo.frames.get(&frame.to_string()) {
+    Some(f) => f,
+    None => continue,
 };

        let dets = if !yolo_frame.detections.is_empty() {
@@ -1391,9 +1388,9 @@ async fn build_co_occurrence_edges_from_pg(

    let mut edge_count = 0;
 for face in &face_rows {
-let yolo_frame = match yolo.frames.iter().find(|f| f.frame == face.frame_number as u32) {
-Some(f) => f,
-None => continue,
+let yolo_frame = match yolo.frames.get(&face.frame_number.to_string()) {
+    Some(f) => f,
+    None => continue,
 };

        let dets = if !yolo_frame.detections.is_empty() {
@@ -2411,7 +2408,9 @@ async fn build_gaze_track_nodes_from_face_json(
    let nodes_table = t("tkg_nodes");
    sqlx::query(&format!(
        "INSERT INTO {} (file_uuid, external_id, label, node_type, properties, created_at) \
-         VALUES ($1, $2, $3, 'gaze_track', $4, NOW())",
+         VALUES ($1, $2, $3, 'gaze_track', $4, NOW()) \
+         ON CONFLICT (file_uuid, node_type, external_id) \
+         DO UPDATE SET properties = COALESCE(EXCLUDED.properties, tkg_nodes.properties)",
        nodes_table
    ))
    .bind(file_uuid)
@@ -3063,7 +3062,9 @@ async fn build_lip_track_nodes_from_face_json(
    let nodes_table = t("tkg_nodes");
    sqlx::query(&format!(
        "INSERT INTO {} (file_uuid, external_id, label, node_type, properties, created_at) \
-         VALUES ($1, $2, $3, 'lip_track', $4, NOW())",
+         VALUES ($1, $2, $3, 'lip_track', $4, NOW()) \
+         ON CONFLICT (file_uuid, node_type, external_id) \
+         DO UPDATE SET properties = COALESCE(EXCLUDED.properties, tkg_nodes.properties)",
        nodes_table
    ))
    .bind(file_uuid)
@@ -3814,10 +3815,10 @@ async fn build_hand_object_edges(pool: &PgPool, file_uuid: &str, output_dir: &st

    let yolo_frames: HashMap<u64, &Vec<YoloDetEntry>> = yolo.frames
        .iter()
-        .filter_map(|f| {
+        .filter_map(|(frame_key, f)| {
            let objs = if !f.objects.is_empty() { &f.objects } else { &f.detections };
            if !objs.is_empty() {
-                Some((f.frame as u64, objs))
+                frame_key.parse::<u64>().ok().map(|n| (n, objs))
            } else {
                None
            }