feat: add cluster-agent endpoint, fix OCR labeling, fix ingestion blocking

- Add POST /api/v1/file/:file_uuid/cluster-agent endpoint for on-demand face clustering
- Fix OCR chunks being labeled as ASRX: use ChunkType.as_str() instead of {:?}
- Rule 1 now deletes old chunks before re-inserting to avoid stale data
- Add fallback face_traced.json when store_traced_faces.py fails
- ingestion_complete now handles status='error' to unblock jobs
- VLM describe tool now uses trace_id with pre-extracted face crops

src/api/agent_search.rs

@@ -96,7 +96,9 @@ const SYSTEM_PROMPT: &str = r#"你是 Momentry 影片分析助手。回答用戶
 7. 人物台詞內容使用 tkg_query 的 speaker_dialogue
 8. 用文字反查人物使用 identity_text（輸入關鍵字→找出誰說/提到這段話）
 9. 畫面分析使用 analyze_frame — 可以分析影片中的任何畫面內容（場景、人物表情、動作、物件等）
-10. **可以同時呼叫多個工具，但需符合以下條件：**
+10. **人物外貌/衣著/顏色問題使用 vlm_describe** — 用 trace_id 查人臉裁切圖，經 VLM 分析描述衣著顏色、款式、配件等
+11. **vlm_describe 使用流程**：先用 tkg_query top_identities/identity_traces 找到該人的 trace_id，再呼叫 vlm_describe(file_uuid + trace_id)
+12. **可以同時呼叫多個工具，但需符合以下條件：**
     - ✅ 查詢多部影片的相同資訊（如：3部影片的人物列表）
     - ✅ 需要組合多個來源的資訊才能回答（如：file_info + tkg_query）
     - ❌ 不要為了「嘗試所有可能」而盲目並行呼叫
@@ -261,6 +263,16 @@ fn make_tools(pool: &sqlx::PgPool) -> Vec<ToolDef> {
             }),
             vec!["file_uuid", "color"],
         ),
+        function_calling::make_tool(
+            "vlm_describe",
+            "Describe a person's appearance (clothing, colors, accessories) by face trace ID using Vision Language Model (LLaVA). Uses pre-extracted face crop images.",
+            serde_json::json!({
+                "file_uuid": {"type": "string", "description": "UUID of the video file"},
+                "trace_id": {"type": "integer", "description": "Face trace ID to analyze"},
+                "prompt": {"type": "string", "description": "Specific question about the person's appearance (optional, default: describes clothing colors and style)"}
+            }),
+            vec!["file_uuid", "trace_id"],
+        ),
     ]
 }
 
