From 7dc910e2de53063efa5d5ec2808acd494339446a Mon Sep 17 00:00:00 2001 From: Accusys Date: Mon, 20 Jul 2026 21:48:27 +0800 Subject: [PATCH] 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 | 15 +- src/api/cluster_agent.rs | 238 ++++++++++++++++++++++++++++++ src/api/mod.rs | 1 + src/api/server.rs | 2 + src/core/agent/tools.rs | 113 ++++++++++++++ src/core/chunk/rule1_ingest.rs | 10 +- src/core/db/postgres_db.rs | 72 ++++----- src/worker/job_worker.rs | 260 ++++++++++++++++++++++++++++++++- 8 files changed, 654 insertions(+), 57 deletions(-) create mode 100644 src/api/cluster_agent.rs diff --git a/src/api/agent_search.rs b/src/api/agent_search.rs index e588ba0..19858f5 100644 --- a/src/api/agent_search.rs +++ b/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 { }), 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 { diff --git a/src/api/cluster_agent.rs b/src/api/cluster_agent.rs new file mode 100644 index 0000000..9f699cd --- /dev/null +++ b/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 { + Router::new().route( + "/api/v1/file/:file_uuid/cluster-agent", + post(trigger_cluster_agent), + ) +} + +pub async fn trigger_cluster_agent( + State(state): State, + Extension(_auth): Extension, + Path(file_uuid): Path, +) -> Result, (StatusCode, Json)> { + 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::("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::() { + 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> = 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 = 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 { + 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) +} diff --git a/src/api/mod.rs b/src/api/mod.rs index 62875ef..5b6e94f 100644 --- a/src/api/mod.rs +++ b/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; diff --git a/src/api/server.rs b/src/api/server.rs index d065b3d..5ec0120 100644 --- a/src/api/server.rs +++ b/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(), diff --git a/src/core/agent/tools.rs b/src/core/agent/tools.rs index d8b7fc8..af86d91 100644 --- a/src/core/agent/tools.rs +++ b/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 { + 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 { + 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::().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()) +} diff --git a/src/core/chunk/rule1_ingest.rs b/src/core/chunk/rule1_ingest.rs index 3117cf5..0cb1d72 100644 --- a/src/core/chunk/rule1_ingest.rs +++ b/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, diff --git a/src/core/db/postgres_db.rs b/src/core/db/postgres_db.rs index 8d99c14..ec7802b 100644 --- a/src/core/db/postgres_db.rs +++ b/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, // Added for global search + pub file_uuid: Option, pub scene_order: i32, pub start_frame: i64, pub end_frame: i64, @@ -836,6 +836,7 @@ pub struct SemanticSearchResult { pub metadata: Option, pub similarity: Option, pub content: Option, + 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 - let has_cjk = query.chars().any(|c| { - ('\u{4E00}'..='\u{9FFF}').contains(&c) - || ('\u{3040}'..='\u{309F}').contains(&c) - || ('\u{30A0}'..='\u{30FF}').contains(&c) - || ('\u{AC00}'..='\u{D7AF}').contains(&c) - }); - - let sql = if has_cjk { - // CJK/Korean: use ILIKE position-based ranking - format!( - "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 { "" } - ) - }; + // Simple LIKE-based matching (no FTS stemming) + let sql = format!( + "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 { "" } + ); 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!( diff --git a/src/worker/job_worker.rs b/src/worker/job_worker.rs index 5524d97..e1ea7d4 100644 --- a/src/worker/job_worker.rs +++ b/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 - .db - .upsert_processor_result(job.id, *processor_type, &job.uuid, "completed") - .await - { - error!("Failed to create completed processor result: {}", e); + // Layer 1: Ensure write success with retry + let mut success = false; + for attempt in 1..=3 { + match self.db.upsert_processor_result(job.id, *processor_type, &job.uuid, "completed").await { + Ok(_) => { + 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> = 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 = 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)]