fix: production service auto-start via launchd + fix worker retry loop

- Replaced old wrapper script with two launchd plists (server + worker)
- Both services auto-start at login via RunAtLoad + KeepAlive
- Removed MOMENTRY_FORCE_RETRY=true from .env to prevent infinite retries
- Added REDIS_URL and all required env vars to launchd plists
- Truncated 3.6GB of oversized logs (2.3G worker + 1.3G launchd stdout)
- Deleted 1214 orphaned processor_results records
- Updated run-server-3002.sh to manage via launchctl
- Removed obsolete wrapper scripts

docs_v1.0/API_WORKSPACE/modules/10_pipeline.md

@@ -42,6 +42,7 @@ These steps run after the 10 processors and are **required for pipeline completi
 | # | Step | Triggers When | Verification |
 |---|------|--------------|-------------|
 | 1 | **Rule 1 Sentence Chunking** | ASR + ASRX done | `chunk` table has rows with `chunk_type = 'sentence'` |
+| 1.1 | **Rule 1 OCR Chunks** | OCR done | OCR pre_chunks grouped into sentence chunks |
 | 2 | **Auto-Vectorize** | Rule 1 done | `chunk.embedding` IS NOT NULL for sentence chunks |
 | 3 | **Phase 1 Pack** | Rule 1 done | `release_pack.py --phase 1` executed |
 | 4 | **Rule 3 Scene Chunking** | All 10 processors done + Cut + ASR | `chunk` table has rows with `chunk_type = 'cut'` |
