From d75c745b95b102eec4ee498a2257258842c8e2b7 Mon Sep 17 00:00:00 2001
From: ericwyuan <ericwyuan@example.com>
Date: Thu, 20 Aug 2026 11:09:10 +0800
Subject: [PATCH] fix: dispatcher crash on invalid failure_stage + per-task
 error isolation

Two fixes:
1. db_layer.update_task_status: map invalid failure_stage values (e.g.
   'process' from Edge) to 'callback' before DB write, preventing
   MariaDB ENUM DataError (1265 "Data truncated")
2. dispatcher._poll_once: wrap each task dispatch in try/except so one
   task's failure doesn't skip remaining tasks in the batch

Root cause: Edge returns failure_stage='process' but DB ENUM only allows
  download/extract/vlm_visual/vlm_fusion/callback. The DataError crashed
  _poll_once(), causing all subsequent PENDING tasks to be skipped.
---
 fam-core/src/fam_core/db_layer.py              | 3 +++
 fam-core/src/fam_core/dispatcher/dispatcher.py | 5 ++++-
 2 files changed, 7 insertions(+), 1 deletion(-)

diff --git a/fam-core/src/fam_core/db_layer.py b/fam-core/src/fam_core/db_layer.py
index a19671c..1c5f9d0 100644
--- a/fam-core/src/fam_core/db_layer.py
+++ b/fam-core/src/fam_core/db_layer.py
@@ -92,6 +92,9 @@ def get_tasks_by_status(status: str, limit=10) -> List[Dict]:
 def update_task_status(task_id: int, status: str, error_message: str = None,
                        failure_stage: str = None):
     """更新任务状态"""
+    valid_stages = {'download', 'extract', 'vlm_visual', 'vlm_fusion', 'callback'}
+    if failure_stage and failure_stage not in valid_stages:
+        failure_stage = 'callback'
     conn = get_conn()
     try:
         cursor = conn.cursor()
diff --git a/fam-core/src/fam_core/dispatcher/dispatcher.py b/fam-core/src/fam_core/dispatcher/dispatcher.py
index 34ea3b6..143dd59 100644
--- a/fam-core/src/fam_core/dispatcher/dispatcher.py
+++ b/fam-core/src/fam_core/dispatcher/dispatcher.py
@@ -165,7 +165,10 @@ class Dispatcher:
         tasks = db_layer.get_pending_tasks(limit=10)
         for task in tasks:
             if self._should_retry(task):
-                self._dispatch_one(task)
+                try:
+                    self._dispatch_one(task)
+                except Exception as e:
+                    logger.error(f"[task_id={task['task_id']}] dispatch 异常: {e}", exc_info=True)
 
     def _run(self):
         """线程主循环"""