fix(worker): P1 修复任务状态覆盖 + 封面metadata属性名错误 (#1471)

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>

apps/web/src/api/template-editor/editPlans.ts

@@ -50,3 +50,29 @@ export async function getGenerationTaskResults(taskId: string): Promise<Generate
   const response = await apiClient.get(`/generation/tasks/${taskId}/results`)
   return response.data.items || response.data || []
 }
+
+/** ── 草稿 clips 批量更新 ── */
+
+export interface EditPlanClipInput {
+  asset_id: string
+  start_time: number
+  duration: number
+  order: number
+}
+
+/**
+ * 批量替换草稿的 clips（先全删再批量插入）
+ * 后端路由：PUT /templates/{template_id}/editor/clips
+ */
+export async function updateEditPlanClips(
+  templateId: string,
+  clips: EditPlanClipInput[],
+  signal?: AbortSignal,
+): Promise<{ count: number }> {
+  const response = await apiClient.put(
+    `/templates/${templateId}/editor/clips`,
+    { clips },
+    { signal },
+  )
+  return response.data
+}

apps/web/src/api/template-editor/index.ts

@@ -51,11 +51,13 @@ export {
 export {
   getEditPlan,
   updateEditPlan,
+  updateEditPlanClips,
   generateEditPlan,
   getGenerationStatus,
   getEditPlanGenerations,
   getGenerationTaskResults,
 } from "./editPlans"
+export type { EditPlanClipInput } from "./editPlans"
 
 // 片段 CRUD + 批量操作
 export {

apps/web/src/pages/generate/components/GenerateStepContent.tsx

@@ -123,6 +123,10 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
     presetVoices,
   } = props
 
+  /* 当前模板的 segments，传给 Step2 构建 clips */
+  const currentTemplate = userTemplates.find((t) => t.id === selectedTemplate)
+  const templateSegments = currentTemplate?.segments
+
   switch (currentStep) {
     case 1:
       return (
@@ -142,6 +146,7 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
           smartSelectedIds={smartSelectedIds}
           onSmartSelectedIdsChange={onSmartSelectedIdsChange}
           selectedTemplate={selectedTemplate}
+          templateSegments={templateSegments}
         />
       )
     case 3:

apps/web/src/pages/generate/components/Step2MaterialSelect.tsx

@@ -2,6 +2,7 @@
  * Step 2 素材选择组件
  */
 import React from "react"
+import type { TemplateSegment } from "@/api/templates/types"
 import { useStep2Materials } from "../hooks/useStep2Materials"
 import MaterialModeTabs from "./material/MaterialModeTabs"
 import ManualMaterialList from "./material/ManualMaterialList"
@@ -17,6 +18,8 @@ interface Step2MaterialSelectProps {
   onSmartSelectedIdsChange: (ids: string[]) => void
   /** 当前选中的模板/草稿 ID，用于自动保存 */
   selectedTemplate?: string
+  /** 当前模板的 segments（用于构建 clips duration） */
+  templateSegments?: TemplateSegment[]
 }
 
 const Step2MaterialSelect: React.FC<Step2MaterialSelectProps> = (props) => {

apps/web/src/pages/generate/hooks/useStep2Materials.ts

@@ -3,7 +3,11 @@
  * 组合素材库加载 + 智能匹配两个子 Hook
  */
 import { useCallback, useEffect, useRef } from "react"
+import type { AssetItem } from "@/api/assets"
+import type { TemplateSegment } from "@/api/templates/types"
+import { updateEditPlanClips } from "@/api/template-editor"
 import { formatDuration } from "../utils/formatDuration"
+import { buildClipsFromAssets } from "../utils/buildClipsFromAssets"
 import { useMaterialLibrary } from "./step2-materials/useMaterialLibrary"
 import { useSmartMatch } from "./step2-materials/useSmartMatch"
 import { useDraftAutoSave } from "./useDraftAutoSave"
@@ -17,6 +21,8 @@ interface UseStep2MaterialsProps {
   onSmartSelectedIdsChange: (ids: string[]) => void
   /** 当前选中的模板/草稿 ID，用于自动保存 */
   selectedTemplate?: string
+  /** 当前模板的 segments（用于构建 clips duration） */
+  templateSegments?: TemplateSegment[]
 }
 
 export function useStep2Materials({
@@ -27,6 +33,7 @@ export function useStep2Materials({
   smartSelectedIds,
   onSmartSelectedIdsChange,
   selectedTemplate,
+  templateSegments,
 }: UseStep2MaterialsProps) {
   const { libraries, selectedLibraryId, setSelectedLibraryId, materials, materialsLoading } =
     useMaterialLibrary()
@@ -57,7 +64,7 @@ export function useStep2Materials({
     handleSmartMatch()
   }, [selectedLibraryId, materialMode, materialsLoading, materials.items, handleSmartMatch])
 
-  /* ── Step2 选择素材后自动保存草稿（防抖 500ms，失败静默） ── */
+  /* ── Step2 选择素材后自动保存草稿 asset_ids（防抖 500ms，失败静默） ── */
   const { scheduleSave } = useDraftAutoSave(selectedTemplate)
   useEffect(() => {
     if (!selectedTemplate) return
@@ -65,6 +72,61 @@ export function useStep2Materials({
     scheduleSave({ asset_ids: ids }, 500)
   }, [selectedTemplate, materialMode, selectedMaterials, smartSelectedIds, scheduleSave])
 
+  /* ── Step2 选择素材后同步写入 edit_plan_clips（防抖 800ms，失败静默） ── */
+  const clipsTimerRef = useRef<ReturnType<typeof setTimeout>>()
+  const clipsAbortRef = useRef<AbortController | null>(null)
+  const templateSegmentsRef = useRef(templateSegments)
+  templateSegmentsRef.current = templateSegments
+  const selectedTemplateRef = useRef(selectedTemplate)
+  selectedTemplateRef.current = selectedTemplate
+  const materialsRef = useRef(materials)
+  materialsRef.current = materials
+  const smartMatchedRef = useRef<AssetItem[]>(smartMatch.smartMatchedResults)
+  smartMatchedRef.current = smartMatch.smartMatchedResults
+
+  useEffect(() => {
+    const tid = selectedTemplateRef.current
+    if (!tid) return
+    const ids = materialMode === "auto" ? smartSelectedIds : selectedMaterials
+    if (!ids.length) return
+
+    if (clipsTimerRef.current) clearTimeout(clipsTimerRef.current)
+    clipsTimerRef.current = setTimeout(async () => {
+      // 取消上一次未完成的请求
+      if (clipsAbortRef.current) clipsAbortRef.current.abort()
+      const controller = new AbortController()
+      clipsAbortRef.current = controller
+
+      const clips = buildClipsFromAssets({
+        selectedIds: ids,
+        materials: materialsRef.current.items,
+        smartMatchedAssets: smartMatchedRef.current,
+        templateSegments: templateSegmentsRef.current || [],
+      })
+
+      try {
+        await updateEditPlanClips(tid, clips, controller.signal)
+      } catch (err) {
+        const name = (err as { name?: string })?.name
+        if (name !== "CanceledError" && name !== "AbortError") {
+          console.warn("[useStep2Materials] 写入 clips 失败:", err)
+        }
+      }
+    }, 800)
+
+    return () => {
+      if (clipsTimerRef.current) clearTimeout(clipsTimerRef.current)
+    }
+  }, [selectedTemplate, materialMode, selectedMaterials, smartSelectedIds, templateSegments])
+
+  // 组件卸载时取消未完成请求
+  useEffect(() => {
+    return () => {
+      if (clipsTimerRef.current) clearTimeout(clipsTimerRef.current)
+      if (clipsAbortRef.current) clipsAbortRef.current.abort()
+    }
+  }, [])
+
   /* ── 手动选择素材 ── */
   const handleToggleMaterial = useCallback(
     (materialId: string) => {

apps/web/src/pages/generate/utils/buildClipsFromAssets.ts

@@ -0,0 +1,57 @@
+/**
+ * 将选中素材 + 模板 segments 构建为 edit_plan_clips 写入数据。
+ *
+ * 逻辑必须与 FrontendPreviewPlayer.tsx 中 buildPlaybackSegments 完全一致：
+ *   assetDuration = asset.duration || asset.metadata?.duration || 30
+ *   tplSeg = templateSegments[i] || lastSegment
+ *   segDuration = clamp(assetDuration, tplSeg.duration_min, tplSeg.duration_max)
+ *   start_time = 0
+ *   duration = segDuration
+ */
+import type { AssetItem } from "@/api/assets"
+import type { TemplateSegment } from "@/api/templates/types"
+import type { EditPlanClipInput } from "@/api/template-editor"
+
+interface BuildClipsOptions {
+  /** 选中的素材 ID 列表（按选择顺序） */
+  selectedIds: string[]
+  /** 已加载的素材列表（用于查 duration） */
+  materials: AssetItem[]
+  /** 智能匹配返回的素材（auto 模式下可能不在 materials 列表中） */
+  smartMatchedAssets?: AssetItem[]
+  /** 模板 segments */
+  templateSegments?: TemplateSegment[]
+}
+
+export function buildClipsFromAssets({
+  selectedIds,
+  materials,
+  smartMatchedAssets = [],
+  templateSegments = [],
+}: BuildClipsOptions): EditPlanClipInput[] {
+  if (!selectedIds.length) return []
+
+  // 合并两个素材来源，建立 id → asset 索引
+  const assetMap = new Map<string, AssetItem>()
+  for (const a of materials) assetMap.set(a.id, a)
+  for (const a of smartMatchedAssets) assetMap.set(a.id, a)
+
+  const lastSeg = templateSegments[templateSegments.length - 1]
+
+  return selectedIds.map((assetId, i) => {
+    const asset = assetMap.get(assetId)
+    const assetDuration = asset?.duration || asset?.metadata?.duration || 30
+
+    const tplSeg = templateSegments[i] || lastSeg
+    const segDuration = tplSeg
+      ? Math.min(tplSeg.duration_max, Math.max(tplSeg.duration_min, assetDuration))
+      : Math.min(assetDuration, 10)
+
+    return {
+      asset_id: assetId,
+      start_time: 0,
+      duration: segDuration,
+      order: i,
+    }
+  })
+}

apps/worker/worker_app/tasks/generation.py

@@ -978,7 +978,7 @@ def _render_video(
 
     Args:
     Returns:
-        (output_path, render_duration, cover_candidates)
+        (output_path, render_duration, cover_candidates, voiceover_path)
     """
     if not downloaded_videos:
         raise RuntimeError(f"素材下载结果为空: task_id={task_id}")
@@ -1206,13 +1206,6 @@ def _upload_and_record(
 # ── Celery Task ──────────────────────────────────────────────────────────────
 
 
-@celery_app.task(
-    bind=True,
-    name="worker.generate_video",
-    max_retries=2,
-    soft_time_limit=600,  # 10 分钟软超时
-    time_limit=660,  # 11 分钟硬超时
-)
 def _sync_task_config_to_plan(source_edit_plan_id: str, task_info: dict, db) -> str | None:
     """将 GenerationTask 的配置同步到 EditPlan.config，返回配音本地路径（如果有）。
 
@@ -1295,7 +1288,7 @@ def _render_from_edit_plan(
     """从 EditPlan 数据库记录直接渲染（不再内存重建clips）。
 
     Returns:
-        (output_path, render_duration, cover_candidates)
+        (output_path, render_duration, cover_candidates, voiceover_path)
     """
     from video_processing.render_adapter import RenderAdapter
     from worker_app.db import SessionLocal
@@ -1335,13 +1328,18 @@ def _render_from_edit_plan(
         output_path = result.output_path
         cover_candidates = getattr(result, "cover_candidates", None)
 
-        return output_path, result.duration, cover_candidates
+        return output_path, result.duration, cover_candidates, voiceover_path
     finally:
         db.close()
-        # 清理临时配音文件
-        # voiceover_path 在外部作用域，这里不直接引用
 
 
+@celery_app.task(
+    bind=True,
+    name="worker.generate_video",
+    max_retries=2,
+    soft_time_limit=600,  # 10 分钟软超时
+    time_limit=660,  # 11 分钟硬超时
+)
 def generate_video(self, task_id: str) -> dict:
     """生成视频任务 — 使用 UnifiedRenderService 统一渲染。
 
@@ -1428,173 +1426,183 @@ def generate_video(self, task_id: str) -> dict:
         # ── 新路径：有 source_edit_plan_id 时直接从数据库 EditPlan 渲染 ──
         source_edit_plan_id = task_info.get("source_edit_plan_id", "")
         if source_edit_plan_id:
-            logger.info(
-                "[task_id=%s] 使用 EditPlan 数据库路径渲染: plan_id=%s",
-                task_id,
-                source_edit_plan_id,
-            )
-            _update_task_progress(task_id, 30, "加载草稿数据")
-
-            if gen_task:
-                gen_task.append_log("渲染模式", "从草稿数据渲染（与预览一致）")
-                _flush_logs(task_id, gen_task)
-
-            output_path, render_duration, cover_candidates = _render_from_edit_plan(
-                task_id=task_id,
-                source_edit_plan_id=source_edit_plan_id,
-                task_info=task_info,
-            )
-
-            if gen_task:
-                gen_task.append_log("渲染", f"渲染完成, 时长={render_duration:.1f}s")
-                _flush_logs(task_id, gen_task)
-
-            _update_task_progress(task_id, 80, "渲染完成")
-
-            # ── 4. 上传 OSS + 查重记录 ───────────────────────────────
-            _update_task_progress(task_id, 85, "开始上传")
-            file_url, duration, file_size, video_count = _upload_and_record(
-                task_id=task_id,
-                output_path=output_path,
-                project_id=project_id,
-                batch_id=batch_id,
-                editing_mode=editing_mode,
-                user_id=user_id,
-                video_name=task_info.get("video_title", ""),
-            )
-
-            if gen_task:
-                gen_task.append_log(
-                    "OSS上传",
-                    f"上传成功, 大小={file_size}",
-                    file_size=file_size,
-                    file_url=file_url,
+            voiceover_tmp_path: str | None = None
+            try:
+                logger.info(
+                    "[task_id=%s] 使用 EditPlan 数据库路径渲染: plan_id=%s",
+                    task_id,
+                    source_edit_plan_id,
                 )
-                _flush_logs(task_id, gen_task)
+                _update_task_progress(task_id, 30, "加载草稿数据")
 
-            _update_task_progress(task_id, 95, "上传完成")
+                if gen_task:
+                    gen_task.append_log("渲染模式", "从草稿数据渲染（与预览一致）")
+                    _flush_logs(task_id, gen_task)
 
-            # ── 4.5 封面帧持久化 ────────────────────────────────────────────
-            try:
-                if cover_candidates:
-                    first = cover_candidates[0]
-                    cover_frame_url = first.get("image_url") or first.get("url") or ""
-                    if cover_frame_url:
-                        _cover_session = SessionLocal()
-                        try:
-                            from packages.adapters.sqlalchemy_impl.models import (
-                                GenerationTaskModel,
-                            )
-
-                            _cover_model = (
-                                _cover_session.query(GenerationTaskModel)
-                                .filter(GenerationTaskModel.id == task_id)
-                                .first()
-                            )
-                            if _cover_model:
-                                _cover_model.cover_url = cover_frame_url
-                                meta = dict(_cover_model.metadata or {})
-                                meta["cover_candidates"] = cover_candidates
-                                _cover_model.metadata = meta
-                                _cover_session.commit()
-                        finally:
-                            _cover_session.close()
-            except Exception:
-                logger.warning("[task_id=%s] 封面帧持久化失败", task_id, exc_info=True)
-
-            # ── 5. 标记完成 ──────────────────────────────────────────────────
-            _update_task_status(task_id, "mark_completed", result_count=video_count)
-
-            # 5.1 更新标题使用次数
-            try:
-                _title_session = SessionLocal()
-                try:
-                    from packages.adapters.sqlalchemy_impl.generation_task_repository import (
-                        SQLAlchemyGenerationTaskRepository,
-                    )
-                    from packages.adapters.sqlalchemy_impl.title_library_repository import (
-                        SQLAlchemyTitleLibraryRepository,
-                    )
-
-                    _task_repo = SQLAlchemyGenerationTaskRepository(_title_session)
-                    _gen_task = _task_repo.get(task_id)
-                    if _gen_task and _gen_task.title_ids and _gen_task.created_by_user_id:
-                        _title_repo = SQLAlchemyTitleLibraryRepository(_title_session)
-                        for _tid in _gen_task.title_ids:
-                            try:
-                                _title_repo.increment_usage_count(_tid, _gen_task.created_by_user_id)
-                            except Exception:
-                                logger.warning(
-                                    "[task_id=%s] 更新标题使用次数失败: title_id=%s",
-                                    task_id,
-                                    _tid,
-                                    exc_info=True,
-                                )
-                finally:
-                    _title_session.close()
-            except Exception:
-                logger.warning("[task_id=%s] 更新标题使用次数异常", task_id, exc_info=True)
-
-            # 5.2 更新素材使用次数
-            try:
-                from worker_app.core.asset_usage import mark_asset_used_for_generation
-
-                _asset_session = SessionLocal()
-                try:
-                    from packages.adapters.sqlalchemy_impl.asset_repository import (
-                        SQLAlchemyAssetRepository,
-                    )
-                    from packages.adapters.sqlalchemy_impl.generation_task_repository import (
-                        SQLAlchemyGenerationTaskRepository,
-                    )
-
-                    _task_repo = SQLAlchemyGenerationTaskRepository(_asset_session)
-                    _asset_repo = SQLAlchemyAssetRepository(_asset_session)
-                    _gen_task = _task_repo.get(task_id)
-                    if _gen_task and _gen_task.asset_ids:
-                        for _aid in _gen_task.asset_ids:
-                            try:
-                                _asset = _asset_repo.get(_aid)
-                                if _asset:
-                                    mark_asset_used_for_generation(_asset)
-                                    _asset_repo.update(_asset)
-                            except Exception:
-                                logger.warning(
-                                    "[task_id=%s] 更新素材使用次数失败: asset_id=%s",
-                                    task_id,
-                                    _aid,
-                                    exc_info=True,
-                                )
-                finally:
-                    _asset_session.close()
-            except Exception:
-                logger.warning("[task_id=%s] 更新素材使用次数异常", task_id, exc_info=True)
-
-            if gen_task:
-                gen_task.append_log(
-                    "任务完成",
-                    f"视频生成完成: 时长={duration:.2f}s, 大小={file_size}",
-                    duration=round(duration, 2),
-                    file_size=file_size,
-                    video_count=video_count,
+                output_path, render_duration, cover_candidates, voiceover_tmp_path = _render_from_edit_plan(
+                    task_id=task_id,
+                    source_edit_plan_id=source_edit_plan_id,
+                    task_info=task_info,
                 )
-                _flush_logs(task_id, gen_task)
 
-            logger.info(
-                "[task_id=%s] [任务完成] duration=%.2fs file_size=%d (edit_plan path)",
-                task_id,
-                duration,
-                file_size,
-            )
+                if gen_task:
+                    gen_task.append_log("渲染", f"渲染完成, 时长={render_duration:.1f}s")
+                    _flush_logs(task_id, gen_task)
 
-            return {
-                "status": "completed",
-                "task_id": task_id,
-                "output_path": str(output_path),
-                "file_size": file_size,
-                "duration": duration,
-                "mode": editing_mode.value,
-            }
+                _update_task_progress(task_id, 80, "渲染完成")
+
+                # ── 4. 上传 OSS + 查重记录 ───────────────────────────────
+                _update_task_progress(task_id, 85, "开始上传")
+                file_url, duration, file_size, video_count = _upload_and_record(
+                    task_id=task_id,
+                    output_path=output_path,
+                    project_id=project_id,
+                    batch_id=batch_id,
+                    editing_mode=editing_mode,
+                    user_id=user_id,
+                    video_name=task_info.get("video_title", ""),
+                )
+
+                if gen_task:
+                    gen_task.append_log(
+                        "OSS上传",
+                        f"上传成功, 大小={file_size}",
+                        file_size=file_size,
+                        file_url=file_url,
+                    )
+                    _flush_logs(task_id, gen_task)
+
+                _update_task_progress(task_id, 95, "上传完成")
+
+                # ── 4.5 封面帧持久化 ────────────────────────────────────────────
+                try:
+                    if cover_candidates:
+                        first = cover_candidates[0]
+                        cover_frame_url = first.get("image_url") or first.get("url") or ""
+                        if cover_frame_url:
+                            _cover_session = SessionLocal()
+                            try:
+                                from packages.adapters.sqlalchemy_impl.models import (
+                                    GenerationTaskModel,
+                                )
+
+                                _cover_model = (
+                                    _cover_session.query(GenerationTaskModel)
+                                    .filter(GenerationTaskModel.id == task_id)
+                                    .first()
+                                )
+                                if _cover_model:
+                                    _cover_model.cover_url = cover_frame_url
+                                    meta = dict(_cover_model.extra_meta or {})
+                                    meta["cover_candidates"] = cover_candidates
+                                    _cover_model.extra_meta = meta
+                                    _cover_session.commit()
+                            finally:
+                                _cover_session.close()
+                except Exception:
+                    logger.warning("[task_id=%s] 封面帧持久化失败", task_id, exc_info=True)
+
+                # ── 5. 标记完成 ──────────────────────────────────────────────────
+                _update_task_status(task_id, "mark_completed", result_count=video_count)
+
+                # 5.1 更新标题使用次数
+                try:
+                    _title_session = SessionLocal()
+                    try:
+                        from packages.adapters.sqlalchemy_impl.generation_task_repository import (
+                            SQLAlchemyGenerationTaskRepository,
+                        )
+                        from packages.adapters.sqlalchemy_impl.title_library_repository import (
+                            SQLAlchemyTitleLibraryRepository,
+                        )
+
+                        _task_repo = SQLAlchemyGenerationTaskRepository(_title_session)
+                        _gen_task = _task_repo.get(task_id)
+                        if _gen_task and _gen_task.title_ids and _gen_task.created_by_user_id:
+                            _title_repo = SQLAlchemyTitleLibraryRepository(_title_session)
+                            for _tid in _gen_task.title_ids:
+                                try:
+                                    _title_repo.increment_usage_count(_tid, _gen_task.created_by_user_id)
+                                except Exception:
+                                    logger.warning(
+                                        "[task_id=%s] 更新标题使用次数失败: title_id=%s",
+                                        task_id,
+                                        _tid,
+                                        exc_info=True,
+                                    )
+                    finally:
+                        _title_session.close()
+                except Exception:
+                    logger.warning("[task_id=%s] 更新标题使用次数异常", task_id, exc_info=True)
+
+                # 5.2 更新素材使用次数
+                try:
+                    from worker_app.core.asset_usage import mark_asset_used_for_generation
+
+                    _asset_session = SessionLocal()
+                    try:
+                        from packages.adapters.sqlalchemy_impl.asset_repository import (
+                            SQLAlchemyAssetRepository,
+                        )
+                        from packages.adapters.sqlalchemy_impl.generation_task_repository import (
+                            SQLAlchemyGenerationTaskRepository,
+                        )
+
+                        _task_repo = SQLAlchemyGenerationTaskRepository(_asset_session)
+                        _asset_repo = SQLAlchemyAssetRepository(_asset_session)
+                        _gen_task = _task_repo.get(task_id)
+                        if _gen_task and _gen_task.asset_ids:
+                            for _aid in _gen_task.asset_ids:
+                                try:
+                                    _asset = _asset_repo.get(_aid)
+                                    if _asset:
+                                        mark_asset_used_for_generation(_asset)
+                                        _asset_repo.update(_asset)
+                                except Exception:
+                                    logger.warning(
+                                        "[task_id=%s] 更新素材使用次数失败: asset_id=%s",
+                                        task_id,
+                                        _aid,
+                                        exc_info=True,
+                                    )
+                    finally:
+                        _asset_session.close()
+                except Exception:
+                    logger.warning("[task_id=%s] 更新素材使用次数异常", task_id, exc_info=True)
+
+                if gen_task:
+                    gen_task.append_log(
+                        "任务完成",
+                        f"视频生成完成: 时长={duration:.2f}s, 大小={file_size}",
+                        duration=round(duration, 2),
+                        file_size=file_size,
+                        video_count=video_count,
+                    )
+                    _flush_logs(task_id, gen_task)
+
+                logger.info(
+                    "[task_id=%s] [任务完成] duration=%.2fs file_size=%d (edit_plan path)",
+                    task_id,
+                    duration,
+                    file_size,
+                )
+
+                return {
+                    "status": "completed",
+                    "task_id": task_id,
+                    "output_path": str(output_path),
+                    "file_size": file_size,
+                    "duration": duration,
+                    "mode": editing_mode.value,
+                }
+
+            finally:
+                # 无论任务成功或失败，都清理临时配音文件，避免磁盘泄漏
+                if voiceover_tmp_path:
+                    try:
+                        Path(voiceover_tmp_path).unlink(missing_ok=True)
+                    except OSError:
+                        logger.warning("[task_id=%s] 清理临时配音文件失败: %s", task_id, voiceover_tmp_path)
 
         # DEPRECATED: 以下为旧路径，仅兼容无 source_edit_plan_id 的旧调用，后续移除
         with tempfile.TemporaryDirectory(prefix="xiaoxia-generation-") as temp_dir:
@@ -1672,10 +1680,27 @@ def generate_video(self, task_id: str) -> dict:
                             logger.info("[task_id=%s] MediaKit 视频理解完成: %d 个素材", task_id, len(asset_urls))
 
                     # 保存分析结果到 extra_meta
-                    if asset_analyses and gen_task:
-                        gen_task.extra_meta = {**(gen_task.extra_meta or {}), "asset_analyses": asset_analyses}
-                        _repo.update(gen_task)
-                        _flush_logs(task_id, gen_task)
+                    if asset_analyses:
+                        _meta_session = SessionLocal()
+                        try:
+                            from packages.adapters.sqlalchemy_impl.models import (
+                                GenerationTaskModel,
+                            )
+
+                            _m = (
+                                _meta_session.query(GenerationTaskModel)
+                                .filter(GenerationTaskModel.id == task_id)
+                                .first()
+                            )
+                            if _m:
+                                existing = dict(_m.extra_meta or {})
+                                existing["asset_analyses"] = asset_analyses
+                                _m.extra_meta = existing
+                                _meta_session.commit()
+                        finally:
+                            _meta_session.close()
+                        if gen_task:
+                            _flush_logs(task_id, gen_task)
                 except Exception:
                     logger.warning("[task_id=%s] MediaKit 视频理解失败，继续渲染", task_id, exc_info=True)
 
@@ -1769,10 +1794,10 @@ def generate_video(self, task_id: str) -> dict:
                             )
                             if _cover_model:
                                 _cover_model.cover_url = cover_frame_url
-                                # 持久化完整候选列表到 metadata
-                                meta = dict(_cover_model.metadata or {})
+                                # 持久化完整候选列表到 extra_meta
+                                meta = dict(_cover_model.extra_meta or {})
                                 meta["cover_candidates"] = cover_candidates
-                                _cover_model.metadata = meta
+                                _cover_model.extra_meta = meta
                                 _cover_session.commit()
                                 logger.info(
                                     "[task_id=%s] 封面帧已持久化(ffmpeg本地抽帧): cover_url=%s candidates=%d",

tests/unit/test_worker_cover_meta_and_status.py

@@ -0,0 +1,70 @@
+"""防回归测试：P1 修复
+- Bug 1: gen_task 过期内存对象 _repo.update() 覆盖 DB status 为 pending
+- Bug 2: 封面模型误用 .metadata（SQLAlchemy 保留属性），应为 .extra_meta
+"""
+
+import ast
+from pathlib import Path
+
+import pytest
+
+GENERATION_FILE = Path(__file__).resolve().parents[2] / "apps" / "worker" / "worker_app" / "tasks" / "generation.py"
+
+
+def _read_source() -> str:
+    return GENERATION_FILE.read_text(encoding="utf-8")
+
+
+class TestCoverModelUsesExtraMeta:
+    """封面持久化必须使用 ORM 属性 extra_meta，而不是 SQLAlchemy 保留的 .metadata。"""
+
+    def test_no_metadata_attribute_access_on_cover_model(self):
+        source = _read_source()
+        # 禁止对 _cover_model.metadata 进行读或写
+        assert "_cover_model.metadata" not in source, (
+            "_cover_model.metadata is the SQLAlchemy reserved MetaData object, "
+            "not the JSON column. Use _cover_model.extra_meta instead."
+        )
+
+    def test_extra_meta_used_for_cover_candidates(self):
+        source = _read_source()
+        assert "_cover_model.extra_meta" in source
+        assert 'meta["cover_candidates"]' in source
+
+
+class TestAssetAnalysesDoesNotOverwriteStatus:
+    """保存 asset_analyses 时不能用过期的 gen_task 内存对象整体 _repo.update，
+    否则会把已被 _update_task_status 改为 running 的 status 覆盖回 pending。"""
+
+    def test_no_stale_repo_update_with_gen_task(self):
+        source = _read_source()
+        # 旧代码：gen_task.extra_meta = {...}; _repo.update(gen_task)
+        # 这行会把内存中的 pending status 写回 DB
+        assert "_repo.update(gen_task)" not in source, (
+            "_repo.update(gen_task) writes a stale in-memory object back to DB, "
+            "overwriting status set by _update_task_status. "
+            "Use an independent session to update only extra_meta."
+        )
+
+    def test_asset_analyses_uses_independent_session(self):
+        """asset_analyses 持久化必须用独立 session 查询最新模型再提交。"""
+        source = _read_source()
+        assert "_meta_session" in source
+        assert "GenerationTaskModel" in source
+        # 必须只更新 extra_meta 字段
+        assert 'existing["asset_analyses"]' in source
+
+
+class TestGenerationTaskModelOrmAttribute:
+    """确认 ORM 属性映射：Python 属性 extra_meta -> DB 列 metadata。"""
+
+    def test_orm_attribute_is_extra_meta(self):
+        from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel
+
+        # ORM 属性必须存在
+        assert hasattr(GenerationTaskModel, "extra_meta")
+        # .metadata 是 SQLAlchemy 声明基类保留的 MetaData，不是列描述符
+        # 它不应该是我们的 JSON 字段
+        from sqlalchemy import MetaData
+
+        assert isinstance(GenerationTaskModel.metadata, MetaData)

tests/unit/test_worker_generate_video_task_binding.py

@@ -0,0 +1,39 @@
+"""Regression test: ensure worker.generate_video Celery task is bound to the
+real generate_video function, not a helper introduced above it.
+
+Context (P0 incident 2026-08-23): a refactor inserted helper function
+_sync_task_config_to_plan directly under the @celery_app.task decorator,
+so Celery registered the helper as "worker.generate_video". Calling the
+task with a single task_id raised TypeError and every generation job
+failed immediately. This test pins the decorator target.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+
+def test_generate_video_task_registered_under_expected_name():
+    from worker_app.tasks.generation import generate_video
+
+    # Celery task object exposes its registered name
+    assert generate_video.name == "worker.generate_video"
+
+
+def test_generate_video_task_signature_has_task_id():
+    from worker_app.tasks.generation import generate_video
+
+    # For bind=True tasks Celery binds self at call time, so run() signature
+    # starts directly with task_id (verified on Celery 5.x).
+    sig = inspect.signature(generate_video.run)
+    params = list(sig.parameters)
+    assert params[0] == "task_id", f"expected task_id as first param, got {params}"
+
+
+def test_sync_task_config_to_plan_is_plain_function():
+    """Helper must NOT be registered as a Celery task."""
+    from worker_app.tasks.generation import _sync_task_config_to_plan
+
+    assert not hasattr(
+        _sync_task_config_to_plan, "run"
+    ), "_sync_task_config_to_plan must be a plain function, not a Celery task"