@@ -296,6 +308,7 @@ async fn execute_tool(pool: &sqlx::PgPool, tool_call: &ToolCall) -> (String, Str
         "get_representative_frame" => tools::exec_get_representative_frame(pool, &args).await,
         "analyze_frame" => tools::exec_analyze_frame(pool, &args).await,
         "search_by_appearance" => tools::exec_search_by_appearance(pool, &args).await,
+        "vlm_describe" => tools::exec_vlm_describe(pool, &args).await,
         _ => Err(format!("Unknown tool: {}", name)),
     };
     let content = match result {

src/api/cluster_agent.rs

@@ -0,0 +1,238 @@
+use axum::{
+    Extension, Json,
+    extract::{Path, State},
+    http::StatusCode,
+    routing::post,
+    Router,
+};
+use serde::Serialize;
+use std::collections::HashMap;
+use std::path::PathBuf;
+
+use super::middleware::UserAuth;
+use super::types::AppState;
+use crate::core::config::OUTPUT_DIR;
+use crate::core::db::schema;
+use crate::core::processor::{process_face_cluster, FaceClusterResult};
+
+#[derive(Debug, Serialize)]
+pub struct ClusterAgentResponse {
+    success: bool,
+    file_uuid: String,
+    message: String,
+    clusters: usize,
+    total_traces: usize,
+}
+
+pub fn cluster_agent_routes() -> Router<AppState> {
+    Router::new().route(
+        "/api/v1/file/:file_uuid/cluster-agent",
+        post(trigger_cluster_agent),
+    )
+}
+
+pub async fn trigger_cluster_agent(
+    State(state): State<AppState>,
+    Extension(_auth): Extension<UserAuth>,
+    Path(file_uuid): Path<String>,
+) -> Result<Json<ClusterAgentResponse>, (StatusCode, Json<serde_json::Value>)> {
+    let videos_table = schema::table_name("videos");
+    let row = sqlx::query(&format!(
+        "SELECT file_path, COALESCE(total_frames, 0) as total_frames, COALESCE(fps, 25.0) as fps, probe_json FROM {} WHERE file_uuid = $1",
+        videos_table
+    ))
+    .bind(&file_uuid)
+    .fetch_optional(state.db.pool())
+    .await
+    .map_err(|e| {
+        tracing::error!("[ClusterAgent] DB error: {}", e);
+        (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(serde_json::json!({"error": e.to_string()})),
+        )
+    })?;
+
+    let (file_path, total_frames, fps) = match row {
+        Some(r) => {
+            use sqlx::Row;
+            let p: String = r.get("file_path");
+            let tf: i64 = r.get("total_frames");
+            let f: f64 = r.get("fps");
+
+            let mut frames = tf;
+            if frames <= 0 {
+                if let Ok(meta_val) = r.try_get::<serde_json::Value, _>("probe_json") {
+                    if let Some(streams) = meta_val.get("streams").and_then(|v| v.as_array()) {
+                        for s in streams {
+                            if s.get("codec_type").and_then(|v| v.as_str()) == Some("video") {
+                                if let Some(nb) = s.get("nb_frames").and_then(|v| v.as_str()) {
+                                    if let Ok(n) = nb.parse::<i64>() {
+                                        frames = n;
+                                        break;
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+            (p, frames, f)
+        }
+        None => {
+            return Err((
+                StatusCode::NOT_FOUND,
+                Json(serde_json::json!({"error": "File not found"})),
+            ))
+        }
+    };
+
+    let output_path =
+        PathBuf::from(OUTPUT_DIR.as_str()).join(format!("{}.face_cluster.json", file_uuid));
+
+    tracing::info!("[ClusterAgent] Starting face clustering for {}", file_uuid);
+    let result = process_face_cluster(
+        &file_path,
+        output_path.to_str().ok_or_else(|| {
+            (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(serde_json::json!({"error": "Invalid output path"})),
+            )
+        })?,
+        Some(&file_uuid),
+        None,
+    )
+    .await
+    .map_err(|e| {
+        tracing::error!("[ClusterAgent] Clustering failed: {}", e);
+        (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(serde_json::json!({"error": format!("Clustering failed: {}", e)})),
+        )
+    })?;
+
+    update_trace_labels_from_cluster(&state, &file_uuid, &result)
+        .await
+        .map_err(|e| {
+            tracing::error!("[ClusterAgent] Failed to update trace labels: {}", e);
+            (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(serde_json::json!({"error": e.to_string()})),
+            )
+        })?;
+
+    let total_traces = count_total_traces(&state, &file_uuid)
+        .await
+        .map_err(|e| {
+            (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(serde_json::json!({"error": e.to_string()})),
+            )
+        })?;
+
+    tracing::info!(
+        "[ClusterAgent] Completed for {}: {} clusters, {} traces",
+        file_uuid,
+        result.clusters.len(),
+        total_traces
+    );
+
+    Ok(Json(ClusterAgentResponse {
+        success: true,
+        file_uuid,
+        message: "Clustering completed".to_string(),
+        clusters: result.clusters.len(),
+        total_traces,
+    }))
+}
+
+async fn update_trace_labels_from_cluster(
+    state: &AppState,
+    file_uuid: &str,
+    result: &FaceClusterResult,
+) -> anyhow::Result<()> {
+    let scroll_filter = serde_json::json!({
+        "must": [
+            {"key": "file_uuid", "match": {"value": file_uuid}}
+        ]
+    });
+
+    let all_faces = state
+        .qdrant
+        .scroll_all_points("_faces", scroll_filter, 1000)
+        .await?;
+
+    let mut frame_to_traces: HashMap<u64, Vec<i64>> = HashMap::new();
+    for point in &all_faces {
+        let payload = &point["payload"];
+        if let (Some(frame), Some(trace_id)) =
+            (payload["frame"].as_u64(), payload["trace_id"].as_i64())
+        {
+            if trace_id > 0 {
+                frame_to_traces.entry(frame).or_default().push(trace_id);
+            }
+        }
+    }
+
+    let mut trace_to_cluster: HashMap<i64, String> = HashMap::new();
+    for frame_data in &result.frames {
+        let frame = frame_data.frame;
+        if let Some(trace_ids) = frame_to_traces.get(&frame) {
+            for (idx, face) in frame_data.faces.iter().enumerate() {
+                if idx < trace_ids.len() {
+                    trace_to_cluster.insert(trace_ids[idx], face.cluster_id.clone());
+                }
+            }
+        }
+    }
+
+    let tkg_table = schema::table_name("tkg_nodes");
+    for (trace_id, cluster_id) in &trace_to_cluster {
+        let external_id = format!("trace_{}", trace_id);
+        let label = cluster_id.clone();
+
+        let rows_updated = sqlx::query(&format!(
+            "UPDATE {} SET label = $1 WHERE file_uuid = $2 AND node_type = 'face_track' AND external_id = $3",
+            tkg_table
+        ))
+        .bind(&label)
+        .bind(file_uuid)
+        .bind(&external_id)
+        .execute(state.db.pool())
+        .await?
+        .rows_affected();
+
+        if rows_updated == 0 {
+            sqlx::query(&format!(
+                "INSERT INTO {} (file_uuid, node_type, external_id, label, properties) VALUES ($1, 'face_track', $2, $3, $4)",
+                tkg_table
+            ))
+            .bind(file_uuid)
+            .bind(&external_id)
+            .bind(&label)
+            .bind(serde_json::json!({"trace_id": trace_id}))
+            .execute(state.db.pool())
+            .await?;
+        }
+    }
+
+    tracing::info!(
+        "[ClusterAgent] Updated {} trace labels for {}",
+        trace_to_cluster.len(),
+        file_uuid
+    );
+
+    Ok(())
+}
+
+async fn count_total_traces(state: &AppState, file_uuid: &str) -> anyhow::Result<usize> {
+    let tkg_table = schema::table_name("tkg_nodes");
+    let count: (i64,) = sqlx::query_as(&format!(
+        "SELECT COUNT(*) FROM {} WHERE file_uuid = $1 AND node_type = 'face_track'",
+        tkg_table
+    ))
+    .bind(file_uuid)
+    .fetch_one(state.db.pool())
+    .await?;
+
+    Ok(count.0 as usize)
+}

src/api/mod.rs

@@ -2,6 +2,7 @@ pub mod agent_api;
 pub mod agent_search;
 pub mod auth;
 pub mod checkin_api;
+pub mod cluster_agent;
 pub mod docs;
 pub mod files;
 pub mod health;

src/api/server.rs

@@ -13,6 +13,7 @@ use super::agent_api;
 use super::agent_search;
 use super::auth;
 use super::checkin_api;
+use super::cluster_agent;
 use super::docs;
 use super::files;
 use super::health;
@@ -127,6 +128,7 @@ pub async fn start_server(host: &str, port: u16) -> anyhow::Result<()> {
         .merge(universal_search_routes())
         .merge(pipeline::pipeline_routes())
         .merge(checkin_api::checkin_routes())
+        .merge(cluster_agent::cluster_agent_routes())
         .merge(profile::profile_routes())
         .layer(axum::middleware::from_fn_with_state(
             state.api_state.clone(),

src/core/agent/tools.rs

@@ -1,5 +1,6 @@
 use base64::{engine::general_purpose::STANDARD as BASE64, Engine};
 use serde_json;
+use std::time::Duration;
 
 use crate::core::db::qdrant_db::QdrantDb;
 use crate::core::db::schema;
@@ -1233,3 +1234,115 @@ pub async fn exec_search_by_appearance(
         Err("Color search output not found".to_string())
     }
 }
+
+fn face_crop_path(file_uuid: &str, trace_id: i32) -> Option<std::path::PathBuf> {
+    let base = std::env::var("MOMENTRY_OUTPUT_DIR")
+        .unwrap_or_else(|_| "/Users/accusys/momentry/output".to_string());
+    let dir = std::path::PathBuf::from(base)
+        .join(".faces")
+        .join(file_uuid)
+        .join(trace_id.to_string());
+    if !dir.exists() {
+        return None;
+    }
+    let mut entries: Vec<_> = match std::fs::read_dir(&dir) {
+        Ok(e) => e.filter_map(|e| e.ok()).collect(),
+        Err(_) => return None,
+    };
+    entries.sort_by_key(|e| e.file_name());
+    entries.first().map(|e| e.path())
+}
+
+pub async fn exec_vlm_describe(
+    pool: &sqlx::PgPool,
+    args: &serde_json::Value,
+) -> Result<String, String> {
+    let file_uuid = args.get("file_uuid").and_then(|v| v.as_str()).unwrap_or("");
+    let trace_id = args.get("trace_id").and_then(|v| v.as_i64()).unwrap_or(0) as i32;
+    let prompt = args
+        .get("prompt")
+        .and_then(|v| v.as_str())
+        .unwrap_or("Describe this person's clothing and appearance. Focus on colors, clothing type, and any distinctive visual features.");
+
+    if file_uuid.is_empty() {
+        return Ok(serde_json::json!({"error": "file_uuid is required"}).to_string());
+    }
+    if trace_id <= 0 {
+        return Ok(serde_json::json!({"error": "trace_id is required and must be > 0"}).to_string());
+    }
+
+    let crop_path = face_crop_path(file_uuid, trace_id)
+        .ok_or_else(|| format!("No face crop found for {} trace {}", file_uuid, trace_id))?;
+
+    let jpeg_bytes = std::fs::read(&crop_path)
+        .map_err(|e| format!("Failed to read face crop: {}", e))?;
+
+    let videos = schema::table_name("videos");
+    let fps: f64 = sqlx::query_scalar(&format!(
+        "SELECT COALESCE(fps, 25.0) FROM {} WHERE file_uuid = $1",
+        videos
+    ))
+    .bind(file_uuid)
+    .fetch_optional(pool)
+    .await
+    .map_err(|e| e.to_string())?
+    .unwrap_or(25.0);
+
+    let frame_name = crop_path
+        .file_stem()
+        .and_then(|s| s.to_str())
+        .and_then(|s| s.parse::<i64>().ok())
+        .unwrap_or(0);
+    let timestamp_secs = frame_name as f64 / fps;
+
+    let base64_img = BASE64.encode(&jpeg_bytes);
+
+    let ollama_url = std::env::var("OLLAMA_URL")
+        .unwrap_or_else(|_| "http://localhost:11434".to_string());
+    let model = std::env::var("VLM_MODEL").unwrap_or_else(|_| "llava".to_string());
+
+    let body = serde_json::json!({
+        "model": model,
+        "prompt": prompt,
+        "images": [base64_img],
+        "stream": false,
+        "options": {
+            "num_predict": 80
+        }
+    });
+
+    let client = reqwest::Client::builder()
+        .timeout(Duration::from_secs(30))
+        .build()
+        .map_err(|e| format!("Failed to create HTTP client: {}", e))?;
+
+    let resp = client
+        .post(format!("{}/api/generate", ollama_url))
+        .json(&body)
+        .send()
+        .await
+        .map_err(|e| format!("Ollama request failed: {}", e))?;
+
+    let resp_json: serde_json::Value = resp
+        .json()
+        .await
+        .map_err(|e| format!("Failed to parse Ollama response: {}", e))?;
+
+    let description = resp_json
+        .get("response")
+        .and_then(|v| v.as_str())
+        .unwrap_or("No description returned")
+        .to_string();
+
+    Ok(serde_json::json!({
+        "tool": "vlm_describe",
+        "result": {
+            "file_uuid": file_uuid,
+            "trace_id": trace_id,
+            "frame": frame_name,
+            "description": description,
+            "time_sec": (timestamp_secs * 100.0).round() / 100.0
+        }
+    })
+    .to_string())
+}

src/core/chunk/rule1_ingest.rs

@@ -30,6 +30,14 @@ pub async fn execute_rule1(db: &PostgresDb, file_uuid: &str, fps: f64) -> Result
     let mut count = 0;
     let mut tx = pool.begin().await?;
 
+    // Delete existing chunks for this file before re-inserting
+    let chunk_table = schema::table_name("chunk");
+    sqlx::query(&format!("DELETE FROM {} WHERE file_uuid = $1", chunk_table))
+        .bind(file_uuid)
+        .execute(&mut *tx)
+        .await?;
+    info!("Rule 1: Deleted old chunks for video {}", file_uuid);
+
     // Phase 1: ASRX segments (pure speech, NO OCR merge)
     for seg in asr_segments.iter() {
         // Skip chunks with no text
@@ -97,7 +105,7 @@ pub async fn execute_rule1(db: &PostgresDb, file_uuid: &str, fps: f64) -> Result
             file_id as i32,
             file_uuid.to_string(),
             format!("{}", count),
-            ChunkType::Sentence,
+            ChunkType::Ocr,
             ChunkRule::Rule1,
             start_time,
             end_time,

src/core/db/postgres_db.rs

@@ -824,7 +824,7 @@ pub struct PostgresCache {
 #[derive(Debug, serde::Serialize, sqlx::FromRow)]
 pub struct SemanticSearchResult {
     pub id: i32,
-    pub file_uuid: Option<String>, // Added for global search
+    pub file_uuid: Option<String>,
     pub scene_order: i32,
     pub start_frame: i64,
     pub end_frame: i64,
@@ -836,6 +836,7 @@ pub struct SemanticSearchResult {
     pub metadata: Option<serde_json::Value>,
     pub similarity: Option<f64>,
     pub content: Option<serde_json::Value>,
+    pub chunk_type: String,
 }
 
 /// Result structure for child chunks
@@ -2519,7 +2520,8 @@ impl PostgresDb {
                      text_content, \
                      metadata, \
                      (1 - (embedding <=> $1::vector)) as similarity, \
-                     content \
+                     content, \
+                     COALESCE(chunk_type, 'sentence') as chunk_type \
                  FROM {} \
                 WHERE file_uuid = $2 AND chunk_type IN ('sentence', 'story_parent', 'llm_parent') AND embedding IS NOT NULL \
                 ORDER BY embedding <=> $1::vector \
@@ -2557,7 +2559,8 @@ impl PostgresDb {
                      text_content, \
                      metadata, \
                      (1 - (embedding <=> $1::vector)) as similarity, \
-                     content \
+                     content, \
+                     COALESCE(chunk_type, 'sentence') as chunk_type \
                  FROM {} \
                 WHERE chunk_type IN ('sentence', 'story_parent', 'llm_parent') AND embedding IS NOT NULL \
                 ORDER BY embedding <=> $1::vector \
@@ -2590,7 +2593,8 @@ impl PostgresDb {
                      text_content as text_content, \
                      metadata, \
                      1.0::float8 as similarity, \
-                     content \
+                     content, \
+                     COALESCE(chunk_type, 'sentence') as chunk_type \
                  FROM {} \
                  WHERE file_uuid = $1 AND chunk_id = $2 AND embedding IS NOT NULL \
                  LIMIT 1",
@@ -2622,7 +2626,8 @@ impl PostgresDb {
                      text_content as text_content, \
                      metadata, \
                      1.0::float8 as similarity, \
-                     content \
+                     content, \
+                     COALESCE(chunk_type, 'sentence') as chunk_type \
                  FROM {} \
                  WHERE file_uuid = $1 AND chunk_id = $2 \
                  LIMIT 1",
@@ -2885,7 +2890,7 @@ impl PostgresDb {
         tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
     ) -> Result<()> {
         let table = schema::table_name("chunk");
-        let ct_str = format!("{:?}", chunk.chunk_type).to_lowercase();
+        let ct_str = chunk.chunk_type.as_str();
         let fps = chunk.fps;
         let start_time = chunk.start_frame as f64 / chunk.fps;
         let end_time = chunk.end_frame as f64 / chunk.fps;
@@ -3419,45 +3424,18 @@ impl PostgresDb {
         let like = format!("%{}%", query.replace('%', "%%"));
         use sqlx::Row;
 
-        // Check if query contains CJK characters
+        // Simple LIKE-based matching (no FTS stemming)
-        let has_cjk = query.chars().any(|c| {
+        let sql = format!(
-            ('\u{4E00}'..='\u{9FFF}').contains(&c)
+            "SELECT chunk_id, file_uuid, chunk_type, text_content, start_time, end_time, \
-                || ('\u{3040}'..='\u{309F}').contains(&c)
+             (1.0 - (POSITION(LOWER($1) IN LOWER(text_content))::float8 / NULLIF(LENGTH(text_content), 0)::float8))::float8 as score \
-                || ('\u{30A0}'..='\u{30FF}').contains(&c)
+             FROM {} \
-                || ('\u{AC00}'..='\u{D7AF}').contains(&c)
+             WHERE text_content ILIKE $2 AND text_content != '' \
-        });
+             {}\
-
+             ORDER BY score DESC \
-        let sql = if has_cjk {
+             LIMIT $3",
-            // CJK/Korean: use ILIKE position-based ranking
+            table,
-            format!(
+            if file_uuid.is_some() { "AND file_uuid = $4 " } else { "" }
-                "SELECT chunk_id, file_uuid, chunk_type, text_content, start_time, end_time, \
+        );
-                 (1.0 - (POSITION(LOWER($1) IN LOWER(text_content))::float8 / NULLIF(LENGTH(text_content), 0)::float8))::float8 as score \
-                 FROM {} \
-                 WHERE text_content ILIKE $2 AND text_content != '' \
-                 {}\
-                 ORDER BY score DESC \
-                 LIMIT $3",
-                table,
-                if file_uuid.is_some() { "AND file_uuid = $4 " } else { "" }
-            )
-        } else {
-            // English: use PostgreSQL full-text search
-            format!(
-                "SELECT chunk_id, file_uuid, chunk_type, text_content, start_time, end_time, \
-                 CASE \
-                      WHEN to_tsvector('english', text_content) @@ plainto_tsquery('english', $1) \
-                      THEN ts_rank(to_tsvector('english', text_content), plainto_tsquery('english', $1))::float8 \
-                      ELSE 0.1::float8 \
-                  END as score \
-                 FROM {} \
-                 WHERE text_content ILIKE $2 AND text_content != '' \
-                 {}\
-                 ORDER BY score DESC \
-                 LIMIT $3",
-                table,
-                if file_uuid.is_some() { "AND file_uuid = $4 " } else { "" }
-            )
-        };
 
         let rows = if let Some(u) = file_uuid {
             sqlx::query(&sql)
@@ -4309,7 +4287,7 @@ impl PostgresDb {
 
     pub async fn store_chunk(&self, chunk: &crate::core::chunk::types::Chunk) -> Result<()> {
         let table = schema::table_name("chunk");
-        let ct_str = format!("{:?}", chunk.chunk_type).to_lowercase();
+        let ct_str = chunk.chunk_type.as_str();
         let start_time = chunk.start_frame as f64 / chunk.fps;
         let end_time = chunk.end_frame as f64 / chunk.fps;
         sqlx::query(&format!(
@@ -4550,7 +4528,7 @@ impl Database for PostgresDb {
 impl crate::core::db::ChunkStore for PostgresDb {
     async fn store_chunk(&self, chunk: &crate::core::chunk::types::Chunk) -> Result<()> {
         let table = schema::table_name("chunk");
-        let ct_str = format!("{:?}", chunk.chunk_type).to_lowercase();
+        let ct_str = chunk.chunk_type.as_str();
         let start_time = chunk.start_frame as f64 / chunk.fps;
         let end_time = chunk.end_frame as f64 / chunk.fps;
         sqlx::query(&format!(

src/worker/job_worker.rs

@@ -1,4 +1,4 @@
-use anyhow::Result;
+use anyhow::{Context, Result};
 use std::collections::HashMap;
 use std::path::PathBuf;
 use std::sync::Arc;
@@ -70,6 +70,14 @@ impl JobWorker {
             self.config.max_concurrent, self.config.poll_interval_secs
         );
 
+        // Layer 3: Health check on startup - repair incomplete jobs
+        if let Err(e) = self.health_check_incomplete_jobs().await {
+            warn!("[HEALTH-CHECK] Initial health check failed: {}", e);
+        }
+
+        let mut last_health_check = std::time::Instant::now();
+        let health_check_interval = std::time::Duration::from_secs(300); // 5 minutes
+
         loop {
             if !self.config.enabled {
                 warn!("Worker is disabled, sleeping for 60s...");
@@ -77,6 +85,14 @@ impl JobWorker {
                 continue;
             }
 
+            // Layer 3: Periodic health check every 5 minutes
+            if last_health_check.elapsed() > health_check_interval {
+                if let Err(e) = self.health_check_incomplete_jobs().await {
+                    warn!("[HEALTH-CHECK] Periodic health check failed: {}", e);
+                }
+                last_health_check = std::time::Instant::now();
+            }
+
             if let Err(e) = self.poll_and_process().await {
                 error!("Error during poll and process: {}", e);
             }
@@ -280,6 +296,80 @@ impl JobWorker {
         Ok(())
     }
 
+    /// Layer 2: Auto-repair missing processor_results based on existing output files
+    async fn check_and_repair_processor_results(&self, job: &crate::core::db::MonitorJob) -> Result<()> {
+        use std::path::Path;
+        
+        // Get output directory from environment
+        let output_dir = std::env::var("MOMENTRY_OUTPUT_DIR")
+            .unwrap_or_else(|_| "/Users/accusys/momentry/output".to_string());
+        
+        // Get current processor_results
+        let existing_results = match self.db.get_latest_processor_results_by_file_uuid(&job.uuid).await {
+            Ok(r) => r,
+            Err(e) => {
+                warn!("[AUTO-REPAIR] Failed to get existing results for {}: {}", job.uuid, e);
+                return Err(e);
+            }
+        };
+        
+        // Check all processors
+        let processors = crate::core::db::ProcessorType::all();
+        let mut repaired_count = 0;
+        
+        for processor_type in processors {
+            let output_path = format!("{}/{}.{}.json", output_dir, job.uuid, processor_type.as_str());
+            
+            // If file exists but no DB record
+            if Path::new(&output_path).exists() {
+                let has_db_record = existing_results.iter().any(|r| 
+                    r.processor_type == processor_type && r.status == crate::core::db::ProcessorJobStatus::Completed
+                );
+                
+                if !has_db_record {
+                    warn!("[AUTO-REPAIR] Creating missing processor_result for {} (file: {})", 
+                          processor_type.as_str(), job.uuid);
+                    
+                    if let Err(e) = self.db.upsert_processor_result(job.id, processor_type, &job.uuid, "completed").await {
+                        error!("[AUTO-REPAIR] Failed to create processor_result for {}: {}", processor_type.as_str(), e);
+                    } else {
+                        repaired_count += 1;
+                    }
+                }
+            }
+        }
+        
+        if repaired_count > 0 {
+            info!("[AUTO-REPAIR] Repaired {} processor_results for {}", repaired_count, job.uuid);
+        }
+        
+        Ok(())
+    }
+
+    /// Layer 3: Periodic health check for incomplete jobs
+    async fn health_check_incomplete_jobs(&self) -> Result<()> {
+        info!("[HEALTH-CHECK] Starting periodic health check");
+        
+        // Get all incomplete jobs
+        let jobs = self.db.get_pending_jobs(100).await?;
+        
+        let mut total_repaired = 0;
+        
+        for job in jobs {
+            if let Err(e) = self.check_and_repair_processor_results(&job).await {
+                warn!("[HEALTH-CHECK] Failed to repair job {}: {}", job.uuid, e);
+            } else {
+                total_repaired += 1;
+            }
+        }
+        
+        if total_repaired > 0 {
+            info!("[HEALTH-CHECK] Checked {} jobs", total_repaired);
+        }
+        
+        Ok(())
+    }
+
     async fn process_job(&self, job: crate::core::db::MonitorJob) -> Result<()> {
         // Check if job still exists in database (may have been deleted by unregister)
         let current_job = self.db.get_monitor_job_by_uuid(&job.uuid).await?;
@@ -345,6 +435,11 @@ impl JobWorker {
             .update_worker_job_status(&job.uuid, job.id, "running", None, 0, total_processor_types)
             .await?;
 
+        // Layer 2: Auto-repair missing processor_results based on existing output files
+        if let Err(e) = self.check_and_repair_processor_results(&job).await {
+            warn!("[AUTO-REPAIR] Failed to check processor_results for {}: {}", job.uuid, e);
+        }
+
         // Get existing processor results for this job AND completed results from previous jobs
         let existing_results = self.db.get_processor_results_by_job(job.id).await?;
         let mut result_map = HashMap::new();
@@ -480,13 +575,44 @@ impl JobWorker {
                         0,
                     )
                     .await?;
-                if let Err(e) = self
+                // Layer 1: Ensure write success with retry
-                    .db
+                let mut success = false;
-                    .upsert_processor_result(job.id, *processor_type, &job.uuid, "completed")
+                for attempt in 1..=3 {
-                    .await
+                    match self.db.upsert_processor_result(job.id, *processor_type, &job.uuid, "completed").await {
-                {
+                        Ok(_) => {
-                    error!("Failed to create completed processor result: {}", e);
+                            success = true;
+                            break;
+                        }
+                        Err(e) if attempt < 3 => {
+                            warn!("[RETRY {}] upsert_processor_result failed for {}: {}", attempt, processor_type.as_str(), e);
+                            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+                        }
+                        Err(e) => {
+                            error!("[FATAL] Failed to record {} completion after 3 retries: {}", processor_type.as_str(), e);
+                        }
+                    }
                 }
+                if !success {
+                    warn!("[CONTINUE] {} completed but processor_result not recorded", processor_type.as_str());
+                }
+                
+                // Special handling for FaceCluster: always update tkg_nodes if file exists
+                if *processor_type == crate::core::db::ProcessorType::FaceCluster {
+                    if let Ok(json_str) = std::fs::read_to_string(&output_path) {
+                        if let Ok(result) = serde_json::from_str::<
+                            crate::core::processor::FaceClusterResult,
+                        >(&json_str)
+                        {
+                            info!("[FaceCluster-SKIP] Updating tkg_nodes from existing file, {} clusters", result.clusters.len());
+                            if let Err(e) =
+                                Self::update_face_track_labels(&self.db, &job.uuid, &result).await
+                            {
+                                error!("Failed to update face track labels: {}", e);
+                            }
+                        }
+                    }
+                }
+                
                 // Load output file and store to pre_chunks
                 // Also dual-write to workspace if available
                 let workspace = WorkspaceDb::open(&job.uuid).await.ok();
@@ -746,6 +872,20 @@ impl JobWorker {
                             }
                         }
                         crate::core::db::ProcessorType::FaceCluster => {
+                            if let Ok(result) = serde_json::from_str::<
+                                crate::core::processor::FaceClusterResult,
+                            >(&json_str)
+                            {
+                                info!("[FaceCluster] Parsing succeeded, {} clusters, {} frames", result.clusters.len(), result.frames.len());
+                                if let Err(e) =
+                                    Self::update_face_track_labels(&self.db, &job.uuid, &result)
+                                        .await
+                                {
+                                    error!("Failed to update face track labels: {}", e);
+                                }
+                            } else {
+                                warn!("[FaceCluster] Failed to parse FaceClusterResult from json_str (len={})", json_str.len());
+                            }
                             info!("Face clustering processor completed for {}", job.uuid);
                             Ok(())
                         }
@@ -1235,6 +1375,13 @@ impl JobWorker {
                                 );
                                 return true;
                             }
+                            if status == "error" {
+                                tracing::error!(
+                                    "[Ingestion] Face trace failed for {} - marking as done (with error)",
+                                    uuid
+                                );
+                                return true;
+                            }
                         }
 
                         if let Some(traces) = traced_data.get("traces") {
@@ -1786,7 +1933,19 @@ impl JobWorker {
                                     .await;
                             }
                             Err(e) => {
-                                error!("❌ Face trace + DB store failed for {}: {}", uuid_clone, e)
+                                error!("❌ Face trace + DB store failed for {}: {}", uuid_clone, e);
+                                // Write fallback face_traced.json to unblock ingestion check
+                                let output_dir = std::env::var("MOMENTRY_OUTPUT_DIR")
+                                    .unwrap_or_else(|_| "/Users/accusys/momentry/output".to_string());
+                                let fallback_path = format!("{}/{}.face_traced.json", output_dir, uuid_clone);
+                                let fallback_content = serde_json::json!({
+                                    "status": "error",
+                                    "error": e.to_string(),
+                                    "traces": {}
+                                });
+                                if let Err(write_err) = std::fs::write(&fallback_path, fallback_content.to_string()) {
+                                    error!("Failed to write fallback face_traced.json for {}: {}", uuid_clone, write_err);
+                                }
                             }
                         }
                     });
@@ -2318,6 +2477,91 @@ impl JobWorker {
         );
         Ok(())
     }
+
+    async fn update_face_track_labels(
+        db: &crate::core::db::PostgresDb,
+        uuid: &str,
+        result: &crate::core::processor::FaceClusterResult,
+    ) -> Result<()> {
+        use std::collections::HashMap;
+
+        let qdrant = crate::core::db::qdrant_db::QdrantDb::new();
+        let scroll_filter = serde_json::json!({
+            "must": [
+                {"key": "file_uuid", "match": {"value": uuid}}
+            ]
+        });
+
+        let all_faces = qdrant
+            .scroll_all_points("_faces", scroll_filter, 1000)
+            .await
+            .context("Failed to scroll _faces from Qdrant")?;
+
+        let mut frame_to_traces: HashMap<u64, Vec<i64>> = HashMap::new();
+        for point in &all_faces {
+            let payload = &point["payload"];
+            if let (Some(frame), Some(trace_id)) = (
+                payload["frame"].as_u64(),
+                payload["trace_id"].as_i64(),
+            ) {
+                if trace_id > 0 {
+                    frame_to_traces.entry(frame).or_default().push(trace_id);
+                }
+            }
+        }
+
+        let mut trace_to_cluster: HashMap<i64, String> = HashMap::new();
+        for frame_data in &result.frames {
+            let frame = frame_data.frame;
+            if let Some(trace_ids) = frame_to_traces.get(&frame) {
+                for (idx, face) in frame_data.faces.iter().enumerate() {
+                    if idx < trace_ids.len() {
+                        trace_to_cluster.insert(trace_ids[idx], face.cluster_id.clone());
+                    }
+                }
+            }
+        }
+
+        let tkg_table = crate::core::db::schema::table_name("tkg_nodes");
+        for (trace_id, cluster_id) in &trace_to_cluster {
+            let external_id = format!("trace_{}", trace_id);
+            let label = cluster_id.clone();
+            
+            let rows_updated = sqlx::query(&format!(
+                "UPDATE {} SET label = $1 WHERE file_uuid = $2 AND node_type = 'face_track' AND external_id = $3",
+                tkg_table
+            ))
+            .bind(&label)
+            .bind(uuid)
+            .bind(&external_id)
+            .execute(db.pool())
+            .await
+            .context("Failed to update tkg_nodes label")?
+            .rows_affected();
+
+            if rows_updated == 0 {
+                sqlx::query(&format!(
+                    "INSERT INTO {} (file_uuid, node_type, external_id, label, properties) VALUES ($1, 'face_track', $2, $3, $4)",
+                    tkg_table
+                ))
+                .bind(uuid)
+                .bind(&external_id)
+                .bind(&label)
+                .bind(serde_json::json!({"trace_id": trace_id}))
+                .execute(db.pool())
+                .await
+                .context("Failed to insert tkg_nodes")?;
+            }
+        }
+
+        info!(
+            "[FaceCluster] Updated {} trace labels for {}",
+            trace_to_cluster.len(),
+            uuid
+        );
+
+        Ok(())
+    }
 }
 
 #[cfg(test)]