apps/worker/worker_app/tasks/generation.py

@@ -863,6 +863,11 @@ def generate_video(self, task_id: str) -> dict:
             if gen_task:
                 gen_task.append_log("任务失败", "缺少 source_edit_plan_id", level="ERROR")
                 _flush_logs(task_id, gen_task)
+            _update_task_status(
+                task_id,
+                "mark_failed",
+                error_message="source_edit_plan_id is required. Please create a preview task first.",
+            )
             return {
                 "status": "failed",
                 "task_id": task_id,

tests/unit/test_empty_plan_id_marks_failed.py

@@ -0,0 +1,32 @@
+"""回归测试：source_edit_plan_id 为空时任务必须被标记为 failed。
+
+背景 (2026-08-24)：确认生成卡在 10%。根因是 worker 在
+source_edit_plan_id 为空时直接 return failed，但没有调用 mark_failed，
+导致 DB 状态永远停在 running。
+"""
+
+from pathlib import Path
+
+GENERATION_PY = Path(__file__).resolve().parents[2] / "apps" / "worker" / "worker_app" / "tasks" / "generation.py"
+
+
+def test_mark_failed_in_else_branch():
+    """generation.py 中 source_edit_plan_id 为空的 else 分支必须调用 mark_failed。"""
+    source = GENERATION_PY.read_text(encoding="utf-8")
+
+    # 定位 else 分支：紧跟在 'source_edit_plan_id 为空' 日志之后的 else 块
+    marker = "source_edit_plan_id 为空"
+    idx = source.find(marker)
+    assert idx != -1, f"generation.py 中未找到 '{marker}'"
+
+    # 从 marker 位置向后搜索到下一个 return 语句
+    after_marker = source[idx:]
+    return_idx = after_marker.find("return {")
+    assert return_idx != -1, "else 分支中未找到 return 语句"
+
+    # 关键断言：marker 和 return 之间必须包含 mark_failed
+    block = source[idx : idx + return_idx]
+    assert "mark_failed" in block, (
+        "else 分支在 return 之前必须调用 _update_task_status(task_id, "
+        "'mark_failed', ...) 以更新 DB 状态，否则任务永远卡在 running"
+    )