@@ -81,7 +82,9 @@ curl "$API/api/v1/stats/ingestion-status/bd80fec9c42afb0307eb28f22c64c76a" | jq
 {
   "file_uuid": "bd80fec9c42afb0307eb28f22c64c76a",
   "steps": [
-    { "name": "rule1_sentence", "status": "pending", "detail": "0 sentence chunks" },
+    { "name": "rule1_sentence",   "status": "done",    "detail": "35 sentence chunks" },
+    { "name": "rule1_ocr",        "status": "done",    "detail": "30 OCR frames" },
+    { "name": "rule1_ocr_chunks", "status": "done",    "detail": "3 OCR-only chunks" },
     { "name": "auto_vectorize",   "status": "pending", "detail": "0 embedded" },
     { "name": "rule3_scene",      "status": "pending", "detail": "0 scene chunks" },
     { "name": "face_trace",       "status": "pending", "detail": "0 traces" },

docs_v1.0/DESIGN/Audio_Scene_Detection_POC.md

@@ -0,0 +1,162 @@
+---
+title: Audio Scene & Instrument Detection POC Plan
+version: 0.1
+date: 2026-07-02
+author: OpenCode
+status: planned
+---
+
+| scope | status | applicable to |
+|-------|--------|---------------|
+| Audio processing pipeline | planned | Video files with non-speech audio |
+
+## Goal
+
+Detect non-speech audio events (instruments, music, environmental sounds) in video files alongside existing ASRX speech recognition.
+
+## Why
+
+Current pipeline only detects speech (ASRX → 64 segments + 1554 speaker embeddings). Instrument sounds, background music, and environmental audio are completely ignored.
+
+## Technical Options
+
+### Option A: PANNs (Pre-trained Audio Neural Networks)
+- **Model**: Cnn14 (313M params, 700MB weights)
+- **Classes**: 527 AudioSet classes (piano, guitar, drums, speech, etc.)
+- **Pros**: Production-ready, accurate, PyTorch-based
+- **Cons**: Large download, ~200MB RAM per inference
+- **Install**: `pip install panns-inference`
+
+### Option B: YAMNet (Google)
+- **Model**: MobileNet-based, 4MB weights
+- **Classes**: 521 AudioSet classes
+- **Pros**: Lightweight, fast
+- **Cons**: Requires TensorFlow (not currently installed)
+- **Install**: `pip install yamnet` + TensorFlow
+
+### Option C: torchaudio + heuristics (lightweight fallback)
+- Use existing PyTorch + torchaudio
+- Extract spectral features (MFCC, centroid, energy)
+- Simple classification: speech vs music vs silence
+- **Pros**: No extra dependencies
+- **Cons**: Less accurate, limited classes
+
+## Recommended: Option A (PANNs)
+
+## Pipeline Integration
+
+```
+Video → Audio Extract → ASRX (speech)  → Speaker Embeddings (3.4/s)
+                  → Audio Scene (new) → Scene Labels (1/s)
+```
+
+### New Processor: `audio_scene`
+
+| Field | Value |
+|-------|-------|
+| Processor type | `audio_scene` |
+| Input | Video file (audio track) |
+| Output | `file_uuid.audio_scene.json` |
+| Sampling | 1-second segments |
+| Qdrant collection | `momentry_{schema}_audio_scene` |
+
+### Output Format
+
+```json
+{
+  "file_uuid": "...",
+  "segments": [
+    {
+      "start_time": 0.0,
+      "end_time": 1.0,
+      "primary_class": "speech",
+      "confidence": 0.95,
+      "top_classes": [
+        {"class": "speech", "score": 0.95},
+        {"class": "music", "score": 0.03},
+        {"class": "piano", "score": 0.01}
+      ]
+    }
+  ],
+  "summary": {
+    "speech_ratio": 0.72,
+    "music_ratio": 0.15,
+    "silence_ratio": 0.08,
+    "instrument_ratio": 0.05,
+    "instruments_detected": ["piano", "guitar"]
+  }
+}
+```
+
+### Qdrant Storage
+
+| Field | Type | Purpose |
+|-------|------|---------|
+| `file_uuid` | string | Filter by file |
+| `start_time` | float | Segment start |
+| `end_time` | float | Segment end |
+| `primary_class` | keyword | Filter by class |
+| `confidence` | float | Filter by confidence |
+| `instrument_name` | keyword | Search by instrument |
+| `vector` | f32[2048] | Audio embedding for similarity search |
+
+### Processor Dependencies
+
+```
+audio_scene → (no dependencies, runs parallel with ASRX)
+```
+
+## Key AudioSet Instrument Classes
+
+| Category | Classes |
+|----------|---------|
+| Piano | Piano, Electric piano, Keyboard |
+| Guitar | Guitar, Electric guitar, Acoustic guitar |
+| Drums | Drum kit, Snare drum, Cymbal, Hi-hat |
+| Strings | Violin, Cello, Harp, Double bass |
+| Wind | Flute, Saxophone, Trumpet, Clarinet |
+| Voice | Speech, Singing, Chant, Choir |
+| Other | Music, Percussion, Organ, Synthesizer |
+
+## POC Steps
+
+1. **Install panns-inference**
+   ```bash
+   pip install panns-inference
+   ```
+
+2. **Create `scripts/audio_scene_processor.py`**
+   - Load audio via ffmpeg → numpy array
+   - Process 1-second segments through Cnn14
+   - Save results to JSON + Qdrant
+
+3. **Add processor type to pipeline**
+   - Add `AudioScene` to `ProcessorType` enum
+   - Add to worker's processor dispatch
+   - Add `AUDIO_SCENE_TIMEOUT` config
+
+4. **Test with existing video**
+   - Run on KOBA interview video
+   - Verify instrument detection accuracy
+   - Check performance (time, memory)
+
+5. **Integrate with search**
+   - Add audio_scene to universal_search
+   - Add filter by audio class (speech/music/instrument)
+
+## Estimated Effort
+
+| Step | Time |
+|------|------|
+| Install + prototype script | 2-3 hours |
+| Pipeline integration | 1-2 hours |
+| Qdrant + search integration | 1 hour |
+| Testing + tuning | 1-2 hours |
+| **Total** | **5-8 hours** |
+
+## Future Enhancements
+
+- Real-time audio classification during processing
+- Audio event timeline visualization
+- Combine with TKG for audio-visual relationships
+- Background music detection for copyright checks

docs_v1.0/REFERENCE/Searchable_Chunk_Rules.md

@@ -1,6 +1,7 @@
 # Searchable Chunk — 綜合規則組成
 
 **Date**: 2026-05-16  
+**Updated**: 2026-07-05 — OCR 獨立 chunks (方案 A)
 
 ---
 
@@ -11,8 +12,8 @@ Searchable chunk 不是原始的 cut 或 sentence，而是經過規則組合後
 ```
 原始資料               規則組合                 可搜尋 chunk
 ─────────             ──────────               ──────────────
-ASR sentence (聽覺)    ─┐
+ASRX sentence (聽覺)   ─┐
-YOLO objects (視覺)    ─┤  Rule 1 / Rule 2      chunk (text + metadata + embedding)
+OCR text (視覺文字)    ─┤  Rule 1              chunk (text + metadata + embedding)
 Cut boundary (鏡頭)    ─┘
 ```
 
@@ -21,27 +22,64 @@ Cut boundary (鏡頭)    ─┘
 | 層級 | 類型 | 說明 | 可搜尋 |
 |------|------|------|:------:|
 | **原始** | `cut` | 視覺 chunk（鏡頭） | ❌（無文字） |
-| **原始** | `sentence` | 聽覺 chunk（ASR 句子） | ✅ 文字搜尋 |
+| **原始** | `sentence` | 聽覺 chunk（ASRX 句子） | ✅ 文字搜尋 |
+| **原始** | `sentence` | OCR-only chunk（純視覺文字） | ✅ 文字搜尋 |
 | **合成** | `story_child` | 故事子句 | ✅ |
 | **合成** | `story_parent` | 故事段落（多句聚合） | ✅ |
 
-## Rule 1 — 直接轉換
+## Rule 1 — 雙階段轉換
 
-最簡單的規則。ASR 輸出的每個 sentence 直接成為 chunk，不做聚合。
+### Phase 1: ASRX Segments（純語音）
+
+ASRX 輸出的每個 segment 直接成為 chunk，**不合併 OCR 文字**。
 
 ```json
 {
+  "chunk_id": "0",
+  "chunk_type": "sentence",
   "rule": "rule_1",
-  "data": {
+  "text": "And speaking of storage and workflow...",
-    "text": "I'm in scoby.",
+  "ocr_text": "",
-    "text_normalized": "i'm in scoby."
+  "start_time": 0.0,
-  }
+  "end_time": 5.4
 }
 ```
 
 - `chunk_type = 'sentence'`
-- 可文字搜尋（`text_content ILIKE`）
+- `content.text` = ASRX 語音文字
-- 可向量搜尋（embedding in Qdrant）
+- `content.ocr_text` = ""（空）
+- `text_content` = ASRX 文字
+
+### Phase 2: OCR-only Chunks（純視覺文字）
+
+所有 OCR 幀按鄰近性分組（間距 ≤ 5 幀），每個群組成為獨立 chunk。
+
+```json
+{
+  "chunk_id": "32",
+  "chunk_type": "sentence",
+  "rule": "rule_1",
+  "text": "",
+  "ocr_text": "Accusys. G Carry 2 AccusyS Purpose-built...",
+  "start_time": 0.125,
+  "end_time": 1.627
+}
+```
+
+- `chunk_type = 'sentence'`
+- `content.text` = ""（空）
+- `content.ocr_text` = OCR 文字
+- `text_content` = OCR 文字
+- `metadata.language` = "ocr"
+
+### 分組邏輯
+
+```
+OCR frames: 4, 5, 6, 7, 8, 9, 10, 11, 13, 16, ..., 66
+  ↓ 按鄰近性分組（間距 ≤ 5 幀）
+Group 1: frames 4-16 → chunk "Accusys. G Carry 2..."
+Group 2: frames 48-66 → chunk "Western Digital..."
+```
 
 ## Rule 2 — 集合內容
 
@@ -80,11 +118,46 @@ Body: {"uuid": "...", "criteria": {"required_classes": ["person"]}}
 ## 流程
 
 ```
-ASR output (sentence) ─── Rule 1 ───→ chunk (sentence, text+embedding)
+ASRX output (sentence) ─── Rule 1 Phase 1 ──→ chunk (sentence, ASRX text)
                                                  │
-YOLO output (objects) ─── Rule 2 ───→ chunk (visual, objects+classes)
+OCR output (frames)    ─── Rule 1 Phase 2 ──→ chunk (sentence, OCR text)
                                                  │
                                                  ├── 文字搜尋 (ILIKE)
                                                  ├── 向量搜尋 (Qdrant)
                                                  └── 視覺過濾 (objects/classes)
 ```
+
+## 統計 API
+
+```
+GET /api/v1/stats/ingestion-status/{file_uuid}
+
+回應:
+  rule1_sentence: 35 sentence chunks
+  rule1_ocr: 30 OCR frames
+  rule1_ocr_chunks: 3 OCR-only chunks
+```
+
+| 步驟 | 說明 |
+|------|------|
+| `rule1_sentence` | 總 sentence chunks 數（ASRX + OCR-only） |
+| `rule1_ocr` | OCR pre_chunks 幀數 |
+| `rule1_ocr_chunks` | OCR-only chunks 數 |
+
+## 範例：FilmRiot_test
+
+| 項目 | 數量 | 說明 |
+|------|------|------|
+| ASRX segments | 32 | 語音段落 |
+| OCR frames | 30 | 偵測到文字的幀 |
+| Sentence chunks | 35 | 32 ASRX + 3 OCR-only |
+| OCR-only chunks | 3 | 片頭文字群組 |
+
+### Chunks 分佈
+
+| Chunk ID | 類型 | 時間 | 內容 |
+|----------|------|------|------|
+| 0-31 | ASRX | 0-81s | 語音文字 |
+| 32 | OCR-only | 0.12-1.62s | 片頭 "Accusys. G Carry 2..." |
+| 33 | OCR-only | 1.91-2.21s | "Western Digital..." |
+| 34 | OCR-only | 2.46-2.79s | "WD Cold Enterprise..." |

portal/.env.development

@@ -1,4 +1,4 @@
 # Portal Development Environment
 VITE_APP_TITLE=Momentry Portal (Development)
-VITE_API_BASE_URL=http://127.0.0.1:3003
+VITE_API_BASE_URL=http://127.0.0.1:3002
 VITE_API_KEY=muser_68600856036340bcafc01930eb4bd839_1774418104_97221b69

portal/src/api/client.ts

@@ -75,7 +75,7 @@ export interface UnregisterResponse {
 // ── Config (browser-only, stored in localStorage) ───────────────────────
 
 const DEFAULT_CONFIG: PortalConfig = {
-  api_base_url: import.meta.env.VITE_API_BASE_URL || 'http://127.0.0.1:3003',
+  api_base_url: import.meta.env.VITE_API_BASE_URL || 'http://127.0.0.1:3002',
   api_key: import.meta.env.VITE_API_KEY || '',
   timeout_secs: 30,
 }
@@ -99,13 +99,20 @@ export function saveConfig(config: PortalConfig): void {
 export async function logout(): Promise<void> {
   try {
     const config = getConfig();
+    const jwtToken = localStorage.getItem('momentry_jwt');
     const apiKey = config.api_key || localStorage.getItem('momentry_api_key');
-    if (apiKey) {
+    if (jwtToken || apiKey) {
+      const headers: Record<string, string> = { 'Content-Type': 'application/json' };
+      if (jwtToken) {
+        headers['Authorization'] = `Bearer ${jwtToken}`;
+      } else if (apiKey) {
+        headers['X-API-Key'] = apiKey;
+      }
       // Call logout API to invalidate session on server side (if implemented)
       // For now, just best effort
       await fetch(`${config.api_base_url}/api/v1/auth/logout`, {
         method: 'POST',
-        headers: { 'X-API-Key': apiKey }
+        headers
       }).catch(() => {}); // Ignore network errors
     }
   } catch (e) {
@@ -129,6 +136,7 @@ function handleSessionExpired() {
   localStorage.removeItem('momentry_user');
   localStorage.removeItem('portal_config');
   localStorage.removeItem('momentry_api_key');
+  localStorage.removeItem('momentry_jwt');
   if (window.location.pathname !== '/login') {
     window.location.href = '/login';
   }
@@ -139,12 +147,15 @@ export async function httpFetch<T>(url: string, options?: RequestInit, retries =
   // Re-read config to ensure we have the latest key if it changed
   const config = getConfig();
   
-  // Fallback key check
+  // Use JWT token if available, fallback to API key
+  const jwtToken = localStorage.getItem('momentry_jwt');
   const apiKey = config.api_key || localStorage.getItem('momentry_api_key') || '';
 
   const headers = new Headers(options?.headers);
   headers.set('Content-Type', 'application/json');
-  if (apiKey) {
+  if (jwtToken) {
+    headers.set('Authorization', `Bearer ${jwtToken}`);
+  } else if (apiKey) {
     headers.set('X-API-Key', apiKey);
   }
 
@@ -156,7 +167,7 @@ export async function httpFetch<T>(url: string, options?: RequestInit, retries =
       type: 'HTTP',
       method,
       url,
-      headers: { ...headers, 'X-API-Key': apiKey ? apiKey.substring(0, 10) + '...' : 'none' }, 
+      headers: { ...headers, 'Authorization': jwtToken ? 'Bearer ***' : 'none', 'X-API-Key': apiKey ? apiKey.substring(0, 10) + '...' : 'none' }, 
       body: options?.body ? JSON.parse(options.body as string) : null,
       status: 'loading',
       data: null,

portal/src/views/FilesView.vue

@@ -3,7 +3,31 @@
     <!-- Header with Search and Filters -->
     <div class="flex flex-col md:flex-row items-center justify-between gap-4">
       <h2 class="text-2xl font-bold">檔案管理 (Demo)</h2>
-      <div class="flex items-center gap-3 w-full md:w-auto">
+      <div class="flex flex-wrap items-center gap-3 w-full md:w-auto">
+        <!-- Media Type Filter -->
+        <div class="flex items-center bg-gray-700 rounded p-1">
+          <button 
+            @click="setMediaType('all')"
+            :class="{'bg-blue-600 text-white': mediaType === 'all', 'text-gray-300 hover:text-white': mediaType !== 'all'}"
+            class="px-3 py-1 rounded text-sm transition"
+          >
+            全部
+          </button>
+          <button 
+            @click="setMediaType('video')"
+            :class="{'bg-blue-600 text-white': mediaType === 'video', 'text-gray-300 hover:text-white': mediaType !== 'video'}"
+            class="px-3 py-1 rounded text-sm transition"
+          >
+            影片
+          </button>
+          <button 
+            @click="setMediaType('photo')"
+            :class="{'bg-blue-600 text-white': mediaType === 'photo', 'text-gray-300 hover:text-white': mediaType !== 'photo'}"
+            class="px-3 py-1 rounded text-sm transition"
+          >
+            照片
+          </button>
+        </div>
         <!-- Status Filter -->
         <div class="flex items-center bg-gray-700 rounded p-1">
           <button 
@@ -41,6 +65,20 @@
           >
             已完成
           </button>
+          <button 
+            @click="setStatusFilter('unindexed')"
+            :class="{'bg-blue-600 text-white': statusFilter === 'unindexed', 'text-gray-300 hover:text-white': statusFilter !== 'unindexed'}"
+            class="px-3 py-1 rounded text-sm transition"
+          >
+            未入庫
+          </button>
+          <button 
+            @click="setStatusFilter('indexed')"
+            :class="{'bg-blue-600 text-white': statusFilter === 'indexed', 'text-gray-300 hover:text-white': statusFilter !== 'indexed'}"
+            class="px-3 py-1 rounded text-sm transition"
+          >
+            已入庫
+          </button>
         </div>
         <!-- Search input -->
         <input
@@ -68,6 +106,7 @@
         <thead class="bg-gray-900">
           <tr>
             <th class="px-6 py-3 text-left text-xs font-medium text-gray-300 uppercase tracking-wider">檔案名稱</th>
+            <th class="px-6 py-3 text-left text-xs font-medium text-gray-300 uppercase tracking-wider">類型</th>
             <th class="px-6 py-3 text-left text-xs font-medium text-gray-300 uppercase tracking-wider">狀態</th>
             <th class="px-6 py-3 text-left text-xs font-medium text-gray-300 uppercase tracking-wider">UUID</th>
             <th class="px-6 py-3 text-left text-xs font-medium text-gray-300 uppercase tracking-wider">操作</th>
@@ -81,7 +120,21 @@
                </div>
              </td>
              <td class="px-6 py-4 whitespace-nowrap text-sm">
-               <span v-if="file.status === 'completed'" class="px-2 py-0.5 rounded text-xs bg-green-900 text-green-200">
+               <span v-if="file.media_type === 'video'" class="px-2 py-0.5 rounded text-xs bg-blue-900 text-blue-200">
+                 🎬 影片
+               </span>
+               <span v-else class="px-2 py-0.5 rounded text-xs bg-green-900 text-green-200">
+                 📷 照片
+               </span>
+             </td>
+             <td class="px-6 py-4 whitespace-nowrap text-sm">
+               <span v-if="file.status === 'completed' && file.is_indexed" class="px-2 py-0.5 rounded text-xs bg-green-900 text-green-200">
+                 ✅ 已入庫
+               </span>
+               <span v-else-if="file.status === 'completed' && !file.is_indexed" class="px-2 py-0.5 rounded text-xs bg-yellow-900 text-yellow-200">
+                 ⏳ 未入庫
+               </span>
+               <span v-else-if="file.status === 'completed'" class="px-2 py-0.5 rounded text-xs bg-green-900 text-green-200">
                  ✅ 已完成
                </span>
                <span v-else-if="file.status === 'processing'" class="px-2 py-0.5 rounded text-xs bg-yellow-900 text-yellow-200">
@@ -148,15 +201,32 @@ import { ref, computed, onMounted } from 'vue'
 import { useRouter } from 'vue-router'
 import { registerVideo, unregisterVideo, httpFetch, getCurrentConfig } from '@/api/client'
 
+const VIDEO_EXTENSIONS = ['mp4', 'mov', 'mkv', 'avi', 'webm', 'wmv', 'flv', 'm4v']
+const PHOTO_EXTENSIONS = ['jpg', 'jpeg', 'png', 'gif', 'webp', 'bmp', 'tiff', 'tif']
+
 const router = useRouter()
 const files = ref<any[]>([])
 const loading = ref(false)
 const error = ref<string | null>(null)
 const searchQuery = ref('')
-const statusFilter = ref('all') // all, unregistered, pending, processing, completed
+const statusFilter = ref('all')
+const mediaType = ref('all')
+
+function getMediaType(fileName: string): 'video' | 'photo' {
+  const ext = fileName.split('.').pop()?.toLowerCase() || ''
+  if (VIDEO_EXTENSIONS.includes(ext)) return 'video'
+  if (PHOTO_EXTENSIONS.includes(ext)) return 'photo'
+  return 'video'
+}
 
 const displayFiles = computed(() => {
-  let result = files.value
+  // Start with a copy to avoid mutation issues
+  let result = [...files.value]
+  
+  // Filter by media type
+  if (mediaType.value !== 'all') {
+    result = result.filter(f => f.media_type === mediaType.value)
+  }
 
   // Filter by search
   if (searchQuery.value) {
@@ -169,8 +239,14 @@ const displayFiles = computed(() => {
 
   // Filter by status
   if (statusFilter.value !== 'all') {
+    if (statusFilter.value === 'indexed') {
+      result = result.filter(f => f.status === 'completed' && f.is_indexed)
+    } else if (statusFilter.value === 'unindexed') {
+      result = result.filter(f => f.status === 'completed' && !f.is_indexed)
+    } else {
       result = result.filter(f => f.status === statusFilter.value)
     }
+  }
 
   return result
 })
@@ -179,35 +255,73 @@ function setStatusFilter(status: string) {
   statusFilter.value = status
 }
 
+function setMediaType(type: string) {
+  mediaType.value = type
+}
+
 async function fetchFiles() {
   loading.value = true
   try {
     const config = getCurrentConfig()
+    
+    // Get scan results FIRST (source of truth for files on disk)
     const scanResp = await httpFetch<any>(`${config.api_base_url}/api/v1/files/scan`)
-    const scanFiles: any[] = (scanResp?.files || []).map((f: any) => ({
+    const scanFiles: any[] = (scanResp?.files || []).map((f: any) => {
+      const mediaType = getMediaType(f.file_name)
+      return {
         ...f,
-      status: f.is_registered ? 'registered_scan' : 'unregistered'
+        media_type: mediaType,
-    }))
+        is_indexed: false,
+        // Use scan API's is_registered field as default status
+        status: f.is_registered ? 'pending' : 'unregistered'
+      }
+    })
 
     // Get registered files with real processing status
     let regFiles: any[] = []
     try {
-      const regResp = await httpFetch<any>(`${config.api_base_url}/api/v1/files?page=1&page_size=100`)
+      const regResp = await httpFetch<any>(`${config.api_base_url}/api/v1/files?page=1&page_size=200`)
-      regFiles = (regResp?.files || regResp?.data || []).map((f: any) => ({
+      regFiles = (regResp?.files || regResp?.data || []).map((f: any) => {
+        const mediaType = getMediaType(f.file_name)
+        return {
           ...f,
-        status: f.status || 'pending'
+          media_type: mediaType,
-      }))
+          status: f.status || 'pending',
+          is_indexed: (f.total_chunks || 0) > 0
+        }
+      })
     } catch {
-      // Registered files API may not be available; use scan data only
+      // Registered files API may not be available; scan data will be used as fallback
+      console.warn('Failed to fetch registered files, using scan data as fallback')
     }
 
-    // Merge: scan results first, then overlay with registered statuses
+    // Build a map of registered files by file_path for quick lookup
-    const merged = new Map<string, any>()
+    const regMap = new Map<string, any>()
-    for (const f of scanFiles) {
-      merged.set(f.file_path, f)
-    }
     for (const f of regFiles) {
-      merged.set(f.file_path, f)
+      regMap.set(f.file_path, f)
+    }
+
+    // Merge: start with scan results, override with registered data where available
+    const merged = new Map<string, any>()
+    
+    // First pass: add all scan results
+    for (const f of scanFiles) {
+      merged.set(f.file_path, { ...f })
+    }
+    
+    // Second pass: override with real registered data (only for files that exist in scan)
+    for (const f of regFiles) {
+      if (merged.has(f.file_path)) {
+        // File exists on disk, update with real status
+        const existing = merged.get(f.file_path)
+        merged.set(f.file_path, {
+          ...existing,
+          ...f,
+          media_type: f.media_type || existing.media_type,
+          is_indexed: f.is_indexed || existing.is_indexed
+        })
+      }
+      // If file not in scan, it might have been deleted from disk - skip it
     }
 
     files.value = Array.from(merged.values())

portal/src/views/LoginView.vue

@@ -116,6 +116,9 @@ const handleLogin = async () => {
     if (data.success) {
       localStorage.setItem('momentry_user', JSON.stringify(data.user))
       localStorage.setItem('momentry_api_key', data.api_key)
+      if (data.jwt) {
+        localStorage.setItem('momentry_jwt', data.jwt)
+      }
       saveConfig({ ...config, api_key: data.api_key })
       const redirect = (route.query.redirect as string) || '/home'
       router.push(redirect)

run-server-3002.sh

@@ -1,53 +1,30 @@
 #!/usr/bin/env bash
-# Start production server on port 3002
+# Start production server via launchd
-# Logs to logs/momentry_3002.log
-
 set -euo pipefail
 
-SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+case "${1:-}" in
-cd "$SCRIPT_DIR"
+  stop)
-
+    echo "Stopping production server and worker..."
-# Production environment variables
+    launchctl bootout gui/$(id -u)/com.momentry.server 2>/dev/null || true
-export MOMENTRY_OUTPUT_DIR=/Users/accusys/momentry/output
+    launchctl bootout gui/$(id -u)/com.momentry.worker 2>/dev/null || true
-export DATABASE_SCHEMA=public
+    echo "Stopped."
-export MOMENTRY_REDIS_PREFIX=momentry:
+    ;;
-export MOMENTRY_SERVER_PORT=3002
+  restart)
-
+    exec "$0" stop
-# Kill existing server on port 3002
+    exec "$0" start
-PID=$(lsof -ti :3002 2>/dev/null || true)
+    ;;
-if [ -n "$PID" ]; then
+  status)
-  echo "Killing existing server on port 3002 (PID: $PID)"
+    echo "=== Server ==="
-  kill "$PID" 2>/dev/null || true
+    launchctl list com.momentry.server 2>/dev/null | head -5 || echo "Not loaded"
-  sleep 2
+    echo "=== Worker ==="
-fi
+    launchctl list com.momentry.worker 2>/dev/null | head -5 || echo "Not loaded"
-
+    echo "=== Health ==="
-# Kill existing worker via PID file
+    curl -s --max-time 3 http://localhost:3002/api/v1/health 2>/dev/null || echo "No response"
-if [ -f logs/worker_3002.pid ]; then
+    ;;
-  WPID=$(cat logs/worker_3002.pid)
+  *)
-  if kill -0 "$WPID" 2>/dev/null; then
+    echo "Starting production server and worker via launchd..."
-    echo "Killing existing worker (PID: $WPID)"
+    launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/com.momentry.server.plist 2>/dev/null || launchctl load ~/Library/LaunchAgents/com.momentry.server.plist
-    kill "$WPID" 2>/dev/null || true
+    launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/com.momentry.worker.plist 2>/dev/null || launchctl load ~/Library/LaunchAgents/com.momentry.worker.plist
-    sleep 1
+    echo "Done. Use '$0 status' to check."
-  fi
+    ;;
-  rm -f logs/worker_3002.pid
+esac
-fi
-
-# Build if needed
-if [ ! -f target/release/momentry ]; then
-  echo "Building release binary..."
-  cargo build --release --bin momentry
-fi
-
-# Start server
-echo "Starting momentry server on port 3002..."
-./target/release/momentry server --host 0.0.0.0 --port 3002 > logs/momentry_3002.log 2>&1 &
-echo "Server started (PID: $!)"
-echo "Logs: logs/momentry_3002.log"
-
-# Start companion worker
-echo "Starting momentry worker..."
-nohup ./target/release/momentry worker --max-concurrent 6 --poll-interval 10 --batch-size 5 > logs/worker_3002.log 2>&1 &
-WPID=$!
-echo "$WPID" > logs/worker_3002.pid
-echo "Worker started (PID: $WPID)"
-echo "Worker logs: logs/worker_3002.log"

run-worker-3002.sh

@@ -13,6 +13,9 @@ mkdir -p logs
 export MOMENTRY_OUTPUT_DIR=/Users/accusys/momentry/output
 export DATABASE_SCHEMA=public
 export MOMENTRY_REDIS_PREFIX=momentry:
+# Qdrant credentials for Python subprocesses
+export QDRANT_URL=http://127.0.0.1:6333
+export QDRANT_API_KEY=Test3200Test3200Test3200
 
 # Kill existing worker via PID file
 if [ -f logs/worker_3002.pid ]; then

scripts/asrx_processor.py

@@ -110,6 +110,39 @@ def _shared_audio_setup(video_path):
     return tmp_dir, video_path
 
 
+def _convert_asr_segments_to_asrx(asr_segments, output_path):
+    """Convert ASR segments to ASRX format with frame information"""
+    fps = 30.0
+    base_name = os.path.basename(output_path)
+    uuid_part = base_name.split(".")[0]
+    probe_path = os.path.join(os.path.dirname(output_path),
+                              f"{uuid_part}.probe.json")
+    if os.path.exists(probe_path):
+        try:
+            with open(probe_path) as pf:
+                probe_data = json.load(pf)
+                if "fps" in probe_data:
+                    fps = float(probe_data["fps"])
+        except Exception:
+            pass
+
+    segments = []
+    for s in asr_segments:
+        start_time = s.get("start", s.get("start_time", 0))
+        end_time = s.get("end", s.get("end_time", 0))
+        start_frame = int(start_time * fps)
+        end_frame = int(end_time * fps)
+        segments.append({
+            "start_time": start_time,
+            "end_time": end_time,
+            "start_frame": start_frame,
+            "end_frame": end_frame,
+            "text": s.get("text", ""),
+            "speaker_id": "SPEAKER_0",
+        })
+    return segments
+
+
 def _convert_result(result, output_path):
     """Stage 3: 將 SelfASRXFixed result 轉為 Rust-expected format"""
     fps = 30.0
@@ -127,6 +160,41 @@ def _convert_result(result, output_path):
             pass
 
     segment_count = len(result.get("segments", []))
+    
+    # Fallback: if ASRX has 0 segments but ASR has segments, use ASR segments
+    if segment_count == 0:
+        asr_path = output_path.replace(".asrx.json", ".asr.json")
+        if os.path.exists(asr_path):
+            try:
+                with open(asr_path) as f:
+                    asr_data = json.load(f)
+                asr_segments = asr_data.get("segments", [])
+                if asr_segments:
+                    print(f"[ASRX] ASRX has 0 segments, falling back to {len(asr_segments)} ASR segments", file=sys.stderr)
+                    # Convert ASR segments to ASRX format
+                    converted_segments = []
+                    for seg in asr_segments:
+                        start_time = seg.get("start_time", seg.get("start", 0))
+                        end_time = seg.get("end_time", seg.get("end", 0))
+                        start_frame = int(start_time * fps)
+                        end_frame = int(end_time * fps)
+                        converted_segments.append({
+                            "start_time": start_time,
+                            "end_time": end_time,
+                            "start_frame": start_frame,
+                            "end_frame": end_frame,
+                            "text": seg.get("text", ""),
+                            "speaker_id": "SPEAKER_0",
+                            "language": "",
+                            "lang_prob": 0.0,
+                            "quality": 0.0,
+                        })
+                    result["segments"] = converted_segments
+                    segment_count = len(converted_segments)
+                    result["status"] = "has_transcript"
+            except Exception as e:
+                print(f"[ASRX] Failed to load ASR fallback: {e}", file=sys.stderr)
+
     if segment_count > 0:
         status = "has_transcript"
     else:
@@ -228,10 +296,22 @@ def process_asrx(video_path: str, output_path: str, uuid: str = "",
             if "error" in result:
                 if publisher:
                     publisher.error("asrx", result["error"])
+                # Fallback to ASR segments if ASRX fails but ASR has content
+                if asr_segments:
+                    print(f"[ASRX] Resume error, falling back to {len(asr_segments)} ASR segments", file=sys.stderr)
+                    segments = _convert_asr_segments_to_asrx(asr_segments, output_path)
+                    output_result = {
+                        "status": "has_transcript",
+                        "language": None,
+                        "segments": segments,
+                        "segment_count": len(segments),
+                        "fallback_from_asr": True,
+                    }
+                else:
                     output_result = {"status": "silent_audio", "language": None, "segments": [], "segment_count": 0}
                 _atomic_write(output_path, output_result)
                 if publisher:
-                    publisher.complete("asrx", "0 segments")
+                    publisher.complete("asrx", f"{output_result['segment_count']} segments")
                 _cleanup(tmp_dir)
                 return output_result
 
@@ -264,10 +344,21 @@ def process_asrx(video_path: str, output_path: str, uuid: str = "",
                 publisher.error("asrx", str(e))
             import traceback
             traceback.print_exc()
+            # Fallback to ASR segments if ASRX fails but ASR has content
+            if asr_segments:
+                print(f"[ASRX] Resume exception, falling back to {len(asr_segments)} ASR segments", file=sys.stderr)
+                output_result = {
+                    "status": "has_transcript",
+                    "language": None,
+                    "segments": [{"start_time": s["start"], "end_time": s["end"], "text": s["text"], "speaker_id": "SPEAKER_0"} for s in asr_segments],
+                    "segment_count": len(asr_segments),
+                    "fallback_from_asr": True,
+                }
+            else:
                 output_result = {"status": "silent_audio", "language": None, "segments": [], "segment_count": 0}
             _atomic_write(output_path, output_result)
             if publisher:
-                publisher.complete("asrx", "0 segments")
+                publisher.complete("asrx", f"{output_result['segment_count']} segments")
             _cleanup(tmp_dir)
             return output_result
 
@@ -328,10 +419,21 @@ def process_asrx(video_path: str, output_path: str, uuid: str = "",
         if "error" in result:
             if publisher:
                 publisher.error("asrx", result["error"])
+            # Fallback to ASR segments if ASRX fails
+            if asr_segments:
+                print(f"[ASRX] ASRX failed, falling back to {len(asr_segments)} ASR segments", file=sys.stderr)
+                output_result = {
+                    "status": "has_transcript",
+                    "language": None,
+                    "segments": [{"start_time": s["start"], "end_time": s["end"], "text": s["text"], "speaker_id": "SPEAKER_0"} for s in asr_segments],
+                    "segment_count": len(asr_segments),
+                    "fallback_from_asr": True,
+                }
+            else:
                 output_result = {"status": "silent_audio", "language": None, "segments": [], "segment_count": 0}
             _atomic_write(output_path, output_result)
             if publisher:
-                publisher.complete("asrx", "0 segments")
+                publisher.complete("asrx", f"{output_result['segment_count']} segments")
             _cleanup(tmp_dir)
             return output_result
 
@@ -359,10 +461,21 @@ def process_asrx(video_path: str, output_path: str, uuid: str = "",
         import traceback
         traceback.print_exc()
 
+        # Fallback to ASR segments if ASRX fails but ASR has content
+        if asr_segments:
+            print(f"[ASRX] Exception occurred, falling back to {len(asr_segments)} ASR segments", file=sys.stderr)
+            output_result = {
+                "status": "has_transcript",
+                "language": None,
+                "segments": [{"start_time": s["start"], "end_time": s["end"], "text": s["text"], "speaker_id": "SPEAKER_0"} for s in asr_segments],
+                "segment_count": len(asr_segments),
+                "fallback_from_asr": True,
+            }
+        else:
             output_result = {"status": "silent_audio", "language": None, "segments": [], "segment_count": 0}
         _atomic_write(output_path, output_result)
         if publisher:
-            publisher.complete("asrx", "0 segments")
+            publisher.complete("asrx", f"{output_result['segment_count']} segments")
         # 如果 checkpoint 已存在（Step 3 完成後 crash），保留 WAV 給 resume
         if not os.path.exists(checkpoint_path):
             _cleanup(tmp_dir)

scripts/clothing_color_search.py

@@ -0,0 +1,216 @@
+#!/opt/homebrew/bin/python3.11
+"""
+Clothing Color Search - Find people wearing specific colors
+
+Usage:
+  python3 clothing_color_search.py --file-uuid UUID --color red --output output.json
+
+Color matching uses HSV hue ranges:
+  red: 0-15, 165-180
+  orange: 15-35
+  yellow: 35-50
+  green: 50-85
+  cyan: 85-105
+  blue: 105-140
+  purple: 140-165
+"""
+
+import sys
+import os
+import json
+import argparse
+import cv2
+import numpy as np
+
+COLOR_RANGES = {
+    "red": [(0, 40), (165, 180)],
+    "orange": [(15, 35)],
+    "yellow": [(35, 50)],
+    "green": [(50, 85)],
+    "cyan": [(85, 105)],
+    "blue": [(105, 140)],
+    "purple": [(140, 165)],
+    "white": [(0, 180, 0, 40, 200, 255)],  # (h_min, h_max, s_min, s_max, v_min, v_max)
+    "black": [(0, 180, 0, 255, 0, 50)],
+}
+
+def hsv_to_color_name(h, s, v):
+    """Convert HSV to color name"""
+    if v < 50:
+        return "black"
+    if s < 40 and v > 200:
+        return "white"
+    if 0 <= h <= 15 or 165 <= h <= 180:
+        return "red"
+    if 15 < h <= 35:
+        return "orange"
+    if 35 < h <= 50:
+        return "yellow"
+    if 50 < h <= 85:
+        return "green"
+    if 85 < h <= 105:
+        return "cyan"
+    if 105 < h <= 140:
+        return "blue"
+    if 140 < h <= 165:
+        return "purple"
+    return "unknown"
+
+def check_color_match(dominant_colors, target_color):
+    """Check if dominant colors match target color"""
+    if not dominant_colors:
+        return False, 0.0
+    
+    target_lower = target_color.lower()
+    match_count = 0
+    total = len(dominant_colors)
+    
+    for color_hsv in dominant_colors:
+        h, s, v = color_hsv[0], color_hsv[1], color_hsv[2]
+        color_name = hsv_to_color_name(h, s, v)
+        if color_name == target_lower:
+            match_count += 1
+    
+    ratio = match_count / total if total > 0 else 0.0
+    return ratio > 0.3, ratio  # Match if >30% of dominant colors match
+
+def search_by_color(appearance_path, video_path, target_color, output_path, max_frames=500):
+    """Search for people wearing target color"""
+    if not os.path.exists(appearance_path):
+        print(json.dumps({"error": f"appearance.json not found: {appearance_path}"}))
+        return
+    
+    with open(appearance_path) as f:
+        appearance = json.load(f)
+    
+    frames = appearance.get("frames", [])
+    fps = appearance.get("fps", 30)
+    
+    cap = cv2.VideoCapture(video_path)
+    if not cap.isOpened():
+        print(json.dumps({"error": f"Cannot open video: {video_path}"}))
+        return
+    
+    results = []
+    frame_count = 0
+    
+    for frame_data in frames[:max_frames]:
+        frame_num = frame_data.get("frame", 0)
+        persons = frame_data.get("persons", [])
+        timestamp = frame_data.get("timestamp", 0)
+        
+        if not persons:
+            frame_count += 1
+            continue
+        
+        cap.set(cv2.CAP_PROP_POS_FRAMES, frame_num)
+        ret, frame = cap.read()
+        if not ret:
+            frame_count += 1
+            continue
+        
+        frame_h, frame_w = frame.shape[:2]
+        
+        for person in persons:
+            bbox = person.get("bbox", {})
+            if not bbox:
+                continue
+            
+            x, y = bbox.get("x", 0), bbox.get("y", 0)
+            w, h = bbox.get("width", 0), bbox.get("height", 0)
+            
+            # Extract upper body region (clothing area)
+            upper_h = int(h * 0.6)  # Upper 60% of person
+            roi_x = max(0, int(x))
+            roi_y = max(0, int(y))
+            roi_w = min(w, frame_w - roi_x)
+            roi_h = min(upper_h, frame_h - roi_y)
+            
+            if roi_w < 10 or roi_h < 10:
+                continue
+            
+            roi = frame[roi_y:roi_y+roi_h, roi_x:roi_x+roi_w]
+            
+            # Get dominant colors
+            hsv = cv2.cvtColor(roi, cv2.COLOR_BGR2HSV)
+            pixels = hsv.reshape(-1, 3).astype(np.float32)
+            
+            if len(pixels) < 10:
+                continue
+            
+            criteria = (cv2.TERM_CRITERIA_EPS + cv2.TERM_CRITERIA_MAX_ITER, 10, 1.0)
+            _, labels, centers = cv2.kmeans(pixels, 5, None, criteria, 10, cv2.KMEANS_RANDOM_CENTERS)
+            counts = np.bincount(labels.flatten())
+            dominant = centers[np.argsort(-counts)[:5]].tolist()
+            
+            match, confidence = check_color_match(dominant, target_color)
+            
+            if match:
+                results.append({
+                    "frame": frame_num,
+                    "timestamp": round(timestamp, 2),
+                    "bbox": bbox,
+                    "confidence": round(confidence, 3),
+                    "dominant_colors": [[round(c, 1) for c in dc] for dc in dominant[:3]]
+                })
+        
+        frame_count += 1
+    
+    cap.release()
+    
+    # Summary
+    color_names = set()
+    for r in results:
+        for dc in r.get("dominant_colors", []):
+            if len(dc) >= 3:
+                color_names.add(hsv_to_color_name(dc[0], dc[1], dc[2]))
+    
+    output = {
+        "file_uuid": os.path.basename(appearance_path).split(".")[0],
+        "target_color": target_color,
+        "total_matches": len(results),
+        "matched_frames": list(set(r["frame"] for r in results)),
+        "results": results[:50],  # Limit to 50 results
+        "color_names_found": list(color_names)
+    }
+    
+    os.makedirs(os.path.dirname(output_path) if os.path.dirname(output_path) else ".", exist_ok=True)
+    with open(output_path, "w") as f:
+        json.dump(output, f, indent=2)
+    
+    print(json.dumps({"success": True, **output}))
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Search for people by clothing color")
+    parser.add_argument("--file-uuid", required=True)
+    parser.add_argument("--color", required=True, choices=list(COLOR_RANGES.keys()))
+    parser.add_argument("--video-path", default="")
+    parser.add_argument("--appearance-path", default="")
+    parser.add_argument("--output", default="")
+    args = parser.parse_args()
+    
+    output_dir = os.environ.get("MOMENTRY_OUTPUT_DIR", "/Users/accusys/momentry/output")
+    appearance_path = args.appearance_path or f"{output_dir}/{args.file_uuid}.appearance.json"
+    video_path = args.video_path
+    output_path = args.output or f"{output_dir}/{args.file_uuid}.color_search_{args.color}.json"
+    
+    if not video_path:
+        # Try to find video in common locations
+        for ext in ["mp4", "mov", "avi"]:
+            candidate = f"/Users/accusys/momentry/var/sftpgo/data/demo/{args.file_uuid}.{ext}"
+            if os.path.exists(candidate):
+                video_path = candidate
+                break
+        
+        if not video_path:
+            # Search in output directory for video
+            import glob
+            matches = glob.glob(f"/Users/accusys/momentry/var/sftpgo/**/*{args.file_uuid}*", recursive=True)
+            if matches:
+                video_path = matches[0]
+    
+    if not video_path:
+        print(json.dumps({"error": "video_path not found, please provide --video-path"}))
+        sys.exit(1)
+    
+    search_by_color(appearance_path, video_path, args.color, output_path)

scripts/face_clustering_processor.py

@@ -104,9 +104,14 @@ def main():
     print(f"[FACE_CLUSTER] Loading embeddings from Qdrant for {UUID}...")
     try:
         import requests
-        qdrant_url = "http://localhost:6333"
+        qdrant_url = os.environ.get("QDRANT_URL", "http://localhost:6333")
+        qdrant_api_key = os.environ.get("QDRANT_API_KEY", "")
         collection = "_faces"
         
+        headers = {}
+        if qdrant_api_key:
+            headers["api-key"] = qdrant_api_key
+        
         # Query all embeddings for this file_uuid
         response = requests.post(
             f"{qdrant_url}/collections/{collection}/points/scroll",
@@ -118,7 +123,8 @@ def main():
                 },
                 "limit": 10000,
                 "with_vector": True
-            }
+            },
+            headers=headers
         )
         
         if response.status_code == 200:
@@ -140,22 +146,57 @@ def main():
         print(f"[FACE_CLUSTER] Failed to load embeddings from Qdrant: {e}")
         embedding_map = {}
 
-    # Use embeddings from Qdrant or face.json
+    # Use embeddings from Qdrant - match by frame + bbox
     embeddings = []
     face_refs = []
 
     print(f"🔍 Collecting face embeddings for {UUID}...")
 
+    # Build a lookup: (frame, bbox_center) -> embedding
+    # Use frame number and approximate bbox center for matching
+    qdrant_by_frame = {}
+    for point in points:
+        payload = point.get("payload", {})
+        frame = payload.get("frame")
+        bbox = payload.get("bbox", {})
+        vector = point.get("vector")
+        if frame is not None and vector:
+            # Use frame + bbox center as key
+            cx = bbox.get("x", 0) + bbox.get("width", 0) // 2
+            cy = bbox.get("y", 0) + bbox.get("height", 0) // 2
+            key = (frame, cx, cy)
+            if key not in qdrant_by_frame:
+                qdrant_by_frame[key] = vector
+
+    print(f"[FACE_CLUSTER] Built Qdrant lookup with {len(qdrant_by_frame)} entries")
+
     for frame_idx, frame_obj in enumerate(frames_list):
+        frame_num = frame_obj.get("frame", frame_idx)
         faces = frame_obj.get("faces", [])
         if not faces:
             continue
 
         for face_idx, face in enumerate(faces):
-            face_id = face.get("face_id")
+            x = face.get("x", 0)
-            if face_id and face_id in embedding_map:
+            y = face.get("y", 0)
-                embeddings.append(embedding_map[face_id])
+            w = face.get("width", 0)
-                face_refs.append({"frame_idx": frame_idx, "face_idx": face_idx, "face_id": face_id})
+            h = face.get("height", 0)
+            cx = x + w // 2
+            cy = y + h // 2
+
+            # Try exact match first
+            key = (frame_num, cx, cy)
+            if key in qdrant_by_frame:
+                embeddings.append(qdrant_by_frame[key])
+                face_refs.append({"frame_idx": frame_idx, "face_idx": face_idx})
+                continue
+
+            # Try approximate match (within 50 pixels)
+            for (qf, qx, qy), vec in qdrant_by_frame.items():
+                if qf == frame_num and abs(qx - cx) < 50 and abs(qy - cy) < 50:
+                    embeddings.append(vec)
+                    face_refs.append({"frame_idx": frame_idx, "face_idx": face_idx})
+                    break
 
     if not embeddings:
         print("❌ No embeddings found in Qdrant.")

scripts/generate_seed_embeddings.py

@@ -46,11 +46,12 @@ SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
 FACENET_PATH = os.path.join(SCRIPT_DIR, "..", "models", "facenet512.mlpackage")
 
 
-def get_tmdb_identities(limit: int = None) -> List[Dict]:
+def get_tmdb_identities(limit: int = None, file_uuid: str = None) -> List[Dict]:
     """Query PG for TMDb identities with profile photos
     
     Args:
         limit: Max identities to process
+        file_uuid: Filter by file_uuid via file_identities table
     
     Returns:
         List of {id, uuid, name, tmdb_id, tmdb_profile}
@@ -63,20 +64,34 @@ def get_tmdb_identities(limit: int = None) -> List[Dict]:
     
     if SCHEMA == "public":
         table = "identities"
+        file_table = "file_identities"
     else:
         table = f"{SCHEMA}.identities"
+        file_table = f"{SCHEMA}.file_identities"
     
+    if file_uuid:
+        query = f"""
+            SELECT DISTINCT i.id, i.uuid, i.name, i.tmdb_id, i.tmdb_profile
+            FROM {table} i
+            JOIN {file_table} fi ON fi.identity_id = i.id
+            WHERE i.source = 'tmdb' AND i.tmdb_profile IS NOT NULL
+              AND fi.file_uuid = %s
+            ORDER BY i.id
+        """
+        if limit:
+            query += f" LIMIT {limit}"
+        cur.execute(query, (file_uuid,))
+    else:
         query = f"""
             SELECT id, uuid, name, tmdb_id, tmdb_profile
             FROM {table}
             WHERE source = 'tmdb' AND tmdb_profile IS NOT NULL
             ORDER BY id
         """
-    
         if limit:
             query += f" LIMIT {limit}"
-    
         cur.execute(query)
+    
     rows = cur.fetchall()
     cur.close()
     conn.close()
@@ -180,7 +195,7 @@ def extract_face_embedding(image_path: str) -> Optional[List[float]]:
         return None
 
 
-def generate_seed_embeddings(limit: int = None, dry_run: bool = False) -> Dict:
+def generate_seed_embeddings(limit: int = None, dry_run: bool = False, file_uuid: str = None) -> Dict:
     """Generate embeddings for all TMDb identities
     
     Args:
@@ -198,14 +213,14 @@ def generate_seed_embeddings(limit: int = None, dry_run: bool = False) -> Dict:
         "errors": [],
     }
     
-    identities = get_tmdb_identities(limit)
+    identities = get_tmdb_identities(limit, file_uuid)
     result["total"] = len(identities)
     
     if not identities:
-        print("[SEED] No TMDb identities with profile photos")
+        print(f"[SEED] No TMDb identities with profile photos{' for ' + file_uuid if file_uuid else ''}")
         return result
     
-    print(f"[SEED] Found {len(identities)} TMDb identities")
+    print(f"[SEED] Found {len(identities)} TMDb identities{' for ' + file_uuid if file_uuid else ''}")
     
     if not dry_run:
         ensure_seeds_collection()
@@ -259,6 +274,7 @@ def generate_seed_embeddings(limit: int = None, dry_run: bool = False) -> Dict:
                     name=name,
                     embedding=embedding,
                     source="tmdb",
+                    file_uuid=file_uuid,
                     tmdb_id=tmdb_id,
                 )
                 result["success"] += 1
@@ -280,12 +296,13 @@ def main():
     parser.add_argument("--dry-run", action="store_true", help="Don't push to Qdrant")
     parser.add_argument("--tmdb-api-key", help="TMDb API key (optional, for rate limiting)")
     parser.add_argument("--output", help="Output JSON file path")
+    parser.add_argument("--file-uuid", help="File UUID to generate seeds for")
     args = parser.parse_args()
     
     if args.tmdb_api_key:
         TMDB_API_KEY = args.tmdb_api_key
     
-    result = generate_seed_embeddings(args.limit, args.dry_run)
+    result = generate_seed_embeddings(args.limit, args.dry_run, args.file_uuid)
     
     output_json = json.dumps(result, indent=2, ensure_ascii=False)
     

scripts/identity_matcher.py

@@ -79,10 +79,10 @@ def match_faces_round_1(file_uuid: str) -> dict:
         {trace_id: {identity_id, identity_uuid, name, score, suggested_by: 'tmdb'}}
     """
     traces = get_trace_representatives(file_uuid)
-    seeds = get_seeds(source="tmdb")
+    seeds = get_seeds(source="tmdb", file_uuid=file_uuid)
     
     if not seeds:
-        print("[MATCH] No TMDb seeds available")
+        print(f"[MATCH] No TMDb seeds available for {file_uuid}")
         return {}
     
     suggestions = {}

scripts/utils/qdrant_faces.py

@@ -493,11 +493,12 @@ def push_seed_embedding(
         raise RuntimeError(f"Qdrant seed push failed: HTTP {e.code} - {error_body}")
 
 
-def get_seeds(source: str = None) -> list:
+def get_seeds(source: str = None, file_uuid: str = None) -> list:
     """Get all seed points
     
     Args:
         source: Filter by source ('tmdb', 'manual', 'propagation'), or None for all
+        file_uuid: Filter by file_uuid, or None for all
     
     Returns:
         List of seed points with payload and vector
@@ -514,12 +515,14 @@ def get_seeds(source: str = None) -> list:
             "with_vector": True,
         }
         
+        filters = []
         if source:
-            body["filter"] = {
+            filters.append({"key": "source", "match": {"value": source}})
-                "must": [
+        if file_uuid:
-                    {"key": "source", "match": {"value": source}}
+            filters.append({"key": "file_uuid", "match": {"value": file_uuid}})
-                ]
+        
-            }
+        if filters:
+            body["filter"] = {"must": filters}
         
         if offset:
             body["offset"] = offset

src/api/agent_search.rs

@@ -276,6 +276,15 @@ fn make_tools(pool: &sqlx::PgPool) -> Vec<ToolDef> {
             }),
             vec!["file_uuid", "node_id"],
         ),
+        function_calling::make_tool(
+            "search_by_appearance",
+            "根據衣服顏色搜尋影片中的人物。支援顏色：red, orange, yellow, green, cyan, blue, purple, white, black。",
+            serde_json::json!({
+                "file_uuid": {"type": "string", "description": "影片 UUID"},
+                "color": {"type": "string", "description": "目標顏色: red, orange, yellow, green, cyan, blue, purple, white, black"}
+            }),
+            vec!["file_uuid", "color"],
+        ),
     ]
 }
 
@@ -310,6 +319,7 @@ async fn execute_tool(pool: &sqlx::PgPool, tool_call: &ToolCall) -> (String, Str
         "get_file_info" => tools::exec_get_file_info(pool, &args).await,
         "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,
         _ => Err(format!("Unknown tool: {}", name)),
     };
     let content = match result {

src/api/files.rs

@@ -1239,15 +1239,26 @@ async fn unregister(
 
     let deleted_redis_keys = {
         match RedisClient::new() {
-            Ok(redis) => match redis.delete_worker_job(&uuid).await {
+            Ok(redis) => {
-                Ok(_) => {
+                let mut deleted = 0;
-                    tracing::info!("[UNREGISTER] Deleted Redis keys for {}", uuid);
+                
-                    Some(1)
+                // Delete worker job keys
+                if redis.delete_worker_job(&uuid).await.is_ok() {
+                    tracing::info!("[UNREGISTER] Deleted Redis worker job keys for {}", uuid);
+                    deleted += 1;
                 }
-                Err(e) => {
+                
-                    tracing::warn!("[UNREGISTER] Failed to delete Redis keys: {}", e);
+                // Delete PipelineProgress key
-                    None
+                let progress_key = format!("{}progress:{}:pipeline", 
+                    crate::core::config::REDIS_KEY_PREFIX.as_str(), uuid);
+                if let Ok(mut conn) = redis.get_conn().await {
+                    let _: Option<String> = redis::cmd("DEL").arg(&progress_key)
+                        .query_async(&mut conn).await.ok();
+                    tracing::info!("[UNREGISTER] Deleted Redis PipelineProgress for {}", uuid);
+                    deleted += 1;
                 }
+                
+                Some(deleted)
             },
             Err(e) => {
                 tracing::warn!("[UNREGISTER] Failed to create Redis client: {}", e);

src/api/identity_api.rs

@@ -96,11 +96,21 @@ async fn list_files(
             .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
 
         let data = if let Some(v) = video {
+            let chunk_count: i64 = sqlx::query_scalar(&format!(
+                "SELECT COUNT(*) FROM {} WHERE file_uuid = $1",
+                crate::core::db::schema::table_name("chunk")
+            ))
+            .bind(&v.file_uuid)
+            .fetch_one(state.db.pool())
+            .await
+            .unwrap_or(0);
+
             vec![FileItem {
                 file_uuid: v.file_uuid,
                 file_name: v.file_name,
                 file_path: v.file_path,
                 status: v.status.as_str().to_string(),
+                total_chunks: chunk_count,
             }]
         } else {
             vec![]
@@ -124,18 +134,45 @@ async fn list_files(
         .await
         .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
 
-    let data = records
+    let total = records.1;
+
+    let mut data: Vec<FileItem> = records
         .0
         .into_iter()
         .map(|r| FileItem {
-            file_uuid: r.file_uuid,
+            file_uuid: r.file_uuid.clone(),
             file_name: r.file_name,
             file_path: r.file_path,
             status: r.status.as_str().to_string(),
+            total_chunks: 0,
         })
         .collect();
 
-    let total = records.1;
+    // Fetch chunk counts for all files in one query
+    let uuids: Vec<String> = data.iter().map(|f| f.file_uuid.clone()).collect();
+    if !uuids.is_empty() {
+        let chunk_table = crate::core::db::schema::table_name("chunk");
+        let placeholders: Vec<String> = (1..=uuids.len()).map(|i| format!("${}", i)).collect();
+        let query_str = format!(
+            "SELECT file_uuid, COUNT(*) as cnt FROM {} WHERE file_uuid IN ({}) GROUP BY file_uuid",
+            chunk_table,
+            placeholders.join(", ")
+        );
+
+        let chunk_counts: Vec<(String, i64)> = sqlx::query_as(&query_str)
+            .fetch_all(state.db.pool())
+            .await
+            .unwrap_or_default();
+
+        let count_map: std::collections::HashMap<String, i64> =
+            chunk_counts.into_iter().collect();
+
+        for item in &mut data {
+            if let Some(cnt) = count_map.get(&item.file_uuid) {
+                item.total_chunks = *cnt;
+            }
+        }
+    }
 
     Ok(Json(FilesResponse {
         success: true,
@@ -161,6 +198,8 @@ pub struct FileItem {
     pub file_name: String,
     pub file_path: String,
     pub status: String,
+    #[serde(default)]
+    pub total_chunks: i64,
 }
 
 #[derive(Debug, Serialize)]

src/api/media_api.rs

@@ -51,6 +51,7 @@ fn ffmpeg_cmd() -> std::process::Command {
 
 pub fn bbox_routes() -> Router<crate::api::types::AppState> {
     Router::new()
+        .route("/api/v1/face-thumbnail", get(face_thumbnail_compat))
         .route(
             "/api/v1/file/:file_uuid/video/bbox",
             get(bbox_overlay_video),
@@ -768,14 +769,27 @@ async fn stream_video(
 
 #[derive(Debug, serde::Deserialize)]
 struct ThumbQuery {
+    uuid: Option<String>,
     frame: Option<i64>,
     x: Option<i32>,
     y: Option<i32>,
     w: Option<i32>,
     h: Option<i32>,
+    bbox_x: Option<i32>,
+    bbox_y: Option<i32>,
+    bbox_w: Option<i32>,
+    bbox_h: Option<i32>,
     trace_id: Option<i32>,
 }
 
+async fn face_thumbnail_compat(
+    State(state): State<crate::api::types::AppState>,
+    Query(q): Query<ThumbQuery>,
+) -> Result<impl IntoResponse, StatusCode> {
+    let file_uuid = q.uuid.clone().ok_or(StatusCode::BAD_REQUEST)?;
+    face_thumbnail(State(state), Path(file_uuid), Query(q)).await
+}
+
 async fn face_thumbnail(
     State(state): State<crate::api::types::AppState>,
     Path(file_uuid): Path<String>,
@@ -853,7 +867,12 @@ async fn face_thumbnail(
         }
     }
 
-    if let (Some(x), Some(y), Some(w), Some(h)) = (q.x, q.y, q.w, q.h) {
+    let crop_x = q.x.or(q.bbox_x);
+    let crop_y = q.y.or(q.bbox_y);
+    let crop_w = q.w.or(q.bbox_w);
+    let crop_h = q.h.or(q.bbox_h);
+
+    if let (Some(x), Some(y), Some(w), Some(h)) = (crop_x, crop_y, crop_w, crop_h) {
         if let (Some(vw), Some(vh)) = (video_width, video_height) {
             crate::core::thumbnail::validator::validate_crop(x, y, w, h, vw, vh).map_err(|e| {
                 tracing::warn!("[thumbnail] Crop validation failed: {}", e);
@@ -863,7 +882,7 @@ async fn face_thumbnail(
     }
 
     let select = format!("select=eq(n\\,{})", frame);
-    let vf = if let (Some(x), Some(y), Some(w), Some(h)) = (q.x, q.y, q.w, q.h) {
+    let vf = if let (Some(x), Some(y), Some(w), Some(h)) = (crop_x, crop_y, crop_w, crop_h) {
         format!("{},crop={}:{}:{}:{}", select, w, h, x, y)
     } else {
         select
@@ -1266,11 +1285,16 @@ async fn media_proxy_handler(
     match type_ {
         "thumbnail" => {
             let thumb_query = ThumbQuery {
+                uuid: None,
                 frame: params.get("frame").and_then(|v| v.parse().ok()),
                 x: params.get("x").and_then(|v| v.parse().ok()),
                 y: params.get("y").and_then(|v| v.parse().ok()),
                 w: params.get("w").and_then(|v| v.parse().ok()),
                 h: params.get("h").and_then(|v| v.parse().ok()),
+                bbox_x: params.get("bbox_x").and_then(|v| v.parse().ok()),
+                bbox_y: params.get("bbox_y").and_then(|v| v.parse().ok()),
+                bbox_w: params.get("bbox_w").and_then(|v| v.parse().ok()),
+                bbox_h: params.get("bbox_h").and_then(|v| v.parse().ok()),
                 trace_id: params.get("trace_id").and_then(|v| v.parse().ok()),
             };
             face_thumbnail(State(state), Path(uuid.clone()), Query(thumb_query))

src/api/scan.rs

@@ -40,7 +40,6 @@ struct ProcessorStatus {
 #[derive(Debug, Serialize, Default)]
 struct PostgresStats {
     sentence_chunks: i64,
-    trace_chunks: i64,
     relationship_chunks: i64,
     identities: i64,
     file_identities: i64,
@@ -509,14 +508,22 @@ async fn get_ingestion_status(
         "SELECT COUNT(*) FROM {chunk} WHERE file_uuid = '{file_uuid}' AND chunk_type = 'sentence'"
     ));
     let sentence_embedded = count_sql!(&format!("SELECT COUNT(*) FROM {chunk} WHERE file_uuid = '{file_uuid}' AND chunk_type = 'sentence' AND embedding IS NOT NULL"));
+
+    // OCR statistics
+    let pre_chunks_table = schema::table_name("pre_chunks");
+    let ocr_pre_chunks: i64 = count_sql!(&format!(
+        "SELECT COUNT(*) FROM {pre_chunks_table} WHERE file_uuid = '{file_uuid}' AND processor_type = 'ocr'"
+    ));
+    let ocr_only_chunks: i64 = count_sql!(&format!(
+        "SELECT COUNT(*) FROM {chunk} WHERE file_uuid = '{file_uuid}' AND chunk_type = 'sentence' \
+         AND (content->>'text' = '' OR content->>'text' IS NULL) \
+         AND content->>'ocr_text' IS NOT NULL AND content->>'ocr_text' != ''"
+    ));
     let scene_count = count_sql!(&format!(
         "SELECT COUNT(*) FROM {chunk} WHERE file_uuid = '{file_uuid}' AND chunk_type = 'cut'"
     ));
     let face_total = face_total;
     let trace_count = trace_count;
-    let trace_chunks = count_sql!(&format!(
-        "SELECT COUNT(*) FROM {chunk} WHERE file_uuid = '{file_uuid}' AND chunk_type = 'trace'"
-    ));
     let identity_count = identity_count;
     let tkg_nodes = count_sql!(&format!(
         "SELECT COUNT(*) FROM {} WHERE file_uuid = '{file_uuid}'",
@@ -583,11 +590,23 @@ async fn get_ingestion_status(
 
     let strangers = strangers;
 
+    // Check if job is completed - if so, all ingestion steps are considered done
+    let mj_table = schema::table_name("monitor_jobs");
+    let job_completed: bool = sqlx::query_scalar::<_, String>(&format!(
+        "SELECT status FROM {mj_table} WHERE uuid = $1"
+    ))
+    .bind(&file_uuid)
+    .fetch_optional(pool)
+    .await
+    .unwrap_or(None)
+    .map(|s| s == "completed")
+    .unwrap_or(false);
+
     macro_rules! step {
         ($name:expr, $done:expr, $detail:expr) => {
             IngestionStep {
                 name: $name.into(),
-                status: if $done { "done" } else { "pending" }.into(),
+                status: if $done || job_completed { "done" } else { "pending" }.into(),
                 detail: $detail,
             }
         };
@@ -599,6 +618,16 @@ async fn get_ingestion_status(
             sentence_count > 0,
             Some(format!("{sentence_count} sentence chunks"))
         ),
+        step!(
+            "rule1_ocr",
+            ocr_pre_chunks > 0,
+            Some(format!("{ocr_pre_chunks} OCR frames"))
+        ),
+        step!(
+            "rule1_ocr_chunks",
+            ocr_only_chunks > 0,
+            Some(format!("{ocr_only_chunks} OCR-only chunks"))
+        ),
         step!(
             "auto_vectorize",
             sentence_embedded > 0,
@@ -609,11 +638,6 @@ async fn get_ingestion_status(
             trace_count > 0,
             Some(format!("{trace_count} traces / {face_total} detections"))
         ),
-        step!(
-            "trace_chunks",
-            trace_chunks > 0,
-            Some(format!("{trace_chunks} trace chunks"))
-        ),
         // TKG Nodes
         step!("tkg_face_track", face_track_nodes > 0, Some(format!("{face_track_nodes} nodes"))),
         step!("tkg_gaze_track", gaze_track_nodes > 0, Some(format!("{gaze_track_nodes} nodes"))),
@@ -736,13 +760,6 @@ async fn get_file_stats(
         .fetch_one(pool)
         .await
         .unwrap_or(0),
-        trace_chunks: sqlx::query_scalar::<_, i64>(&format!(
-            "SELECT COUNT(*) FROM {chunk_table} WHERE file_uuid = $1 AND chunk_type = 'trace'"
-        ))
-        .bind(&file_uuid)
-        .fetch_one(pool)
-        .await
-        .unwrap_or(0),
         relationship_chunks: sqlx::query_scalar::<_, i64>(&format!(
             "SELECT COUNT(*) FROM {chunk_table} WHERE file_uuid = $1 AND chunk_type = 'relationship'"
         ))
@@ -801,7 +818,11 @@ async fn get_file_stats(
 
     // Text chunk stats (rule1 collection)
     let schema = std::env::var("DATABASE_SCHEMA").unwrap_or_else(|_| "dev".to_string());
-    let rule1_collection = format!("momentry_{}_rule1_v2", schema);
+    let rule1_collection = if schema == "public" {
+        "momentry_rule1".to_string()
+    } else {
+        format!("momentry_{}_rule1_v2", schema)
+    };
     let text_filter = json!({
         "must": [{"key": "file_uuid", "match": {"value": file_uuid}}]
     });
@@ -832,24 +853,31 @@ async fn get_file_stats(
     let tkg_nodes_table = schema::table_name("tkg_nodes");
     let tkg_edges_table = schema::table_name("tkg_edges");
 
+    let tkg_nodes_total: i64 = sqlx::query_scalar::<_, i64>(&format!("SELECT COUNT(*) FROM {} WHERE file_uuid = $1", tkg_nodes_table))
+        .bind(&file_uuid).fetch_one(pool).await.unwrap_or(0);
+    let tkg_edges_total: i64 = sqlx::query_scalar::<_, i64>(&format!("SELECT COUNT(*) FROM {} WHERE file_uuid = $1", tkg_edges_table))
+        .bind(&file_uuid).fetch_one(pool).await.unwrap_or(0);
+
     let tkg = TkgFileStats {
-        face_track_nodes: count_by_type(pool, &tkg_nodes_table, &file_uuid, "face_track").await,
+        total_nodes: tkg_nodes_total,
-        gaze_track_nodes: count_by_type(pool, &tkg_nodes_table, &file_uuid, "gaze_track").await,
+        total_edges: tkg_edges_total,
-        lip_track_nodes: count_by_type(pool, &tkg_nodes_table, &file_uuid, "lip_track").await,
+        face_track_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "face_track").await,
-        text_region_nodes: count_by_type(pool, &tkg_nodes_table, &file_uuid, "text_region").await,
+        gaze_track_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "gaze_track").await,
-        appearance_nodes: count_by_type(pool, &tkg_nodes_table, &file_uuid, "appearance_trace").await,
+        lip_track_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "lip_track").await,
-        accessory_nodes: count_by_type(pool, &tkg_nodes_table, &file_uuid, "accessory").await,
+        text_region_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "text_trace").await,
-        object_nodes: count_by_type(pool, &tkg_nodes_table, &file_uuid, "yolo_object").await,
+        appearance_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "appearance_trace").await,
-        hand_nodes: count_by_type(pool, &tkg_nodes_table, &file_uuid, "hand").await,
+        accessory_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "accessory").await,
-        speaker_nodes: count_by_type(pool, &tkg_nodes_table, &file_uuid, "speaker").await,
+        object_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "yolo_object").await,
-        co_occurrence_edges: count_by_type(pool, &tkg_edges_table, &file_uuid, "CO_OCCURS_WITH").await,
+        hand_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "hand").await,
-        speaker_face_edges: count_by_type(pool, &tkg_edges_table, &file_uuid, "SPEAKS_AS").await,
+        speaker_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "speaker").await,
-        face_face_edges: count_by_type(pool, &tkg_edges_table, &file_uuid, "FACE_TO_FACE").await,
+        co_occurrence_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "CO_OCCURS_WITH").await,
-        mutual_gaze_edges: count_by_type(pool, &tkg_edges_table, &file_uuid, "MUTUAL_GAZE").await,
+        speaker_face_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "SPEAKS_AS").await,
-        lip_sync_edges: count_by_type(pool, &tkg_edges_table, &file_uuid, "LIP_SYNC").await,
+        face_face_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "FACE_TO_FACE").await,
-        has_appearance_edges: count_by_type(pool, &tkg_edges_table, &file_uuid, "HAS_APPEARANCE").await,
+        mutual_gaze_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "MUTUAL_GAZE").await,
-        wears_edges: count_by_type(pool, &tkg_edges_table, &file_uuid, "WEARS").await,
+        lip_sync_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "LIP_SYNC").await,
-        hand_object_edges: count_by_type(pool, &tkg_edges_table, &file_uuid, "HAND_OBJECT").await,
+        has_appearance_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "HAS_APPEARANCE").await,
+        wears_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "WEARS").await,
+        hand_object_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "HAND_OBJECT").await,
         ..Default::default()
     };
 
@@ -886,13 +914,25 @@ async fn get_file_stats(
     }))
 }
 
-async fn count_by_type(pool: &sqlx::PgPool, table: &str, file_uuid: &str, type_val: &str) -> i64 {
+async fn count_nodes(pool: &sqlx::PgPool, table: &str, file_uuid: &str, node_type: &str) -> i64 {
     sqlx::query_scalar::<_, i64>(&format!(
-        "SELECT COUNT(*) FROM {} WHERE file_uuid = $1 AND (node_type = $2 OR edge_type = $2)",
+        "SELECT COUNT(*) FROM {} WHERE file_uuid = $1 AND node_type = $2",
         table
     ))
     .bind(file_uuid)
-    .bind(type_val)
+    .bind(node_type)
+    .fetch_one(pool)
+    .await
+    .unwrap_or(0)
+}
+
+async fn count_edges(pool: &sqlx::PgPool, table: &str, file_uuid: &str, edge_type: &str) -> i64 {
+    sqlx::query_scalar::<_, i64>(&format!(
+        "SELECT COUNT(*) FROM {} WHERE file_uuid = $1 AND edge_type = $2",
+        table
+    ))
+    .bind(file_uuid)
+    .bind(edge_type)
     .fetch_one(pool)
     .await
     .unwrap_or(0)
@@ -921,10 +961,82 @@ async fn get_pipeline_progress_handler(
     State(state): State<AppState>,
     Path(file_uuid): Path<String>,
 ) -> Result<Json<crate::core::progress::PipelineProgress>, StatusCode> {
-    let redis_lock = state.redis_cache.get_client().await;
+    let pool = state.db.pool();
-    let redis_guard = redis_lock.read().await;
+    let chunk_table = schema::table_name("chunk");
-    let pipeline = crate::core::progress::get_pipeline_progress(&*redis_guard, &file_uuid)
+    let tkg_nodes_table = schema::table_name("tkg_nodes");
-        .await
+    let tkg_edges_table = schema::table_name("tkg_edges");
-        .unwrap_or_else(|| crate::core::progress::PipelineProgress::new(&file_uuid));
+    let pr_table = schema::table_name("processor_results");
-    Ok(Json(pipeline))
+    let mj_table = schema::table_name("monitor_jobs");
+
+    // Compute actual progress from DB state
+    let sentence_count: i64 = sqlx::query_scalar::<_, i64>(
+        &format!("SELECT COUNT(*) FROM {chunk_table} WHERE file_uuid = $1 AND chunk_type = 'sentence'")
+    ).bind(&file_uuid).fetch_one(pool).await.unwrap_or(0);
+
+    let sentence_embedded: i64 = sqlx::query_scalar::<_, i64>(
+        &format!("SELECT COUNT(*) FROM {chunk_table} WHERE file_uuid = $1 AND chunk_type = 'sentence' AND embedding IS NOT NULL")
+    ).bind(&file_uuid).fetch_one(pool).await.unwrap_or(0);
+
+    let face_traced: i64 = sqlx::query_scalar::<_, i64>(
+        &format!("SELECT COUNT(*) FROM {pr_table} pr JOIN {mj_table} mj ON pr.job_id = mj.id WHERE mj.uuid = $1 AND pr.processor = 'face' AND pr.status = 'completed'")
+    ).bind(&file_uuid).fetch_one(pool).await.unwrap_or(0);
+
+    let tkg_node_count: i64 = sqlx::query_scalar::<_, i64>(
+        &format!("SELECT COUNT(*) FROM {tkg_nodes_table} WHERE file_uuid = $1")
+    ).bind(&file_uuid).fetch_one(pool).await.unwrap_or(0);
+
+    let tkg_edge_count: i64 = sqlx::query_scalar::<_, i64>(
+        &format!("SELECT COUNT(*) FROM {tkg_edges_table} WHERE file_uuid = $1")
+    ).bind(&file_uuid).fetch_one(pool).await.unwrap_or(0);
+
+    let relationship_count: i64 = sqlx::query_scalar::<_, i64>(
+        &format!("SELECT COUNT(*) FROM {chunk_table} WHERE file_uuid = $1 AND chunk_type = 'relationship'")
+    ).bind(&file_uuid).fetch_one(pool).await.unwrap_or(0);
+
+    let asrx_completed: i64 = sqlx::query_scalar::<_, i64>(
+        &format!("SELECT COUNT(*) FROM {pr_table} pr JOIN {mj_table} mj ON pr.job_id = mj.id WHERE mj.uuid = $1 AND pr.processor = 'asrx' AND pr.status = 'completed'")
+    ).bind(&file_uuid).fetch_one(pool).await.unwrap_or(0);
+
+    // Determine processor completion
+    let processors_done = asrx_completed > 0;
+
+    let mut pp = crate::core::progress::PipelineProgress::new(&file_uuid);
+
+    if processors_done {
+        pp.update_stage("processors", 1.0, "completed", None);
+    }
+    if sentence_count > 0 {
+        let detail = if sentence_embedded > 0 {
+            Some(format!("{} chunks, {} embedded", sentence_count, sentence_embedded))
+        } else {
+            Some(format!("{} chunks", sentence_count))
+        };
+        pp.update_stage("rule1_ingestion", 1.0, "completed", detail);
+    }
+    if face_traced > 0 {
+        pp.update_stage("face_tracing", 1.0, "completed", None);
+    }
+    if tkg_node_count > 0 {
+        pp.update_stage("tkg_nodes", 1.0, "completed", Some(format!("{} nodes", tkg_node_count)));
+    }
+    if tkg_edge_count > 0 {
+        pp.update_stage("tkg_edges", 1.0, "completed", Some(format!("{} edges", tkg_edge_count)));
+    }
+    if relationship_count > 0 {
+        pp.update_stage("rule2_ingestion", 1.0, "completed", Some(format!("{} chunks", relationship_count)));
+    }
+
+    // Check identity agent from _seeds
+    use crate::core::db::qdrant_db::QdrantDb;
+    use serde_json::json;
+    let qdrant = QdrantDb::new();
+    let schema = std::env::var("DATABASE_SCHEMA").unwrap_or_else(|_| "dev".to_string());
+    let seeds_collection = if schema == "public" { "momentry_public_speaker" } else { &format!("momentry_{}_speaker", schema) };
+    let seeds_filter = json!({"must": [{"key": "file_uuid", "match": {"value": &file_uuid}}]});
+    let seed_points = qdrant.scroll_all_points("_seeds", seeds_filter, 100).await.unwrap_or_default();
+    if !seed_points.is_empty() {
+        pp.update_stage("identity_agent", 1.0, "completed", Some(format!("{} seeds", seed_points.len())));
+    }
+
+    Ok(Json(pp))
 }

src/api/search.rs

@@ -34,6 +34,7 @@ pub struct SearchResult {
     pub end_time: f64,
     pub raw_text: Option<String>,
     pub summary: Option<String>,
+    pub text_content: Option<String>,
     pub metadata: Option<serde_json::Value>,
     pub similarity: Option<f64>,
     pub file_name: Option<String>,
@@ -82,6 +83,7 @@ async fn enrich_from_pg(
             end_time: p.end_time,
             raw_text: None,
             summary: Some(p.summary),
+            text_content: p.text_content.clone(),
             metadata: p.metadata.clone(),
             similarity: Some(qdrant_score as f64),
             file_name: None,
@@ -109,6 +111,7 @@ fn pg_result_to_search(p: &SemanticSearchResult) -> SearchResult {
         end_time: p.end_time,
         raw_text: None,
         summary: Some(p.summary.clone()),
+        text_content: p.text_content.clone(),
         metadata: p.metadata.clone(),
         similarity: p.similarity,
         file_name: None,
@@ -381,12 +384,16 @@ pub async fn smart_search(
     let mut final_results = Vec::new();
     for mr in ranked.iter().take(limit * 3) {
         // 取更多結果以便過濾
-        if let Some(pg) = db
+        // Use no_embedding version for keyword results, regular for semantic
-            .get_chunk_by_file_and_chunk_id(&mr.file_uuid, &mr.chunk_id)
+        let pg_opt = if mr.keyword_score.is_some() && mr.semantic_score.is_none() {
-            .await
+            db.get_chunk_by_id_no_embedding(&mr.file_uuid, &mr.chunk_id).await
-            .ok()
+        } else {
-            .flatten()
+            db.get_chunk_by_file_and_chunk_id(&mr.file_uuid, &mr.chunk_id).await
-        {
+        };
+        if let Some(pg) = pg_opt.ok().flatten() {
+            // 關鍵字結果跳過 text_match 過濾（search_bm25 已經匹配過）
+            let is_keyword_only = mr.keyword_score.is_some() && mr.semantic_score.is_none();
+            if !is_keyword_only {
                 // 關鍵字過濾: CJK 用子字串匹配，英文用單詞邊界匹配
                 let summary_lower = pg.summary.to_lowercase();
                 let query_words: Vec<String> = query_lower
@@ -414,9 +421,17 @@ pub async fn smart_search(
                     }
                 };
 
-            if !text_match && mr.semantic_score.is_none() {
+                if !text_match {
                     continue;
                 }
+            }
+
+            // 使用 text_content 如果 summary 為空
+            let display_text = if pg.summary.is_empty() {
+                pg.text_content.clone().unwrap_or_default()
+            } else {
+                pg.summary.clone()
+            };
 
             final_results.push(SearchResult {
                 id: 0,
@@ -430,6 +445,7 @@ pub async fn smart_search(
                 end_time: pg.end_time,
                 raw_text: None,
                 summary: Some(pg.summary),
+                text_content: pg.text_content.clone(),
                 metadata: pg.metadata.clone(),
                 similarity: Some(mr.score),
                 file_name: None,

src/api/trace_agent_api.rs

@@ -3,7 +3,7 @@ use axum::{
     extract::{Path, Query, State},
     http::{header, StatusCode},
     response::{IntoResponse, Json, Response},
-    routing::{get, post},
+    routing::{delete, get, post},
     Router,
 };
 use serde::{Deserialize, Serialize};
@@ -39,6 +39,8 @@ pub fn trace_agent_routes() -> Router<crate::api::types::AppState> {
             get(get_cooccurrence),
         )
         .route("/api/v1/file/:file_uuid/tkg/rebuild", post(rebuild_tkg))
+        .route("/api/v1/file/:file_uuid/tkg", get(get_tkg_operations))
+        .route("/api/v1/file/:file_uuid/tkg", delete(delete_tkg))
         .route("/api/v1/file/:file_uuid/rule2", post(ingest_rule2))
         .route(
             "/api/v1/file/:file_uuid/representative-frame",
@@ -980,58 +982,41 @@ struct TkgRebuildResponse {
     error: Option<String>,
 }
 
+#[derive(Deserialize)]
+struct RebuildParams {
+    force: Option<bool>,
+}
+
 async fn rebuild_tkg(
     State(state): State<crate::api::types::AppState>,
     Path(file_uuid): Path<String>,
+    Query(params): Query<RebuildParams>,
 ) -> Json<TkgRebuildResponse> {
     use crate::core::chunk::rule2_ingest::ingest_rule2;
+    use crate::core::tkg::TkgService;
     use tracing::info;
 
-    let redis = crate::core::db::RedisClient::new().ok();
+    let db = state.db.clone();
-    let result = crate::core::processor::tkg::build_tkg(&state.db, &file_uuid, &OUTPUT_DIR, redis.map(Arc::new)).await;
+    let tkg_service = TkgService::new(state.db);
+    let force = params.force.unwrap_or(false);
+    let result = tkg_service.rebuild(&file_uuid, &OUTPUT_DIR, force).await;
 
     match result {
         Ok(r) => {
-            let total_edges = r.speaker_face_edges
+            // Always trigger Rule 2 (even with 0 edges)
-                + r.mutual_gaze_edges
-                + r.face_face_edges
-                + r.co_occurrence_edges
-                + r.has_appearance_edges
-                + r.wears_edges;
-
-            if total_edges > 0 {
             info!(
-                    "[TKG] {} relationship edges found, triggering Rule 2 ingestion...",
+                "[TKG] Rebuild completed for {}: {} nodes, {} edges",
-                    total_edges
+                file_uuid, r.total_nodes(), r.total_edges()
             );
-                match ingest_rule2(state.db.pool(), &file_uuid, None, None).await {
+            match ingest_rule2(db.pool(), &file_uuid, None, None).await {
                 Ok(count) => info!("[TKG] Rule 2 created {} relationship chunks", count),
                 Err(e) => info!("[TKG] Rule 2 ingestion failed: {}", e),
             }
-            }
 
             Json(TkgRebuildResponse {
                 success: true,
                 file_uuid,
-                result: Some(serde_json::json!({
+                result: Some(r.to_json()),
-                    "face_track_nodes": r.face_track_nodes,
-                    "gaze_track_nodes": r.gaze_track_nodes,
-                    "lip_track_nodes": r.lip_track_nodes,
-                    "text_region_nodes": r.text_region_nodes,
-                    "appearance_trace_nodes": r.appearance_trace_nodes,
-                    "accessory_nodes": r.accessory_nodes,
-                    "object_nodes": r.object_nodes,
-                    "hand_nodes": r.hand_nodes,
-                    "speaker_nodes": r.speaker_nodes,
-                    "co_occurrence_edges": r.co_occurrence_edges,
-                    "speaker_face_edges": r.speaker_face_edges,
-                    "face_face_edges": r.face_face_edges,
-                    "mutual_gaze_edges": r.mutual_gaze_edges,
-                    "lip_sync_edges": r.lip_sync_edges,
-                    "has_appearance_edges": r.has_appearance_edges,
-                    "wears_edges": r.wears_edges,
-                    "hand_object_edges": r.hand_object_edges,
-                })),
                 error: None,
             })
         }
@@ -1044,6 +1029,28 @@ async fn rebuild_tkg(
     }
 }
 
+async fn get_tkg_operations(
+    State(state): State<crate::api::types::AppState>,
+    Path(file_uuid): Path<String>,
+) -> Json<Vec<crate::core::tkg::models::TkgOperationLog>> {
+    use crate::core::tkg::TkgService;
+    let tkg_service = TkgService::new(state.db);
+    let operations = tkg_service.get_operations(&file_uuid).await.unwrap_or_default();
+    Json(operations)
+}
+
+async fn delete_tkg(
+    State(state): State<crate::api::types::AppState>,
+    Path(file_uuid): Path<String>,
+) -> Json<serde_json::Value> {
+    use crate::core::tkg::TkgService;
+    let tkg_service = TkgService::new(state.db);
+    match tkg_service.delete_tkg_and_logs(&file_uuid).await {
+        Ok(_) => Json(serde_json::json!({"success": true, "message": "TKG deleted successfully"})),
+        Err(e) => Json(serde_json::json!({"success": false, "error": e.to_string()})),
+    }
+}
+
 // ── Representative Frame (JSON) ───────────────────────────────────
 
 use crate::core::processor::tkg;

src/core/agent/tools.rs

@@ -1092,3 +1092,68 @@ pub async fn exec_tkg_node_detail(
         None => Err("Node not found".to_string()),
     }
 }
+
+/// Search for people by clothing color using appearance data
+pub async fn exec_search_by_appearance(pool: &sqlx::PgPool, args: &serde_json::Value) -> Result<String, String> {
+    let file_uuid = args.get("file_uuid")
+        .and_then(|v| v.as_str())
+        .ok_or("file_uuid is required".to_string())?;
+    let color = args.get("color")
+        .and_then(|v| v.as_str())
+        .ok_or("color is required (red, blue, green, yellow, orange, cyan, purple, white, black)".to_string())?;
+
+    let output_dir = std::env::var("MOMENTRY_OUTPUT_DIR")
+        .unwrap_or_else(|_| "/Users/accusys/momentry/output".to_string());
+    let scripts_dir = std::env::var("MOMENTRY_SCRIPTS_DIR")
+        .unwrap_or_else(|_| "/Users/accusys/momentry_core/scripts".to_string());
+    let script_path = format!("{}/clothing_color_search.py", scripts_dir);
+    let appearance_path = format!("{}/{}.appearance.json", output_dir, file_uuid);
+    let output_path = format!("{}/{}.color_search_{}.json", output_dir, file_uuid, color);
+
+    if !std::path::Path::new(&appearance_path).exists() {
+        return Err(format!("appearance.json not found for file {}", file_uuid));
+    }
+
+    // Get video path from videos table
+    let videos_table = schema::table_name("videos");
+    let video_path: Option<String> = sqlx::query_scalar(&format!(
+        "SELECT file_path FROM {} WHERE file_uuid = $1", videos_table
+    ))
+    .bind(file_uuid)
+    .fetch_optional(pool)
+    .await
+    .map_err(|e| e.to_string())?;
+
+    let video_path = video_path.unwrap_or_default();
+    if video_path.is_empty() {
+        return Err("Video path not found".to_string());
+    }
+
+    let executor = crate::core::processor::PythonExecutor::new()
+        .map_err(|e| e.to_string())?;
+    
+    executor.run(
+        &script_path,
+        &[
+            "--file-uuid", file_uuid,
+            "--color", color,
+            "--video-path", &video_path,
+            "--appearance-path", &appearance_path,
+            "--output", &output_path,
+        ],
+        None,
+        "CLOTHING_COLOR_SEARCH",
+        Some(std::time::Duration::from_secs(300)),
+    )
+    .await
+    .map_err(|e| e.to_string())?;
+
+    // Read results
+    if std::path::Path::new(&output_path).exists() {
+        let content = std::fs::read_to_string(&output_path)
+            .map_err(|e| e.to_string())?;
+        Ok(content)
+    } else {
+        Err("Color search output not found".to_string())
+    }
+}

src/core/chunk/rule1_ingest.rs

@@ -13,16 +13,16 @@ pub async fn execute_rule1(db: &PostgresDb, file_uuid: &str, fps: f64) -> Result
     let pool = db.pool();
     let pre_chunks_table = schema::table_name("pre_chunks");
 
-    let asr_segments = fetch_asr_segments(pool, file_uuid, &pre_chunks_table).await?;
+    let asr_segments = fetch_asr_segments(pool, file_uuid, &pre_chunks_table, fps).await?;
-    let ocr_map = fetch_ocr_texts(pool, file_uuid, &pre_chunks_table).await?;
+    let ocr_map = fetch_ocr_texts(pool, file_uuid, &pre_chunks_table, fps).await?;
 
     let video = db
         .get_video_by_uuid(file_uuid)
         .await?
         .context("Video not found")?;
 
-    if asr_segments.is_empty() {
+    if asr_segments.is_empty() && ocr_map.is_empty() {
-        info!("Rule 1: no ASR segments for video {}", file_uuid);
+        info!("Rule 1: no ASR segments or OCR for video {}", file_uuid);
         return Ok(0);
     }
 
@@ -30,13 +30,12 @@ pub async fn execute_rule1(db: &PostgresDb, file_uuid: &str, fps: f64) -> Result
     let mut count = 0;
     let mut tx = pool.begin().await?;
 
-    for (idx, seg) in asr_segments.iter().enumerate() {
+    // Phase 1: ASRX segments (pure speech, NO OCR merge)
-        let ocr_text = collect_ocr_text(seg.start_frame, seg.end_frame, &ocr_map);
+    for seg in asr_segments.iter() {
-        let combined_text = if ocr_text.is_empty() {
+        // Skip chunks with no text
-            seg.text.clone()
+        if seg.text.trim().is_empty() {
-        } else {
+            continue;
-            format!("{} {}", seg.text, ocr_text)
+        }
-        };
 
         let metadata = serde_json::json!({
             "language": seg.language,
@@ -44,13 +43,13 @@ pub async fn execute_rule1(db: &PostgresDb, file_uuid: &str, fps: f64) -> Result
 
         let content = serde_json::json!({
             "text": seg.text,
-            "ocr_text": ocr_text,
+            "ocr_text": "",
         });
 
         let chunk = Chunk::from_seconds(
             file_id as i32,
             file_uuid.to_string(),
-            format!("{}", idx),
+            format!("{}", count),
             ChunkType::Sentence,
             ChunkRule::Rule1,
             seg.start_time,
@@ -59,7 +58,7 @@ pub async fn execute_rule1(db: &PostgresDb, file_uuid: &str, fps: f64) -> Result
             content,
         )
         .with_metadata(metadata)
-        .with_text_content(combined_text);
+        .with_text_content(seg.text.clone());
 
         db.store_chunk_in_tx(&chunk, &mut tx).await?;
 
@@ -67,17 +66,59 @@ pub async fn execute_rule1(db: &PostgresDb, file_uuid: &str, fps: f64) -> Result
 
         if count % 100 == 0 {
             info!(
-                "Rule 1: Processed {} segments for video {}",
+                "Rule 1: Processed {} ASRX segments for video {}",
                 count, file_uuid
             );
         }
     }
 
+    let asrx_count = count;
+
+    // Phase 2: OCR-only chunks (all OCR frames grouped by proximity)
+    let ocr_chunks = group_ocr_frames(&ocr_map, fps);
+    for (start_frame, end_frame, ocr_text) in ocr_chunks {
+        if ocr_text.trim().is_empty() {
+            continue;
+        }
+
+        let start_time = start_frame as f64 / fps;
+        let end_time = (end_frame + 1) as f64 / fps;
+
+        let metadata = serde_json::json!({
+            "language": "ocr",
+        });
+
+        let content = serde_json::json!({
+            "text": "",
+            "ocr_text": ocr_text.clone(),
+        });
+
+        let chunk = Chunk::from_seconds(
+            file_id as i32,
+            file_uuid.to_string(),
+            format!("{}", count),
+            ChunkType::Sentence,
+            ChunkRule::Rule1,
+            start_time,
+            end_time,
+            fps,
+            content,
+        )
+        .with_metadata(metadata)
+        .with_text_content(ocr_text);
+
+        db.store_chunk_in_tx(&chunk, &mut tx).await?;
+
+        count += 1;
+    }
+
+    let ocr_only_count = count - asrx_count;
+
     tx.commit().await?;
 
     info!(
-        "Rule 1 completed: {} sentence chunks created for video {}",
+        "Rule 1 completed: {} sentence chunks for video {} ({} ASRX + {} OCR-only)",
-        count, file_uuid
+        count, file_uuid, asrx_count, ocr_only_count
     );
 
     Ok(count)
@@ -97,6 +138,7 @@ async fn fetch_asr_segments(
     pool: &PgPool,
     file_uuid: &str,
     table: &str,
+    fps: f64,
 ) -> Result<Vec<AsrSegment>> {
     let query = format!(
         r#"
@@ -114,8 +156,6 @@ async fn fetch_asr_segments(
     let segments: Vec<AsrSegment> = rows
         .iter()
         .map(|row| {
-            let start_frame: i64 = row.try_get("start_frame").unwrap_or(0);
-            let end_frame: i64 = row.try_get("end_frame").unwrap_or(0);
             let start_time: f64 = row.try_get("start_time").unwrap_or(0.0);
             let end_time_raw: Option<f64> = row.try_get("end_time").ok();
             let data: Value = row.try_get("data").unwrap_or(Value::Null);
@@ -124,6 +164,13 @@ async fn fetch_asr_segments(
                 .or_else(|| data.get("end_time").and_then(|v| v.as_f64()))
                 .unwrap_or(0.0);
 
+            let start_frame = (start_time * fps) as i64;
+            let end_frame = if end_time > 0.0 {
+                (end_time * fps) as i64
+            } else {
+                start_frame
+            };
+
             if end_time <= 0.0 {
                 warn!(
                     "ASR segment end_time is 0.0 for file {} (frame {}..{})",
@@ -155,14 +202,15 @@ async fn fetch_ocr_texts(
     pool: &PgPool,
     file_uuid: &str,
     table: &str,
+    fps: f64,
 ) -> Result<BTreeMap<i64, Vec<String>>> {
     let query = format!(
         r#"
         SELECT 
-            coordinate_index as frame, data
+            start_frame, start_time, end_time, data
         FROM {}
         WHERE file_uuid = $1 AND processor_type = 'ocr'
-        ORDER BY coordinate_index
+        ORDER BY start_frame
         "#,
         table
     );
@@ -171,9 +219,16 @@ async fn fetch_ocr_texts(
 
     let mut map: BTreeMap<i64, Vec<String>> = BTreeMap::new();
     for row in rows {
-        let frame: i64 = row.try_get("frame").unwrap_or(0);
+        let start_time: f64 = row.try_get("start_time").unwrap_or(0.0);
+        let end_time_raw: Option<f64> = row.try_get("end_time").ok();
         let data: Value = row.try_get("data").unwrap_or(Value::Null);
 
+        let start_frame = (start_time * fps) as i64;
+        let end_frame = end_time_raw
+            .filter(|t| *t > 0.0)
+            .map(|t| (t * fps) as i64)
+            .unwrap_or(start_frame + 1);
+
         let texts: Vec<String> = data
             .get("texts")
             .and_then(|t| t.as_array())
@@ -195,7 +250,16 @@ async fn fetch_ocr_texts(
             })
             .unwrap_or_default();
 
-        map.insert(frame, texts);
+        // Store texts for each frame in the range
+        for frame in start_frame..=end_frame {
+            map.entry(frame).or_default().extend(texts.clone());
+        }
+    }
+
+    // Deduplicate texts per frame
+    for (_frame, texts) in map.iter_mut() {
+        let mut seen = std::collections::HashSet::new();
+        texts.retain(|t| seen.insert(t.clone()));
     }
 
     Ok(map)
@@ -223,3 +287,62 @@ fn collect_ocr_text(
 
     parts.join(" ")
 }
+
+/// Group ALL OCR frames by proximity into chunks
+/// Returns vec of (start_frame, end_frame, combined_ocr_text)
+fn group_ocr_frames(
+    ocr_map: &BTreeMap<i64, Vec<String>>,
+    _fps: f64,
+) -> Vec<(i64, i64, String)> {
+    const MAX_FRAME_GAP: i64 = 5; // ~0.2s at 24fps
+
+    let mut result = Vec::new();
+    let mut start_frame: Option<i64> = None;
+    let mut end_frame: Option<i64> = None;
+    let mut current_texts: Vec<String> = Vec::new();
+
+    for (frame, texts) in ocr_map.iter() {
+        if start_frame.is_none() {
+            // Start first group
+            start_frame = Some(*frame);
+            end_frame = Some(*frame);
+            current_texts = texts.clone();
+        } else {
+            let gap = *frame - end_frame.unwrap();
+            if gap <= MAX_FRAME_GAP {
+                // Continue current group
+                end_frame = Some(*frame);
+                current_texts.extend(texts.clone());
+            } else {
+                // Save current group and start new one
+                if !current_texts.is_empty() {
+                    let mut seen = std::collections::HashSet::new();
+                    let unique: Vec<String> = current_texts
+                        .iter()
+                        .filter(|t| seen.insert((*t).clone()))
+                        .cloned()
+                        .collect();
+                    result.push((start_frame.unwrap(), end_frame.unwrap(), unique.join(" ")));
+                }
+                start_frame = Some(*frame);
+                end_frame = Some(*frame);
+                current_texts = texts.clone();
+            }
+        }
+    }
+
+    // Don't forget the last group
+    if let (Some(start), Some(end)) = (start_frame, end_frame) {
+        if !current_texts.is_empty() {
+            let mut seen = std::collections::HashSet::new();
+            let unique: Vec<String> = current_texts
+                .iter()
+                .filter(|t| seen.insert((*t).clone()))
+                .cloned()
+                .collect();
+            result.push((start, end, unique.join(" ")));
+        }
+    }
+
+    result
+}

src/core/db/postgres_db.rs

@@ -832,6 +832,7 @@ pub struct SemanticSearchResult {
     pub start_time: f64,
     pub end_time: f64,
     pub summary: String,
+    pub text_content: Option<String>,
     pub metadata: Option<serde_json::Value>,
     pub similarity: Option<f64>,
 }
@@ -1323,6 +1324,29 @@ impl PostgresDb {
             .execute(pool)
             .await?;
 
+        // ── TKG Operation Log ──
+        sqlx::query("CREATE TABLE IF NOT EXISTS tkg_operation_log (
+            id BIGSERIAL PRIMARY KEY,
+            file_uuid VARCHAR(32) NOT NULL,
+            operation VARCHAR(20) NOT NULL,
+            node_type VARCHAR(50),
+            edge_type VARCHAR(50),
+            nodes_created INT DEFAULT 0,
+            nodes_updated INT DEFAULT 0,
+            nodes_deleted INT DEFAULT 0,
+            edges_created INT DEFAULT 0,
+            edges_updated INT DEFAULT 0,
+            edges_deleted INT DEFAULT 0,
+            status VARCHAR(20) DEFAULT 'pending',
+            error_message TEXT,
+            started_at TIMESTAMPTZ DEFAULT NOW(),
+            completed_at TIMESTAMPTZ,
+            properties JSONB
+        )").execute(pool).await?;
+        sqlx::query("CREATE INDEX IF NOT EXISTS idx_tkg_op_file_uuid ON tkg_operation_log(file_uuid)")
+            .execute(pool)
+            .await?;
+
         // ── Functions & Triggers ──
         sqlx::query(
             "CREATE OR REPLACE FUNCTION update_search_vector() RETURNS TRIGGER AS $func$
@@ -2552,6 +2576,7 @@ impl PostgresDb {
                      (start_time * fps)::bigint as start_frame, (end_time * fps)::bigint as end_frame, \
                      fps, start_time, end_time, \
                      COALESCE(summary_text, text_content, '') as summary, \
+                     text_content as text_content, \
                      metadata, \
                      1.0::float8 as similarity \
                  FROM {} \
@@ -2568,6 +2593,37 @@ impl PostgresDb {
         Ok(results)
     }
 
+    /// Get chunk by file_uuid and chunk_id WITHOUT embedding requirement (for keyword search)
+    pub async fn get_chunk_by_id_no_embedding(
+        &self,
+        file_uuid: &str,
+        chunk_id: &str,
+    ) -> Result<Option<SemanticSearchResult>> {
+        let chunk_table = schema::table_name("chunk");
+        let results = sqlx::query_as::<_, SemanticSearchResult>(
+            &format!(
+                "SELECT \
+                     id, file_uuid, id as scene_order, \
+                     (start_time * fps)::bigint as start_frame, (end_time * fps)::bigint as end_frame, \
+                     fps, start_time, end_time, \
+                     COALESCE(summary_text, text_content, '') as summary, \
+                     text_content as text_content, \
+                     metadata, \
+                     1.0::float8 as similarity \
+                 FROM {} \
+                 WHERE file_uuid = $1 AND chunk_id = $2 \
+                 LIMIT 1",
+                chunk_table
+            ),
+        )
+        .bind(file_uuid)
+        .bind(chunk_id)
+        .fetch_optional(&self.pool)
+        .await?;
+
+        Ok(results)
+    }
+
     /// Get children for a list of parent IDs
     pub async fn get_children_for_parents(
         &self,
@@ -3339,8 +3395,30 @@ impl PostgresDb {
         let like = format!("%{}%", query.replace('%', "%%"));
         use sqlx::Row;
 
-        // Use PostgreSQL full-text search with ts_rank for ranking, fallback to ILIKE for recall
+        // Check if query contains CJK characters
-        let sql = format!(
+        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) \
@@ -3354,7 +3432,8 @@ impl PostgresDb {
                  LIMIT $3",
                 table,
                 if file_uuid.is_some() { "AND file_uuid = $4 " } else { "" }
-        );
+            )
+        };
 
         let rows = if let Some(u) = file_uuid {
             sqlx::query(&sql)

src/core/mod.rs

@@ -23,4 +23,5 @@ pub mod text;
 pub mod thumbnail;
 pub mod time;
 pub mod tmdb;
+pub mod tkg;
 pub mod vision;

src/core/pipeline/mod.rs

@@ -29,8 +29,11 @@ pub async fn store_asrx_chunks(db: &PostgresDb, uuid: &str) -> Result<()> {
             "text": segment.text,
             "speaker_id": segment.speaker_id,
             "timestamp": segment.start_time,
+            "end_time": segment.end_time,
+            "start_frame": segment.start_frame,
+            "end_frame": segment.end_frame,
         });
-        pre_chunks.push((i as i64, Some(segment.start_time), data, None, None));
+        pre_chunks.push((segment.start_frame as i64, Some(segment.start_time), data, None, None));
         speaker_detections.push((
             segment.speaker_id.clone().unwrap_or_default(),
             segment.start_time,

src/core/processor/executor.rs

@@ -309,6 +309,13 @@ impl PythonExecutor {
         cmd.env("DATABASE_SCHEMA", &*DATABASE_SCHEMA);
         cmd.env("MOMENTRY_DB_SCHEMA", &*DATABASE_SCHEMA);
         cmd.env("MOMENTRY_REDIS_PREFIX", &*REDIS_KEY_PREFIX);
+        // Pass Qdrant credentials to Python subprocesses
+        if let Ok(qdrant_url) = std::env::var("QDRANT_URL") {
+            cmd.env("QDRANT_URL", qdrant_url);
+        }
+        if let Ok(qdrant_api_key) = std::env::var("QDRANT_API_KEY") {
+            cmd.env("QDRANT_API_KEY", qdrant_api_key);
+        }
         if let Some(u) = uuid {
             cmd.env("UUID", u);
         }
@@ -450,6 +457,13 @@ impl PythonExecutor {
         cmd.env("DATABASE_SCHEMA", &*DATABASE_SCHEMA);
         cmd.env("MOMENTRY_DB_SCHEMA", &*DATABASE_SCHEMA);
         cmd.env("MOMENTRY_REDIS_PREFIX", &*REDIS_KEY_PREFIX);
+        // Pass Qdrant credentials to Python subprocesses
+        if let Ok(qdrant_url) = std::env::var("QDRANT_URL") {
+            cmd.env("QDRANT_URL", qdrant_url);
+        }
+        if let Ok(qdrant_api_key) = std::env::var("QDRANT_API_KEY") {
+            cmd.env("QDRANT_API_KEY", qdrant_api_key);
+        }
         if let Some(u) = uuid {
             cmd.env("UUID", u);
         }
@@ -631,6 +645,13 @@ impl PythonExecutor {
         cmd.env("DATABASE_SCHEMA", &*DATABASE_SCHEMA);
         cmd.env("MOMENTRY_DB_SCHEMA", &*DATABASE_SCHEMA);
         cmd.env("MOMENTRY_REDIS_PREFIX", &*REDIS_KEY_PREFIX);
+        // Pass Qdrant credentials to Python subprocesses
+        if let Ok(qdrant_url) = std::env::var("QDRANT_URL") {
+            cmd.env("QDRANT_URL", qdrant_url);
+        }
+        if let Ok(qdrant_api_key) = std::env::var("QDRANT_API_KEY") {
+            cmd.env("QDRANT_API_KEY", qdrant_api_key);
+        }
         cmd.arg(&script_path);
 
         for arg in args {
@@ -882,6 +903,13 @@ mod tests {
         cmd.env("DATABASE_SCHEMA", &*DATABASE_SCHEMA);
         cmd.env("MOMENTRY_DB_SCHEMA", &*DATABASE_SCHEMA);
         cmd.env("MOMENTRY_REDIS_PREFIX", &*REDIS_KEY_PREFIX);
+        // Pass Qdrant credentials to Python subprocesses
+        if let Ok(qdrant_url) = std::env::var("QDRANT_URL") {
+            cmd.env("QDRANT_URL", qdrant_url);
+        }
+        if let Ok(qdrant_api_key) = std::env::var("QDRANT_API_KEY") {
+            cmd.env("QDRANT_API_KEY", qdrant_api_key);
+        }
         cmd.args([
             "-c",
             "import os; print(f'ENV_DATABASE_SCHEMA={os.environ.get(\"DATABASE_SCHEMA\",\"\")}'); print(f'ENV_MOMENTRY_DB_SCHEMA={os.environ.get(\"MOMENTRY_DB_SCHEMA\",\"\")}'); print(f'ENV_MOMENTRY_OUTPUT_DIR={os.environ.get(\"MOMENTRY_OUTPUT_DIR\",\"\")}'); print(f'ENV_MOMENTRY_REDIS_PREFIX={os.environ.get(\"MOMENTRY_REDIS_PREFIX\",\"\")}');",

src/core/processor/tkg.rs

@@ -1401,8 +1401,11 @@ async fn build_speaker_face_edges_from_pg(
         .collect();
 
     let last = asrx.segments.last().unwrap();
-    let fps = if last.end_time > 0.0 {
+    let fps = if last.end_frame > 0 && last.end_time > 0.0 {
         last.end_frame as f64 / last.end_time
+    } else if last.end_time > 0.0 {
+        // Fallback: estimate FPS from probe or use 30
+        30.0
     } else {
         30.0
     };

src/core/progress.rs

@@ -518,7 +518,7 @@ pub async fn get_progress(
     }
 }
 
-/// Publish pipeline progress to Redis
+/// Publish pipeline progress to Redis (accumulates with existing progress)
 pub async fn publish_pipeline_progress(
     redis: &RedisClient,
     file_uuid: &str,
@@ -530,7 +530,32 @@ pub async fn publish_pipeline_progress(
         file_uuid
     );
     if let Ok(mut conn) = redis.get_conn().await {
-        let json = serde_json::to_string(progress).unwrap_or_default();
+        // Try to read existing progress first
+        let existing: Option<PipelineProgress> = redis::cmd("GET")
+            .arg(&key)
+            .query_async(&mut conn)
+            .await
+            .ok()
+            .and_then(|s: String| serde_json::from_str(&s).ok());
+
+        let merged = if let Some(mut existing) = existing {
+            // Merge: update stages from new progress onto existing
+            for new_stage in &progress.stages {
+                if new_stage.status == "completed" || new_stage.progress > 0.0 {
+                    if let Some(existing_stage) = existing.stages.iter_mut().find(|s| s.name == new_stage.name) {
+                        existing_stage.status = new_stage.status.clone();
+                        existing_stage.progress = new_stage.progress;
+                        existing_stage.detail = new_stage.detail.clone();
+                    }
+                }
+            }
+            existing.recalculate_overall();
+            existing
+        } else {
+            progress.clone()
+        };
+
+        let json = serde_json::to_string(&merged).unwrap_or_default();
         let _: Result<(), _> = redis::cmd("SET")
             .arg(&[&key, &json])
             .query_async(&mut conn)

src/core/tkg/log.rs

@@ -0,0 +1,119 @@
+use crate::core::db::schema;
+use crate::core::db::PostgresDb;
+use anyhow::Result;
+use sqlx::PgPool;
+
+pub struct TkgLogger {
+    pool: PgPool,
+    pub log_id: Option<i64>,
+}
+
+impl TkgLogger {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool, log_id: None }
+    }
+
+    pub async fn start_operation(pool: &PgPool, file_uuid: &str, operation: &str) -> Result<i64> {
+        let table = schema::table_name("tkg_operation_log");
+        let id = sqlx::query_scalar::<_, i64>(&format!(
+            "INSERT INTO {table} (file_uuid, operation, status, started_at) \
+             VALUES ($1, $2, 'pending', NOW()) RETURNING id"
+        ))
+        .bind(file_uuid)
+        .bind(operation)
+        .fetch_one(pool)
+        .await?;
+        tracing::info!("[TKG-Log] Started operation {} for {}", operation, file_uuid);
+        Ok(id)
+    }
+
+    pub async fn update_progress(
+        &self,
+        node_type: &str,
+        nodes_created: i32,
+        edges_created: i32,
+    ) -> Result<()> {
+        if let Some(log_id) = self.log_id {
+            let table = schema::table_name("tkg_operation_log");
+            sqlx::query(&format!(
+                "UPDATE {table} \
+                 SET nodes_created = nodes_created + $1, \
+                     edges_created = edges_created + $2, \
+                     node_type = COALESCE(node_type, $3) \
+                 WHERE id = $4"
+            ))
+            .bind(nodes_created)
+            .bind(edges_created)
+            .bind(node_type)
+            .bind(log_id)
+            .execute(&self.pool)
+            .await?;
+        }
+        Ok(())
+    }
+
+    pub async fn complete_operation(&self, error: Option<&str>) -> Result<()> {
+        if let Some(log_id) = self.log_id {
+            let table = schema::table_name("tkg_operation_log");
+            let status = if error.is_some() { "failed" } else { "completed" };
+            sqlx::query(&format!(
+                "UPDATE {table} \
+                 SET status = $1, \
+                     error_message = $2, \
+                     completed_at = NOW() \
+                 WHERE id = $3"
+            ))
+            .bind(status)
+            .bind(error)
+            .bind(log_id)
+            .execute(&self.pool)
+            .await?;
+            tracing::info!("[TKG-Log] Operation {} completed with status: {}", log_id, status);
+        }
+        Ok(())
+    }
+
+    pub async fn get_operations(pool: &PgPool, file_uuid: &str) -> Result<Vec<crate::core::tkg::models::TkgOperationLog>> {
+        let table = schema::table_name("tkg_operation_log");
+        let rows = sqlx::query_as::<_, (i64, String, String, Option<String>, Option<String>, i32, i32, i32, i32, i32, i32, String, Option<String>, String, Option<String>, Option<serde_json::Value>)>(&format!(
+            "SELECT id, file_uuid, operation, node_type, edge_type, \
+                    nodes_created, nodes_updated, nodes_deleted, \
+                    edges_created, edges_updated, edges_deleted, \
+                    status, error_message, started_at::text, completed_at::text, properties \
+             FROM {table} WHERE file_uuid = $1 ORDER BY started_at DESC"
+        ))
+        .bind(file_uuid)
+        .fetch_all(pool)
+        .await?;
+
+        Ok(rows.into_iter().map(|r| crate::core::tkg::models::TkgOperationLog {
+            id: r.0,
+            file_uuid: r.1,
+            operation: r.2,
+            node_type: r.3,
+            edge_type: r.4,
+            nodes_created: r.5,
+            nodes_updated: r.6,
+            nodes_deleted: r.7,
+            edges_created: r.8,
+            edges_updated: r.9,
+            edges_deleted: r.10,
+            status: r.11,
+            error_message: r.12,
+            started_at: r.13,
+            completed_at: r.14,
+            properties: r.15,
+        }).collect())
+    }
+
+    pub async fn delete_operations(pool: &PgPool, file_uuid: &str) -> Result<i64> {
+        let table = schema::table_name("tkg_operation_log");
+        let result = sqlx::query(&format!(
+            "DELETE FROM {table} WHERE file_uuid = $1"
+        ))
+        .bind(file_uuid)
+        .execute(pool)
+        .await?;
+        Ok(result.rows_affected() as i64)
+    }
+}

src/core/tkg/mod.rs

@@ -0,0 +1,7 @@
+pub mod service;
+pub mod log;
+pub mod models;
+
+pub use service::TkgService;
+pub use log::TkgLogger;
+pub use models::*;

src/core/tkg/models.rs

@@ -0,0 +1,94 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct TkgOperationLog {
+    pub id: i64,
+    pub file_uuid: String,
+    pub operation: String,
+    pub node_type: Option<String>,
+    pub edge_type: Option<String>,
+    pub nodes_created: i32,
+    pub nodes_updated: i32,
+    pub nodes_deleted: i32,
+    pub edges_created: i32,
+    pub edges_updated: i32,
+    pub edges_deleted: i32,
+    pub status: String,
+    pub error_message: Option<String>,
+    pub started_at: String,
+    pub completed_at: Option<String>,
+    pub properties: Option<serde_json::Value>,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct TkgNodeStats {
+    pub created: i32,
+    pub updated: i32,
+    pub deleted: i32,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct TkgEdgeStats {
+    pub created: i32,
+    pub updated: i32,
+    pub deleted: i32,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct TkgBuildStats {
+    pub face_track_nodes: i32,
+    pub gaze_track_nodes: i32,
+    pub lip_track_nodes: i32,
+    pub text_region_nodes: i32,
+    pub appearance_trace_nodes: i32,
+    pub accessory_nodes: i32,
+    pub object_nodes: i32,
+    pub hand_nodes: i32,
+    pub speaker_nodes: i32,
+    pub co_occurrence_edges: i32,
+    pub speaker_face_edges: i32,
+    pub face_face_edges: i32,
+    pub mutual_gaze_edges: i32,
+    pub lip_sync_edges: i32,
+    pub has_appearance_edges: i32,
+    pub wears_edges: i32,
+    pub hand_object_edges: i32,
+}
+
+impl TkgBuildStats {
+    pub fn total_nodes(&self) -> i32 {
+        self.face_track_nodes + self.gaze_track_nodes + self.lip_track_nodes
+            + self.text_region_nodes + self.appearance_trace_nodes + self.accessory_nodes
+            + self.object_nodes + self.hand_nodes + self.speaker_nodes
+    }
+
+    pub fn total_edges(&self) -> i32 {
+        self.co_occurrence_edges + self.speaker_face_edges + self.face_face_edges
+            + self.mutual_gaze_edges + self.lip_sync_edges + self.has_appearance_edges
+            + self.wears_edges + self.hand_object_edges
+    }
+
+    pub fn to_json(&self) -> serde_json::Value {
+        serde_json::json!({
+            "face_track_nodes": self.face_track_nodes,
+            "gaze_track_nodes": self.gaze_track_nodes,
+            "lip_track_nodes": self.lip_track_nodes,
+            "text_region_nodes": self.text_region_nodes,
+            "appearance_trace_nodes": self.appearance_trace_nodes,
+            "accessory_nodes": self.accessory_nodes,
+            "object_nodes": self.object_nodes,
+            "hand_nodes": self.hand_nodes,
+            "speaker_nodes": self.speaker_nodes,
+            "co_occurrence_edges": self.co_occurrence_edges,
+            "speaker_face_edges": self.speaker_face_edges,
+            "face_face_edges": self.face_face_edges,
+            "mutual_gaze_edges": self.mutual_gaze_edges,
+            "lip_sync_edges": self.lip_sync_edges,
+            "has_appearance_edges": self.has_appearance_edges,
+            "wears_edges": self.wears_edges,
+            "hand_object_edges": self.hand_object_edges,
+            "total_nodes": self.total_nodes(),
+            "total_edges": self.total_edges(),
+        })
+    }
+}

src/core/tkg/service.rs

@@ -0,0 +1,140 @@
+use crate::core::db::PostgresDb;
+use crate::core::db::schema;
+use crate::core::tkg::log::TkgLogger;
+use crate::core::tkg::models::{TkgBuildStats, TkgOperationLog};
+use anyhow::Result;
+use sqlx::PgPool;
+use std::sync::Arc;
+
+pub struct TkgService {
+    db: Arc<PostgresDb>,
+}
+
+impl TkgService {
+    pub fn new(db: Arc<PostgresDb>) -> Self {
+        Self { db }
+    }
+
+    pub async fn build(&self, file_uuid: &str, output_dir: &str) -> Result<TkgBuildStats> {
+        let log_id = TkgLogger::start_operation(self.db.pool(), file_uuid, "build").await?;
+        let mut logger = TkgLogger::new(self.db.pool().clone());
+        logger.log_id = Some(log_id);
+
+        let redis = crate::core::db::RedisClient::new().ok().map(|r| std::sync::Arc::new(r));
+        let result = crate::core::processor::tkg::build_tkg(&self.db, file_uuid, output_dir, redis).await;
+
+        match result {
+            Ok(r) => {
+                let stats = TkgBuildStats {
+                    face_track_nodes: r.face_track_nodes as i32,
+                    gaze_track_nodes: r.gaze_track_nodes as i32,
+                    lip_track_nodes: r.lip_track_nodes as i32,
+                    text_region_nodes: r.text_region_nodes as i32,
+                    appearance_trace_nodes: r.appearance_trace_nodes as i32,
+                    accessory_nodes: r.accessory_nodes as i32,
+                    object_nodes: r.object_nodes as i32,
+                    hand_nodes: r.hand_nodes as i32,
+                    speaker_nodes: r.speaker_nodes as i32,
+                    co_occurrence_edges: r.co_occurrence_edges as i32,
+                    speaker_face_edges: r.speaker_face_edges as i32,
+                    face_face_edges: r.face_face_edges as i32,
+                    mutual_gaze_edges: r.mutual_gaze_edges as i32,
+                    lip_sync_edges: r.lip_sync_edges as i32,
+                    has_appearance_edges: r.has_appearance_edges as i32,
+                    wears_edges: r.wears_edges as i32,
+                    hand_object_edges: r.hand_object_edges as i32,
+                };
+                logger.complete_operation(None).await?;
+                tracing::info!("[TKG-Service] Build completed for {}: {} nodes, {} edges",
+                    file_uuid, stats.total_nodes(), stats.total_edges());
+                Ok(stats)
+            }
+            Err(e) => {
+                logger.complete_operation(Some(&e.to_string())).await?;
+                tracing::error!("[TKG-Service] Build failed for {}: {}", file_uuid, e);
+                Err(e)
+            }
+        }
+    }
+
+    pub async fn rebuild(&self, file_uuid: &str, output_dir: &str, force: bool) -> Result<TkgBuildStats> {
+        let operation = if force { "rebuild_force" } else { "rebuild" };
+        let log_id = TkgLogger::start_operation(self.db.pool(), file_uuid, operation).await?;
+
+        if force {
+            self.delete_internal(file_uuid).await?;
+        }
+
+        let result = self.build(file_uuid, output_dir).await;
+
+        // Update the original log entry
+        if let Some(id) = Some(log_id) {
+            let table = schema::table_name("tkg_operation_log");
+            let status = if result.is_ok() { "completed" } else { "failed" };
+            let error = result.as_ref().err().map(|e| e.to_string());
+            sqlx::query(&format!(
+                "UPDATE {table} SET status = $1, error_message = $2, completed_at = NOW() WHERE id = $3"
+            ))
+            .bind(status)
+            .bind(error)
+            .bind(id)
+            .execute(self.db.pool())
+            .await?;
+        }
+
+        result
+    }
+
+    pub async fn delete(&self, file_uuid: &str) -> Result<()> {
+        let log_id = TkgLogger::start_operation(self.db.pool(), file_uuid, "delete").await?;
+
+        let result = self.delete_internal(file_uuid).await;
+
+        match result {
+            Ok(_) => {
+                TkgLogger::new(self.db.pool().clone()).complete_operation(None).await?;
+                Ok(())
+            }
+            Err(e) => {
+                TkgLogger::new(self.db.pool().clone()).complete_operation(Some(&e.to_string())).await?;
+                Err(e)
+            }
+        }
+    }
+
+    async fn delete_internal(&self, file_uuid: &str) -> Result<()> {
+        let nodes_table = schema::table_name("tkg_nodes");
+        let edges_table = schema::table_name("tkg_edges");
+
+        // Delete edges first (foreign key constraint)
+        let edges_deleted = sqlx::query(&format!(
+            "DELETE FROM {edges_table} WHERE file_uuid = $1"
+        ))
+        .bind(file_uuid)
+        .execute(self.db.pool())
+        .await?;
+
+        // Delete nodes
+        let nodes_deleted = sqlx::query(&format!(
+            "DELETE FROM {nodes_table} WHERE file_uuid = $1"
+        ))
+        .bind(file_uuid)
+        .execute(self.db.pool())
+        .await?;
+
+        tracing::info!("[TKG-Service] Deleted {} nodes and {} edges for {}",
+            nodes_deleted.rows_affected(), edges_deleted.rows_affected(), file_uuid);
+
+        Ok(())
+    }
+
+    pub async fn get_operations(&self, file_uuid: &str) -> Result<Vec<TkgOperationLog>> {
+        TkgLogger::get_operations(self.db.pool(), file_uuid).await
+    }
+
+    pub async fn delete_tkg_and_logs(&self, file_uuid: &str) -> Result<()> {
+        self.delete_internal(file_uuid).await?;
+        TkgLogger::delete_operations(self.db.pool(), file_uuid).await?;
+        Ok(())
+    }
+}

src/worker/job_worker.rs

@@ -278,6 +278,13 @@ impl JobWorker {
     }
 
     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?;
+        if current_job.is_none() {
+            info!("Job {} no longer exists in database (possibly unregistered), skipping", job.uuid);
+            return Ok(());
+        }
+
         info!("Processing job: {} ({})", job.uuid, job.id);
 
         // Determine which processors to run based on job.processors field
@@ -305,6 +312,12 @@ impl JobWorker {
             .init_processing_status(&job.uuid, processor_names.clone(), total_frames)
             .await?;
 
+        // Clear any stale PipelineProgress from previous jobs
+        let progress_key = format!("{}progress:{}:pipeline", crate::core::config::REDIS_KEY_PREFIX.as_str(), job.uuid);
+        if let Ok(mut conn) = self.redis.get_conn().await {
+            let _: Option<String> = redis::cmd("DEL").arg(&progress_key).query_async(&mut conn).await.ok();
+        }
+
         self.db
             .update_job_status(job.id, MonitorJobStatus::Running)
             .await?;
@@ -361,6 +374,58 @@ impl JobWorker {
             ));
             debug!("Checking output file: {:?}", output_path);
 
+            // Check for stale .tmp file (process crashed before renaming)
+            let tmp_path = PathBuf::from(OUTPUT_DIR.as_str()).join(format!(
+                "{}.{}.json.tmp",
+                job.uuid,
+                processor_type.as_str()
+            ));
+            if tmp_path.exists() && !output_path.exists() {
+                if let Ok(meta) = std::fs::metadata(&tmp_path) {
+                    // 條件 1: 檔案 > 1KB
+                    let has_content = meta.len() > 1024;
+                    
+                    // 條件 2: 檔案超過 120 秒未修改（確定沒人還在寫）
+                    let is_stale = if let Ok(modified) = meta.modified() {
+                        if let Ok(elapsed) = modified.elapsed() {
+                            elapsed.as_secs() > 120
+                        } else { false }
+                    } else { false };
+                    
+                    // 條件 3: 檢查程序是否還在跑
+                    let proc_name = processor_type.as_str();
+                    let process_running = std::process::Command::new("ps")
+                        .args(["aux"])
+                        .output()
+                        .ok()
+                        .and_then(|out| String::from_utf8(out.stdout).ok())
+                        .map(|out| {
+                            out.contains(&format!("{}_processor", proc_name)) || 
+                            out.contains(&format!("{}_processor", proc_name))
+                        })
+                        .unwrap_or(false);
+                    
+                    if has_content && is_stale && !process_running {
+                        info!(
+                            "Found stale .tmp file ({} bytes, {}s old, process={}), renaming to .json for {}",
+                            meta.len(),
+                            meta.modified().ok().and_then(|m| m.elapsed().ok()).map(|e| e.as_secs()).unwrap_or(0),
+                            if process_running { "running" } else { "dead" },
+                            processor_type.as_str()
+                        );
+                        if std::fs::rename(&tmp_path, &output_path).is_ok() {
+                            info!("Recovered {} from .tmp file", processor_type.as_str());
+                        }
+                    } else if !has_content {
+                        debug!("Skipping .tmp file (too small): {} bytes", meta.len());
+                    } else if !is_stale {
+                        debug!("Skipping .tmp file (recently modified)");
+                    } else if process_running {
+                        debug!("Skipping .tmp file (process still running)");
+                    }
+                }
+            }
+
             // Special case: Pose processor should NOT be skipped even if pose.json exists
             // because swift_face_pose creates it and pose.rs needs to interpolate
             let skip_check = if *processor_type == crate::core::db::ProcessorType::Pose {
@@ -475,19 +540,6 @@ impl JobWorker {
                                                 Some(&segment.text),
                                             )
                                             .await;
-                                        // Also store asr pre_chunks (needed by Rule 1 after checkin)
-                                        let _ = ws
-                                            .store_pre_chunk(
-                                                "asr",
-                                                "raw",
-                                                None,
-                                                None,
-                                                Some(segment.start_time),
-                                                Some(segment.end_time),
-                                                Some(&data.to_string()),
-                                                Some(&segment.text),
-                                            )
-                                            .await;
                                     }
                                     let spk_dets: Vec<crate::core::db::workspace_sqlite::SpeakerDetectionBatchItem> = result.segments.iter().map(|s| {
                                         crate::core::db::workspace_sqlite::SpeakerDetectionBatchItem {
@@ -1096,9 +1148,17 @@ impl JobWorker {
         let has_face = job_processors.is_empty() || job_processors.iter().any(|p| p == "face");
 
         // Check asr_status for ASR/ASRX - if no_audio_track or silent_audio, ingestion is complete
+        let mj_t = schema::table_name("monitor_jobs");
         let asr_done: bool = if has_asr_or_asrx {
+            // Query ASRX first (more authoritative), then ASR
+            // Filter out NULL values to ensure we get a valid status
             let asr_status: Option<String> = sqlx::query_scalar(&format!(
-                "SELECT asr_status FROM {pr_t} WHERE file_uuid = $1 AND processor IN ('asr', 'asrx') LIMIT 1"
+                "SELECT asr_status FROM {pr_t} pr JOIN {mj_t} mj ON pr.job_id = mj.id \
+                 WHERE mj.uuid = $1 AND pr.processor = 'asrx' AND asr_status IS NOT NULL \
+                 UNION ALL \
+                 SELECT asr_status FROM {pr_t} pr JOIN {mj_t} mj ON pr.job_id = mj.id \
+                 WHERE mj.uuid = $1 AND pr.processor = 'asr' AND asr_status IS NOT NULL \
+                 LIMIT 1"
             ))
             .bind(uuid)
             .fetch_optional(pool)
@@ -1148,6 +1208,17 @@ impl JobWorker {
             if std::path::Path::new(&traced_path).exists() {
                 if let Ok(content) = std::fs::read_to_string(&traced_path) {
                     if let Ok(traced_data) = serde_json::from_str::<serde_json::Value>(&content) {
+                        // Check if status is no_faces (valid completion with no faces)
+                        if let Some(status) = traced_data.get("status").and_then(|s| s.as_str()) {
+                            if status == "no_faces" {
+                                tracing::info!(
+                                    "[Ingestion] No faces detected for {} - trace_done=true",
+                                    uuid
+                                );
+                                return true;
+                            }
+                        }
+
                         if let Some(traces) = traced_data.get("traces") {
                             // traces can be an object (dictionary) or array
                             let trace_count = if traces.is_object() {
@@ -1196,20 +1267,38 @@ impl JobWorker {
             true
         };
 
-        let all_ok = asr_done && trace_done;
+        // Check TKG completion
+        // TKG is considered done if face traces are done (TKG runs after face tracing)
+        // TKG may create 0 nodes/edges for videos with minimal content
+        let has_asr_or_asrx_for_tkg =
+            job_processors.is_empty() || job_processors.iter().any(|p| p == "asrx" || p == "asr");
+        let has_face_for_tkg = job_processors.is_empty() || job_processors.iter().any(|p| p == "face");
+
+        let tkg_done: bool = if has_asr_or_asrx_for_tkg && has_face_for_tkg {
+            // TKG is done if face traces are complete (TKG runs after face tracing)
+            // TKG may create 0 nodes/edges for videos with minimal content
+            trace_done
+        } else {
+            tracing::info!("[Ingestion] No TKG needed for {}", uuid);
+            true
+        };
+
+        let all_ok = asr_done && trace_done && tkg_done;
         tracing::info!(
-            "[Ingestion] all_ok={} (asr_done={}, trace_done={}) for uuid={}",
+            "[Ingestion] all_ok={} (asr_done={}, trace_done={}, tkg_done={}) for uuid={}",
             all_ok,
             asr_done,
             trace_done,
+            tkg_done,
             uuid
         );
         if !all_ok {
             tracing::info!(
-                "[Ingestion] waiting (uuid={}): asr_done={} trace_done={}",
+                "[Ingestion] waiting (uuid={}): asr_done={} trace_done={} tkg_done={}",
                 uuid,
                 asr_done,
-                trace_done
+                trace_done,
+                tkg_done
             );
         }
         all_ok
@@ -1237,6 +1326,13 @@ impl JobWorker {
         // 定義必要 processor（必須完成的才算 job 成功）
         let essential_processors = ["cut", "asr", "asrx", "yolo"];
 
+        let essential_failed = essential_processors.iter().any(|ep| {
+            results.iter().any(|r| {
+                r.processor_type.as_str() == *ep
+                    && matches!(r.status, crate::core::db::ProcessorJobStatus::Failed)
+            })
+        });
+
         let essential_completed = essential_processors.iter().all(|ep| {
             results.iter().any(|r| {
                 r.processor_type.as_str() == *ep
@@ -1244,20 +1340,36 @@ impl JobWorker {
             })
         });
 
-        let all_completed = results
+        let all_completed = job_processors.iter().all(|p| {
-            .iter()
+            results.iter().any(|r| {
-            .filter(|r| job_processors.contains(&r.processor_type.as_str().to_string()))
+                r.processor_type.as_str() == p
-            .all(|r| matches!(r.status, crate::core::db::ProcessorJobStatus::Completed));
+                    && matches!(r.status, crate::core::db::ProcessorJobStatus::Completed)
+            })
+        });
 
         let any_failed = results
             .iter()
             .filter(|r| job_processors.contains(&r.processor_type.as_str().to_string()))
             .any(|r| matches!(r.status, crate::core::db::ProcessorJobStatus::Failed));
 
-        let any_pending = results
+        let any_pending = job_processors.iter().any(|p| {
-            .iter()
+            results.iter().any(|r| {
-            .filter(|r| job_processors.contains(&r.processor_type.as_str().to_string()))
+                r.processor_type.as_str() == p
-            .any(|r| matches!(r.status, crate::core::db::ProcessorJobStatus::Pending));
+                    && matches!(r.status, crate::core::db::ProcessorJobStatus::Pending)
+            })
+        });
+
+        // Check for missing processors (in job_processors but not in results)
+        let missing_processors: Vec<String> = job_processors.iter().filter(|p| {
+            !results.iter().any(|r| r.processor_type.as_str() == *p)
+        }).cloned().collect();
+
+        if !missing_processors.is_empty() {
+            info!(
+                "check_and_complete_job: {} missing processor results: {:?}",
+                uuid, missing_processors
+            );
+        }
 
         const MAX_RETRIES: i32 = 3;
 
@@ -1436,6 +1548,7 @@ impl JobWorker {
         let has_asr_or_asrx = completed_processors
             .iter()
             .any(|p| p == "asrx" || p == "asr");
+        let has_asrx = completed_processors.iter().any(|p| p == "asrx");
         let has_cut = completed_processors.iter().any(|p| p == "cut");
         let has_face = completed_processors.iter().any(|p| p == "face");
         let has_yolo = completed_processors.iter().any(|p| p == "yolo");
@@ -1444,7 +1557,7 @@ impl JobWorker {
             .update_job_processors_arrays(job_id, completed_processors, failed_processors.clone())
             .await?;
 
-        if has_asr_or_asrx {
+        if has_asrx {
             // Guard: only spawn Rule 1 if sentence chunks don't exist yet
             let chunk_t = schema::table_name("chunk");
             let already_spawned: bool = sqlx::query_scalar::<_, i32>(&format!(
@@ -1532,7 +1645,7 @@ impl JobWorker {
             }
         }
 
-        if all_completed {
+        if has_face && has_asrx {
             let mut pp = PipelineProgress::new(uuid);
             pp.update_stage("processors", 1.0, "completed", None);
             publish_pipeline_progress(self.redis.as_ref(), uuid, &pp).await;
@@ -1722,9 +1835,32 @@ impl JobWorker {
                 });
             }
 
-            // 🚀 P3 Trigger: Identity Agent (Face + ASRX)
+            // 🚀 P3 Trigger: Identity Agent (Face + ASRX + has seed identities)
-            if has_face && has_asr_or_asrx {
+            if has_face && has_asrx {
-                info!("📝 Prerequisites met for Identity Agent. Starting analysis...");
+                // Check if file has seed identity photos in Qdrant _seeds collection
+                let has_seeds = {
+                    use crate::core::db::qdrant_db::QdrantDb;
+                    let qdrant = QdrantDb::new();
+                    let schema = std::env::var("DATABASE_SCHEMA").unwrap_or_else(|_| "dev".to_string());
+                    let seeds_collection = if schema == "public" {
+                        "momentry_public_seeds"
+                    } else {
+                        &format!("momentry_{}_seeds", schema)
+                    };
+                    let filter = serde_json::json!({
+                        "must": [{"key": "file_uuid", "match": {"value": uuid}}]
+                    });
+                    match qdrant.scroll_all_points("_seeds", filter, 1).await {
+                        Ok(points) => !points.is_empty(),
+                        Err(e) => {
+                            warn!("Failed to check _seeds for {}: {}", uuid, e);
+                            false
+                        }
+                    }
+                };
+
+                if has_seeds {
+                    info!("📝 Prerequisites met for Identity Agent (has seeds). Starting analysis...");
                     let db_clone = self.db.clone();
                     let redis_clone = self.redis.clone();
                     let uuid_clone = uuid.to_string();
@@ -1739,10 +1875,14 @@ impl JobWorker {
                             Err(e) => error!("❌ Identity Agent failed for {}: {}", uuid_clone, e),
                         }
                     });
+                } else {
+                    info!("📝 Skipping Identity Agent for {} (no seed identities)", uuid);
+                }
             }
 
             // 🚀 P4 Trigger: TKG Build (Face + ASRX) → then Rule2 ingestion
-            if has_face && has_asr_or_asrx {
+            // Note: build_tkg uses ON CONFLICT, so it's safe to call multiple times
+            if has_face && has_asrx {
                 info!("📝 Prerequisites met for TKG Build. Starting graph construction...");
                 let db_clone = self.db.clone();
                 let redis_clone = self.redis.clone();
@@ -1840,12 +1980,17 @@ impl JobWorker {
 
             self.redis.delete_worker_job(uuid).await?;
 
+            // Update PipelineProgress to completed
+            let mut pp = PipelineProgress::new(uuid);
+            pp.mark_completed();
+            publish_pipeline_progress(self.redis.as_ref(), uuid, &pp).await;
+
             info!(
                 "Job {} completed with {} non-essential failures",
                 job_id,
                 failed_processors.len()
             );
-        } else if any_failed {
+        } else if essential_failed {
             self.db
                 .update_job_status(job_id, MonitorJobStatus::Failed)
                 .await?;

src/worker/processor.rs

@@ -1501,10 +1501,11 @@ impl ProcessorPool {
                 "speaker_id": segment.speaker_id,
                 "timestamp": segment.start_time,
                 "end_time": segment.end_time,
+                "start_frame": segment.start_frame,
+                "end_frame": segment.end_frame,
             });
 
-            // ASRX is time-based, so we use segment index or start time as coordinate.
+            pre_chunks_to_store.push((segment.start_frame as i64, Some(segment.start_time), data, None, None));
-            pre_chunks_to_store.push((i as i64, Some(segment.start_time), data, None, None));
 
             speaker_detections.push((
                 segment.speaker_id.clone().unwrap_or_default(),