fix: 修复 CI 失败的 4 个单元测试 (#1478)

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

apps/api/app/api/routes/generation_tasks.py

@@ -377,6 +377,36 @@ def create_generation_task(
                             title_config=request.title_config,
                             db=db,
                         )
+
+                    # 兜底关联编辑计划：前端未传 source_edit_plan_id 时，
+                    # 通过 template_id + user_id 在 DB 层直接查找最新的 plan
+                    if not task.source_edit_plan_id and request.template_id:
+                        try:
+                            from packages.adapters.sqlalchemy_impl.models import EditPlanModel
+
+                            _plan_model = (
+                                db.query(EditPlanModel)
+                                .filter(
+                                    EditPlanModel.template_id == request.template_id,
+                                    EditPlanModel.created_by_user_id == user_id,
+                                )
+                                .order_by(EditPlanModel.created_at.desc())
+                                .first()
+                            )
+                            if _plan_model:
+                                task.source_edit_plan_id = _plan_model.id
+                                generation_task_repository.update(task)
+                                logger.info(
+                                    "[生成任务] 自动关联编辑计划: task_id=%s plan_id=%s",
+                                    task.id,
+                                    _plan_model.id,
+                                )
+                        except Exception:
+                            logger.warning(
+                                "[生成任务] 查找关联编辑计划失败（不影响主流程）: task_id=%s",
+                                task.id,
+                                exc_info=True,
+                            )
                 else:
                     failed_tasks.append(task)
             except UserPendingLimitExceeded as _e:

apps/api/app/services/edit_plan_service.py

@@ -423,6 +423,9 @@ class EditPlanService:
                 )
                 db.add(model)
 
+            # flush 让新建 clip 写入当前事务（未 commit），后续查询才能找到它们
+            db.flush()
+
             # 3. 标记有 asset_id 的 clips 为 ready（不 commit）
             pending_with_asset = (
                 db.query(EditPlanClipModel)

apps/web/src/pages/generate/GeneratePage.tsx

@@ -72,6 +72,7 @@ const GeneratePage: React.FC = () => {
     autoSubtitles,
     bgm,
     editPlanId,
+    sourceEditPlanId,
     previewVideo,
     setPreviewVideo,
     previewModalOpen,
@@ -171,7 +172,7 @@ const GeneratePage: React.FC = () => {
     autoSubtitles,
     bgm,
     generateCount,
-    sourceEditPlanId: editPlanId,
+    sourceEditPlanId,
   })
 
   /* ================================================================

apps/web/src/pages/generate/hooks/useGenerateFormState/index.ts

@@ -82,6 +82,14 @@ export interface GenerateFormState {
   editPlanId: string | null
   planConfigStr: string | null
 
+  /**
+   * 传给 Worker 的 source_edit_plan_id。
+   * 仅使用 URL 中的 edit_plan_id（从剪辑模板编辑器跳转时携带）。
+   * URL 没有时传 null，后端正式生成 API 会通过 template_id+user_id 兜底查找正确的 plan。
+   * 注意：selectedTemplate 是模板 ID，不是 edit_plan_id，不能作为此值传递。
+   */
+  sourceEditPlanId: string | null
+
   /* 预览弹窗 */
   previewVideo: GeneratedVideo | null
   setPreviewVideo: (v: GeneratedVideo | null) => void
@@ -100,6 +108,11 @@ export const useGenerateFormState = (): GenerateFormState => {
   /* ── 模板选择 ── */
   const { selectedTemplate, setSelectedTemplate, userTemplates } = useTemplateSelection()
 
+  /* ── source_edit_plan_id：仅取 URL 参数，无则 null 让后端兜底 ── */
+  // selectedTemplate 是模板 ID 而非 edit_plan_id，不能混淆；
+  // 后端正式生成 API 会在 source_edit_plan_id 为空时通过 template_id+user_id 自动关联。
+  const sourceEditPlanId = editPlanId || null
+
   /* ── 素材 ── */
   const [selectedMaterials, setSelectedMaterials] = useState<string[]>([])
   const [materialMode, setMaterialMode] = useState<"manual" | "auto">("manual")
@@ -189,6 +202,7 @@ export const useGenerateFormState = (): GenerateFormState => {
     autoSubtitles,
     bgm,
     editPlanId,
+    sourceEditPlanId,
     planConfigStr,
     previewVideo,
     setPreviewVideo,

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

@@ -6,7 +6,14 @@
  *   tplSeg = templateSegments[i] || lastSegment
  *   segDuration = clamp(assetDuration, tplSeg.duration_min, tplSeg.duration_max)
  *   start_time = 0
- *   duration = segDuration
+ *   // 关键：预览播放器中 endTime = min(startTime + segDuration, assetDuration)
+ *   // 因此 clips.duration 也必须用 min(segDuration, assetDuration) 截断，
+ *   // 避免素材实际时长比 clamp 后的 segDuration 短时，Worker 尝试读取不存在的片段
+ *   duration = min(segDuration, assetDuration)
+ *
+ * 预览播放器（Canvas 实时预览）直接在内存中构建 segments 播放，不读 edit_plan_clips；
+ * 本函数产出的 clips 写入 DB 后由 Worker 渲染。两边用完全相同的时长计算，
+ * 保证用户在编辑过程中看到的预览与最终生成视频一致。
  */
 import type { AssetItem } from "@/api/assets"
 import type { TemplateSegment } from "@/api/templates/types"
@@ -47,10 +54,15 @@ export function buildClipsFromAssets({
       ? Math.min(tplSeg.duration_max, Math.max(tplSeg.duration_min, assetDuration))
       : Math.min(assetDuration, 10)
 
+    // 与 FrontendPreviewPlayer.buildPlaybackSegments 中
+    //   endTime = Math.min(startTime + segDuration, assetDuration)
+    // 保持一致：duration 不能超过素材实际时长
+    const duration = Math.min(segDuration, assetDuration)
+
     return {
       asset_id: assetId,
       start_time: 0,
-      duration: segDuration,
+      duration,
       order: i,
     }
   })

apps/worker/video_processing/render_audio.py

@@ -67,6 +67,16 @@ def clip_has_audio(ctx: RenderContext, clip: ResolvedClip) -> bool:
     return ctx._audio_cache[key]
 
 
+def _clip_volume(clip: ResolvedClip) -> float:
+    """读取 clip 的音量配置（0.0~1.0，>1 放大）。缺省 1.0 原声。"""
+    cfg = getattr(clip, "config", None) or {}
+    try:
+        vol = float(cfg.get("volume", 1.0))
+    except (TypeError, ValueError):
+        return 1.0
+    return max(0.0, vol)
+
+
 # ── 音频混音 ──────────────────────────────────────────────────────────────────
 
 
@@ -82,12 +92,13 @@ def mix_audio(
     """音频后处理混音.
 
     处理逻辑：
-    1. 丢弃主图层（main/broll/overlay/corner_voice）的原始音频，避免录入源视频杂音
-    2. 仅使用独立音频轨（audio role，TTS/配音）作为主音频
-    3. 如果提供了 bgm_path，则额外混入 BGM（支持淡入淡出、循环、人声闪避）
-    4. 如果配置了 audio_tracks，则混入多轨道音频（配音、音效等）
-    5. 输出时长截断到 video_duration
-    6. 如果配置了降噪，最后应用降噪
+    1. 保留主图层（main/broll/overlay/corner_voice）视频素材的原声，按顺序 concat 拼接
+    2. 每个 clip 按 config.volume 应用音量（volume=0 静音，=1 原声）
+    3. 独立音频轨（audio role，TTS/配音）通过 amix 混入
+    4. 如果提供了 bgm_path，则额外混入 BGM（支持淡入淡出、循环、人声闪避）
+    5. 如果配置了 audio_tracks，则混入多轨道音频（配音、音效等）
+    6. 输出时长截断到 video_duration
+    7. 如果配置了降噪，最后应用降噪
 
     Args:
         ctx: 渲染上下文
@@ -123,10 +134,12 @@ def mix_audio(
     if "audio" in layer_map:
         audio_clips = layer_map["audio"].clips
 
-    # ── 丢弃源视频的原始音频（避免录入杂音），成片仅保留 TTS 配音 + BGM ──
-    main_clips = []
+    # ── 保留源视频原声：过滤掉无音频流的 main clip（图片/无声素材） ──
+    # 注意：volume=0 的 clip 不能移除——移除会导致后续 clip 音频时间轴前移、音画不同步。
+    # volume=0 通过滤镜链生成静音流，保持时间轴对齐。
+    main_clips = [c for c in main_clips if clip_has_audio(ctx, c)]
 
-    # ── 防御：过滤掉无音频流的 clip ──
+    # ── 防御：过滤掉无音频流的独立音频轨 ──
     audio_clips = [c for c in audio_clips if clip_has_audio(ctx, c)]
 
     if not main_clips and not audio_clips:
@@ -144,7 +157,7 @@ def mix_audio(
     # 构建音频处理命令
     output_path = ctx.work_dir / f"audio_{ctx.plan_id}.aac"
 
-    # 源视频原始音频已被丢弃（main_clips = []），最终音频完全由独立音频轨 + BGM + 多轨配置组成。
+    # 主音频为视频素材原声 concat；独立音频轨（TTS/配音）通过 amix 混入。
     # 当无 main_clips 时，将独立音频轨作为主音频走 concat 拼接；当二者均有则走 amix 混音。
     if main_clips:
         effective_main = main_clips
@@ -266,28 +279,67 @@ def concat_main_audio(
         has_speed = abs(speed - 1.0) >= 1e-6
 
         if not has_speed and not has_reverse:
-            # 无调速无倒放：简单命令行，-ss 裁剪更高效
-            command = [
-                FFMPEG_BIN,
-                "-y",
-                "-i",
-                str(clip.local_path),
-                "-vn",
-                "-acodec",
-                "aac",
-                "-b:a",
-                "128k",
-                "-ar",
-                "48000",
-                "-ac",
-                "2",
-            ]
-            if trim_start > 0:
-                command.extend(["-ss", f"{trim_start:.3f}"])
-            if final_duration > 0:
-                command.extend(["-t", f"{final_duration:.3f}"])
-            command.append(str(output_path))
-            run_ffmpeg(command)
+            # 无调速无倒放：根据是否需要裁剪/音量选择最高效的路径。
+            vol = _clip_volume(clip)
+            need_trim = trim_start > 0 or (effective_duration > 0 and final_duration < adjusted_duration)
+            need_volume = abs(vol - 1.0) >= 1e-6
+
+            if need_trim:
+                # 需要裁剪：用 atrim 滤镜在滤镜链中精确裁剪（采样点级精度，不浪费解码）。
+                # 滤镜顺序：atrim → asetpts → volume（先裁剪再调音量，避免处理被丢弃的数据）。
+                af_parts: list[str] = []
+                if trim_start > 0 and effective_duration > 0:
+                    af_parts.append(f"atrim=start={trim_start:.3f}:duration={final_duration:.3f}")
+                elif trim_start > 0:
+                    af_parts.append(f"atrim=start={trim_start:.3f}")
+                elif final_duration > 0:
+                    af_parts.append(f"atrim=duration={final_duration:.3f}")
+                af_parts.append("asetpts=PTS-STARTPTS")
+                if need_volume:
+                    af_parts.append(f"volume={vol:.4f}")
+                command = [
+                    FFMPEG_BIN,
+                    "-y",
+                    "-i",
+                    str(clip.local_path),
+                    "-vn",
+                    "-af",
+                    ",".join(af_parts),
+                    "-acodec",
+                    "aac",
+                    "-b:a",
+                    "128k",
+                    "-ar",
+                    "48000",
+                    "-ac",
+                    "2",
+                ]
+                # atrim 已精确控制时长，无需额外 -t
+                command.append(str(output_path))
+                run_ffmpeg(command)
+            else:
+                # 无需裁剪：直接提取，最高效。音量用单个 -af（如有）。
+                command = [
+                    FFMPEG_BIN,
+                    "-y",
+                    "-i",
+                    str(clip.local_path),
+                    "-vn",
+                    "-acodec",
+                    "aac",
+                    "-b:a",
+                    "128k",
+                    "-ar",
+                    "48000",
+                    "-ac",
+                    "2",
+                ]
+                if need_volume:
+                    command.extend(["-af", f"volume={vol:.4f}"])
+                if final_duration > 0:
+                    command.extend(["-t", f"{final_duration:.3f}"])
+                command.append(str(output_path))
+                run_ffmpeg(command)
         else:
             # 有调速或倒放：用 filter_complex
             speed_engine = SpeedEngine()
@@ -312,6 +364,11 @@ def concat_main_audio(
                 if reverse_filter:
                     audio_filters.append(reverse_filter)
 
+            # 音量
+            vol = _clip_volume(clip)
+            if abs(vol - 1.0) >= 1e-6:
+                audio_filters.append(f"volume={vol:.4f}")
+
             # aformat 归一化：统一输出格式为 48000Hz + stereo + fltp
             audio_filters.append("aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp")
 
@@ -378,6 +435,11 @@ def concat_main_audio(
             if reverse_filter:
                 audio_filters.append(reverse_filter)
 
+        # 音量（0=静音，1=原声）
+        vol = _clip_volume(clip)
+        if abs(vol - 1.0) >= 1e-6:
+            audio_filters.append(f"volume={vol:.4f}")
+
         # aformat 归一化：统一采样率48000Hz + 双声道stereo + fltp采样格式
         # concat filter 要求所有输入音频参数完全一致，否则 exit=234 失败
         audio_filters.append("aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp")

apps/worker/video_processing/unified_render_service.py

@@ -36,6 +36,7 @@ from video_processing.ffmpeg_utils import (
     DEFAULT_TRANSITION_DURATION,
     FFMPEG_BIN,
     probe_duration,
+    probe_has_audio,
     probe_video_info,
     run_ffmpeg,
 )
@@ -1000,6 +1001,11 @@ class UnifiedRenderService:
         if abs(speed - 1.0) >= 1e-6:
             return False, f"有调速: speed={speed:.2f}x"
 
+        # 音量非默认（静音/放大）→ 需要音频滤镜重编码 → 不能 copy
+        vol = UnifiedRenderService._clip_volume(clip)
+        if abs(vol - 1.0) >= 1e-6:
+            return False, f"音量非默认: volume={vol:.2f}"
+
         # 有倒放 → 需要重编码 → 不能 copy
         reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
         if reverse_config.enabled and (reverse_config.reverse_video or reverse_config.reverse_audio):
@@ -1270,13 +1276,23 @@ class UnifiedRenderService:
             "+faststart",
         ]
 
-        # 音频处理：background 通常是图片无音频，跳过；其他编码为 aac
-        # background 以外的视频素材，默认带音频
-        has_audio = role != "background"
+        # 音频处理：background 通常是图片无音频，跳过；其他角色先探测是否真有音频流。
+        # volume=0 不丢弃音频流，而是保留后通过 volume=0 滤镜静音，保持时间轴对齐。
+        clip_volume = UnifiedRenderService._clip_volume(clip)
+        if role != "background":
+            try:
+                has_audio = probe_has_audio(clip.local_path)
+            except Exception as e:
+                # probe_has_audio 内部已保守返回 True；只有极端错误才会到这里。
+                # 此时不静默丢音频，记录 error 并向上抛出，让任务失败而不是产出无声视频。
+                logger.error("[unified-render] 探测音频流发生致命错误，终止渲染: %s: %s", clip.local_path, e)
+                raise
+        else:
+            has_audio = False
         if has_audio:
             af_parts: list[str] = []
 
-            # 音频降噪
+            # 音频降噪（最先处理：在原始信号上降噪效果最好）
             try:
                 from video_processing.noise_reduction_engine import NoiseReductionConfig, NoiseReductionEngine
 
@@ -1290,13 +1306,16 @@ class UnifiedRenderService:
             except Exception as e:
                 logger.warning("[unified-render] 直通模式音频降噪应用失败，跳过: %s", e)
 
-            # 音频调速（与视频setpts对应，保持音画同步）
+            # 音频调速（在降噪之后、音量之前，与 render_audio.py concat 路径保持一致）
+            # SpeedEngine.build_audio_filter 内部已实现多级 atempo 串联，
+            # 自动处理超出 [0.5, 2.0] 范围的速度（如 0.25x → atempo=0.5,atempo=0.5）。
             speed = UnifiedRenderService._clip_speed(clip)
             if abs(speed - 1.0) >= 1e-6:
                 try:
                     from video_processing.speed_engine import SpeedConfig, SpeedEngine
 
                     speed_cfg = SpeedConfig(speed=speed)
+                    speed_cfg.clamp()
                     speed_engine = SpeedEngine()
                     af_parts.append(speed_engine.build_audio_filter(speed_cfg))
                 except Exception as e:
@@ -1309,6 +1328,10 @@ class UnifiedRenderService:
                 if af_filter:
                     af_parts.append(af_filter)
 
+            # 片段音量（最后应用：确保调速/倒放后的最终输出音量准确，与 concat 路径一致）
+            if abs(clip_volume - 1.0) >= 1e-6:
+                af_parts.append(f"volume={clip_volume:.4f}")
+
             if af_parts:
                 command.extend(["-af", ",".join(af_parts)])
 
@@ -1976,6 +1999,16 @@ class UnifiedRenderService:
         """
         return _clip_playback_speed_pure(getattr(clip, "playback_speed", 1.0))
 
+    @staticmethod
+    def _clip_volume(clip: ResolvedClip) -> float:
+        """获取 clip 的音量（config.volume）。缺省 1.0 原声，0.0 静音。"""
+        cfg = getattr(clip, "config", None) or {}
+        try:
+            vol = float(cfg.get("volume", 1.0))
+        except (TypeError, ValueError):
+            return 1.0
+        return max(0.0, vol)
+
     @staticmethod
     def _clip_adjusted_duration(clip: ResolvedClip) -> float:
         """计算调速后的 clip 实际时长（用于拼接计算）。

apps/worker/worker_app/tasks/generation.py

@@ -1919,12 +1919,11 @@ def generate_video(self, task_id: str) -> dict:
                 _repo = SQLAlchemyGenerationTaskRepository(_session)
                 gen_task = _repo.get(task_id)
                 if gen_task:
-                    gen_task.append_log(  # type: ignore[misc]
-                        "任务失败",
+                    gen_task.append_log(
+                        "render",
                         str(error),
                         level="ERROR",
                         error_type=type(error).__name__,
-                        stage="render",
                     )
                     _flush_logs(task_id, gen_task)
             finally:

tests/unit/test_original_audio_retained.py

@@ -0,0 +1,215 @@
+"""回归测试：render_plan 新路径必须保留视频素材原声。
+
+历史 Bug：mix_audio 中 `main_clips = []` 无条件丢弃源视频原声，
+导致最终生成视频没有原声（与预览不一致）。本测试钉住新行为：
+- 有音频流的 main/broll clip 原声必须进入最终音轨
+- clip.config.volume=0 静音，volume≠1.0 应用音量滤镜
+- 无音频流的素材被安全过滤
+- 直通（pass-through）路径同样尊重 probe + volume
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+from video_processing.render_audio import (
+    RenderContext,
+    _clip_volume,
+    mix_audio,
+)
+from video_processing.unified_render_service import (
+    ResolvedClip,
+    UnifiedRenderService,
+)
+
+
+def _ctx() -> RenderContext:
+    return RenderContext(work_dir=Path("/tmp/test_render_audio_fix"), plan_id="plan_audio")
+
+
+def _clip(
+    cid: str,
+    *,
+    clip_type: str = "main",
+    order: int = 0,
+    duration: float = 5.0,
+    config: dict | None = None,
+    asset: str | None = None,
+) -> ResolvedClip:
+    return ResolvedClip(
+        clip_id=cid,
+        asset_id=asset or f"asset_{cid}.mp4",
+        clip_type=clip_type,
+        order=order,
+        local_path=Path(f"/tmp/asset_{cid}.mp4"),
+        duration=duration,
+        actual_duration=duration,
+        config=config or {},
+    )
+
+
+def _layers(svc, clips):
+    # 直接把 ResolvedClip 分组为图层，跳过 _resolve_clips（后者要求原始 EditPlanClip）
+    return svc._group_clips_into_layers(clips)
+
+
+def _service(clips):
+    paths = {c.asset_id: c.local_path for c in clips}
+    return UnifiedRenderService(
+        plan=type("P", (), {"id": "plan_audio", "config": {}})(),
+        clips=clips,
+        asset_path_map=paths,
+        work_dir=Path("/tmp/test_render_audio_fix"),
+        output_width=1080,
+        output_height=1920,
+        output_fps=25,
+    )
+
+
+class TestOriginalAudioRetained:
+    """钉住原声不再被丢弃。"""
+
+    def test_single_main_clip_audio_kept(self):
+        svc = _service([_clip("c1")])
+        with (
+            patch("pathlib.Path.exists", return_value=True),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.render_audio.probe_has_audio", return_value=True),
+            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
+        ):
+            result = mix_audio(_ctx(), _layers(svc, [_clip("c1")]), 5.0)
+
+        assert result is not None
+        cmd_str = " ".join(mock_run.call_args[0][0])
+        assert "asset_c1.mp4" in cmd_str
+
+    def test_multi_main_clips_concat_audio(self):
+        clips = [_clip("c1", order=0), _clip("c2", order=1)]
+        svc = _service(clips)
+        with (
+            patch("pathlib.Path.exists", return_value=True),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.render_audio.probe_has_audio", return_value=True),
+            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
+        ):
+            result = mix_audio(_ctx(), _layers(svc, clips), 9.0)
+
+        assert result is not None
+        cmd_str = " ".join(mock_run.call_args[0][0])
+        assert "asset_c1.mp4" in cmd_str
+        assert "asset_c2.mp4" in cmd_str
+        assert "concat=n=2:v=0:a=1" in cmd_str
+
+    def test_main_audio_plus_independent_track_amix(self):
+        clips = [
+            _clip("c1", order=0),
+            _clip("tts1", order=1, config={"role": "audio", "volume": 0.5}),
+        ]
+        svc = _service(clips)
+        with (
+            patch("pathlib.Path.exists", return_value=True),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.render_audio.probe_has_audio", return_value=True),
+            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
+        ):
+            result = mix_audio(_ctx(), _layers(svc, clips), 5.0)
+
+        assert result is not None
+        cmd_str = " ".join(mock_run.call_args[0][0])
+        assert "amix=inputs=2" in cmd_str
+        assert "asset_c1.mp4" in cmd_str
+        assert "asset_tts1.mp4" in cmd_str
+        assert "volume=0.5" in cmd_str
+
+    def test_silent_clip_volume_zero_retained_with_silence_filter(self):
+        """volume=0 的素材必须保留在 concat 中（用 volume=0 滤镜静音），不能移除以避免音画不同步。"""
+        clips = [_clip("mute", order=0, config={"volume": 0})]
+        svc = _service(clips)
+        with (
+            patch("pathlib.Path.exists", return_value=True),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.render_audio.probe_has_audio", return_value=True),
+            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
+        ):
+            result = mix_audio(_ctx(), _layers(svc, clips), 5.0)
+
+        # 有音频流 → 应生成音频文件，且 ffmpeg 命令包含 volume=0.0000 静音滤镜
+        assert result is not None
+        mock_run.assert_called_once()
+        cmd_str = " ".join(mock_run.call_args[0][0])
+        assert "volume=0" in cmd_str
+
+    def test_no_audio_stream_returns_none(self):
+        clips = [_clip("c1")]
+        svc = _service(clips)
+        with (
+            patch("pathlib.Path.exists", return_value=True),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.render_audio.probe_has_audio", return_value=False),
+            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
+        ):
+            result = mix_audio(_ctx(), _layers(svc, clips), 5.0)
+
+        assert result is None
+        mock_run.assert_not_called()
+
+    def test_volume_helper_default_and_override(self):
+        assert _clip_volume(_clip("c1")) == 1.0
+        assert _clip_volume(_clip("c2", config={"volume": 0.3})) == pytest.approx(0.3)
+        assert _clip_volume(_clip("c3", config={"volume": 0})) == 0.0
+
+
+class TestPassThroughAudioProbe:
+    """直通路径必须先探测音频，不能无条件假设 main 有音频。"""
+
+    def test_pass_through_probes_audio_before_encoding(self):
+        clips = [_clip("c1")]
+        svc = _service(clips)
+        with (
+            patch("pathlib.Path.exists", return_value=True),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch(
+                "video_processing.unified_render_service.probe_has_audio",
+                return_value=False,
+            ) as mock_probe,
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(clips)
+            has_audio = svc._render_pass_through(layers, Path("/tmp/out.mp4"), video_duration=5.0)
+
+        assert has_audio is False
+        mock_probe.assert_called()
+        cmd_str = " ".join(mock_run.call_args[0][0])
+        assert "aac" not in cmd_str
+
+    def test_pass_through_volume_non_default_disables_stream_copy(self):
+        svc = _service([_clip("c1", config={"volume": 0.5})])
+        clip = _clip("c1", config={"volume": 0.5})
+        can_copy, reason = svc._can_use_stream_copy(clip)
+        assert can_copy is False
+        assert "音量" in reason
+
+    def test_clip_volume_static_helper(self):
+        assert UnifiedRenderService._clip_volume(_clip("c1")) == 1.0
+        assert UnifiedRenderService._clip_volume(_clip("c2", config={"volume": 0.7})) == pytest.approx(0.7)
+
+    def test_pass_through_probe_exception_raises(self):
+        """probe_has_audio 抛致命异常时必须向上抛出，不能静默丢音频产出无声视频。"""
+        clips = [_clip("c1")]
+        svc = _service(clips)
+        with (
+            patch("pathlib.Path.exists", return_value=True),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch(
+                "video_processing.unified_render_service.probe_has_audio",
+                side_effect=RuntimeError("probe failed"),
+            ),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+            pytest.raises(RuntimeError, match="probe failed"),
+        ):
+            layers = svc._group_clips_into_layers(clips)
+            svc._render_pass_through(layers, Path("/tmp/out.mp4"), video_duration=5.0)
+
+        mock_run.assert_not_called()

tests/unit/test_templates_editor_api.py

@@ -1,12 +1,11 @@
 """
 templates_editor.py 模板编辑器 API 端点单元测试
 
-覆盖核心端点（25个测试用例）：
+覆盖核心端点（23个测试用例）：
 - 草稿：GET/PUT/发布
 - 片段：list/create/get/update/delete/split/merge
 - BGM：GET/PUT
 - 时间线：GET
-- 生成状态查询
 - 预设：BGM预设
 """
 
@@ -395,30 +394,6 @@ class TestTimelineRoute:
         mock_plan_svc.list_clips.assert_called()
 
 
-# ---------------------------------------------------------------------------
-# 生成端点测试
-# ---------------------------------------------------------------------------
-
-
-class TestGenerationRoutes:
-    """生成端点测试"""
-
-    def test_generation_status_success(self, client):
-        c, _, mock_plan_svc = client
-        resp = c.get(BASE + "/generation-status")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert "generation_task_id" in data
-        assert "clips" in data
-
-    def test_generations_list_success(self, client):
-        c, _, mock_plan_svc = client
-        resp = c.get(BASE + "/generations")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert "items" in data or "tasks" in data or isinstance(data, dict)
-
-
 # ---------------------------------------------------------------------------
 # 字幕端点测试
 # ---------------------------------------------------------------------------

tests/unit/test_three_bugs_fix.py

@@ -0,0 +1,404 @@
+"""Regression tests for 3 bug fixes: flush, append_log, plan_id fallback."""
+
+from __future__ import annotations
+
+import json
+import os
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+
+
+# ── Bug 1: db.flush() before pending query ─────────────────────────────────
+
+
+class TestReplaceAllClipsFlush:
+    """replace_all_clips_transactional must flush before querying pending clips."""
+
+    def _make_svc_and_db(self, pending_results):
+        """Helper: create service + db mock. pending_results = list returned by pending query."""
+        from app.services.edit_plan_service import EditPlanService
+
+        db = MagicMock()
+        # Delete query
+        delete_query = MagicMock()
+        delete_query.filter.return_value.delete.return_value = 0
+        # Pending query: single .filter() with multiple conditions
+        ready_query = MagicMock()
+        ready_query.filter.return_value.all.return_value = pending_results
+        db.query.side_effect = [delete_query, ready_query]
+        return db, EditPlanService
+
+    def _setup_clip_mocks(self, mock_clip_cls, mock_model_cls, asset_id="asset-1"):
+        mock_entity = MagicMock()
+        mock_entity.id = "clip-1"
+        mock_entity.plan_id = "plan-1"
+        mock_entity.clip_type = "main"
+        mock_entity.order = 0
+        mock_entity.asset_id = asset_id
+        mock_entity.text_content = ""
+        mock_entity.start_time = 0.0
+        mock_entity.duration = 3.0
+        mock_entity.transition_effect = "cut"
+        mock_entity.transition_duration = 0.0
+        mock_entity.playback_speed = 1.0
+        mock_entity.status.value = "pending"
+        mock_entity.config = {}
+        mock_clip_cls.create.return_value = mock_entity
+        mock_model_cls.return_value = MagicMock()
+        return mock_entity
+
+    @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel")
+    @patch("app.services.edit_plan_service.EditPlanClip")
+    def test_flush_called_between_add_and_query(self, mock_clip_cls, mock_model_cls):
+        """db.flush() must be called after db.add() and before the pending query."""
+        self._setup_clip_mocks(mock_clip_cls, mock_model_cls)
+        db, SvcClass = self._make_svc_and_db([])
+
+        clip_repo = MagicMock()
+        clip_repo.session = db
+        svc = SvcClass.__new__(SvcClass)
+        svc._clip_repo = clip_repo
+
+        svc.replace_all_clips_transactional(
+            "plan-1",
+            [{"asset_id": "asset-1", "start_time": 0.0, "duration": 3.0, "order": 0}],
+        )
+
+        db.flush.assert_called_once()
+        # Verify ordering: add → flush → query → commit
+        method_names = [c[0] for c in db.method_calls]
+        add_idx = method_names.index("add")
+        flush_idx = method_names.index("flush")
+        commit_idx = method_names.index("commit")
+        assert add_idx < flush_idx < commit_idx
+
+    @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel")
+    @patch("app.services.edit_plan_service.EditPlanClip")
+    def test_flush_marks_new_clips_ready(self, mock_clip_cls, mock_model_cls):
+        """After flush, new clips with asset_id are found and marked ready."""
+        self._setup_clip_mocks(mock_clip_cls, mock_model_cls)
+
+        # Use a plain object so we can verify attribute mutation
+        class FakeClip:
+            status = "pending"
+
+        pending_clip = FakeClip()
+        db, SvcClass = self._make_svc_and_db([pending_clip])
+
+        clip_repo = MagicMock()
+        clip_repo.session = db
+        svc = SvcClass.__new__(SvcClass)
+        svc._clip_repo = clip_repo
+
+        svc.replace_all_clips_transactional(
+            "plan-1",
+            [{"asset_id": "asset-1", "start_time": 0.0, "duration": 3.0, "order": 0}],
+        )
+
+        assert pending_clip.status == "ready"
+
+
+# ── Bug 2: append_log no TypeError ─────────────────────────────────────────
+
+
+class TestAppendLogNoConflict:
+    """append_log must not receive duplicate 'stage' parameter."""
+
+    def _make_task(self):
+        from packages.domain.generation_task import GenerationTask
+
+        return GenerationTask.create(
+            project_id="proj-1",
+            asset_library_id="lib-1",
+            strategy_id="one_take",
+            template_id="tpl-1",
+            asset_ids=["a1"],
+            created_by_user_id="user-1",
+        )
+
+    def test_append_log_with_stage_as_first_positional(self):
+        """append_log(stage, message, ...) works correctly."""
+        task = self._make_task()
+        task.append_log("render", "some error", level="ERROR", error_type="RuntimeError")
+
+        entries = json.loads(task.logs)
+        assert len(entries) == 1
+        assert entries[0]["stage"] == "render"
+        assert entries[0]["message"] == "some error"
+        assert entries[0]["level"] == "ERROR"
+        assert entries[0]["error_type"] == "RuntimeError"
+
+    def test_duplicate_stage_raises_type_error(self):
+        """Sanity check: passing stage both positionally and as kwarg raises TypeError."""
+        task = self._make_task()
+        with pytest.raises(TypeError):
+            task.append_log(
+                "任务失败",  # positional → stage
+                "some error",
+                level="ERROR",
+                stage="render",  # duplicate → TypeError
+            )
+
+
+# ── Bug 3: plan_id fallback in create_generation_task ──────────────────────
+
+
+class TestPlanIdFallback:
+    """Formal generation API should fallback to find plan by template_id + user_id."""
+
+    def test_fallback_code_present_in_source(self):
+        """Verify the fallback logic is present in the generation_tasks module."""
+        import inspect
+
+        from apps.api.app.api.routes import generation_tasks
+
+        source = inspect.getsource(generation_tasks.create_generation_task)
+        assert "EditPlanModel" in source
+        assert "兜底关联编辑计划" in source
+        assert "自动关联编辑计划" in source
+
+    def test_fallback_only_runs_when_source_edit_plan_id_empty(self):
+        """Verify the condition checks for empty source_edit_plan_id."""
+        import inspect
+
+        from apps.api.app.api.routes import generation_tasks
+
+        source = inspect.getsource(generation_tasks.create_generation_task)
+        assert "not task.source_edit_plan_id and request.template_id" in source
+
+    def test_list_by_template_method_exists(self):
+        """Verify SQLAlchemyEditPlanRepository.list_by_template is callable."""
+        from packages.adapters.sqlalchemy_impl.edit_plan_repository import (
+            SQLAlchemyEditPlanRepository,
+        )
+
+        mock_db = MagicMock()
+        mock_session = MagicMock()
+        mock_db.query.return_value = mock_session
+        mock_session.filter.return_value = mock_session
+        mock_session.order_by.return_value = mock_session
+        mock_session.offset.return_value = mock_session
+        mock_session.limit.return_value.all.return_value = []
+
+        repo = SQLAlchemyEditPlanRepository(mock_db)
+        result = repo.list_by_template("tpl-1", limit=20)
+        assert result == []
+
+
+class TestPlanIdFallbackExecution:
+    """Test that the fallback logic actually executes when source_edit_plan_id is empty."""
+
+    @staticmethod
+    def _make_mock_task(source_edit_plan_id=""):
+        t = MagicMock()
+        t.id = "task-1"
+        t.project_id = "proj-1"
+        t.asset_library_id = ""
+        t.strategy_id = "one_take"
+        t.voice_library_id = ""
+        t.template_id = "tpl-1"
+        t.asset_ids = []
+        t.title_ids = []
+        t.voice_ids = []
+        t.source_edit_plan_id = source_edit_plan_id
+        t.asset_select_mode = "manual"
+        t.batch_id = ""
+        t.video_title = ""
+        t.resolution = ""
+        t.bgm_config = None
+        t.is_preview = False
+        t.source_task_id = ""
+        t.output_width = 1280
+        t.output_height = 720
+        t.cover_url = ""
+        t.custom_title = ""
+        t.title_config = {}
+        t.logs = "[]"
+        t.status = "pending"
+        t.progress = 0.0
+        t.error_message = ""
+        t.error_info = None
+        t.created_at = "2026-01-01T00:00:00Z"
+        t.updated_at = "2026-01-01T00:00:00Z"
+        t.started_at = None
+        t.completed_at = None
+        t.created_by_user_id = "user-1"
+        t.auto_retry_enabled = False
+        t.auto_retry_max = 0
+        t.auto_retry_count = 0
+        t.result_count = 0
+        return t
+
+    @staticmethod
+    def _make_request(source_edit_plan_id="", template_id="tpl-1"):
+        req = MagicMock()
+        req.template_id = template_id
+        req.source_edit_plan_id = source_edit_plan_id
+        req.asset_ids = []
+        req.asset_select_mode = "manual"
+        req.asset_select_count = 0
+        req.voice_library_id = ""
+        req.title_ids = []
+        req.voice_ids = []
+        req.strategy_id = "one_take"
+        req.count = 1
+        req.video_title = ""
+        req.resolution = ""
+        req.bgm_config = None
+        req.auto_retry_enabled = False
+        req.auto_retry_max = 0
+        req.is_preview = False
+        req.source_task_id = ""
+        req.output_width = 0
+        req.output_height = 0
+        req.cover_url = ""
+        req.custom_title = ""
+        req.title_config = {}
+        req.project_id = None
+        req.asset_library_id = None
+        return req
+
+    def _run_create_task(self, mock_task, mock_request, mock_db, mock_gen_repo):
+        """Helper to run create_generation_task with common mocks."""
+        from app.schemas.generation_task import GenerationTaskResponse
+
+        with (
+            patch(
+                "apps.api.app.api.routes.generation_tasks._resolve_project_and_library", return_value=("proj-1", None)
+            ),
+            patch("apps.api.app.api.routes.generation_tasks.CreateGenerationTaskUseCase") as mock_uc_cls,
+            patch("apps.api.app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True),
+            patch("apps.api.app.api.routes.generation_tasks._to_generation_task_response") as mock_resp_fn,
+        ):
+            mock_uc_cls.return_value.execute.return_value = mock_task
+            mock_resp_fn.return_value = GenerationTaskResponse(
+                id="task-1",
+                project_id="proj-1",
+                asset_library_id="",
+                strategy_id="one_take",
+                voice_library_id="",
+                template_id="tpl-1",
+                asset_ids=[],
+                title_ids=[],
+                voice_ids=[],
+                source_edit_plan_id="",
+                asset_select_mode="",
+                batch_id="",
+                video_title="",
+                resolution="",
+                bgm_config={},
+                is_preview=False,
+                source_task_id="",
+                output_width=1280,
+                output_height=720,
+                cover_url="",
+                custom_title="",
+                title_config={},
+                logs="[]",
+                status="pending",
+                progress=0.0,
+                error_message="",
+                created_at="2026-01-01T00:00:00Z",
+                updated_at="2026-01-01T00:00:00Z",
+                started_at=None,
+                completed_at=None,
+                created_by_user_id="user-1",
+                auto_retry_enabled=False,
+                auto_retry_max=0,
+                auto_retry_count=0,
+                result_count=0,
+            )
+
+            from apps.api.app.api.routes.generation_tasks import create_generation_task
+
+            return create_generation_task(
+                request=mock_request,
+                authenticated_user=MagicMock(user=MagicMock(id="user-1")),
+                generation_task_repository=mock_gen_repo,
+                project_repository=MagicMock(),
+                asset_library_repository=MagicMock(),
+                asset_repository=MagicMock(),
+                db=mock_db,
+            )
+
+    def test_fallback_sets_plan_id_when_empty(self):
+        """When source_edit_plan_id is empty, fallback finds plan via DB query."""
+        mock_task = self._make_mock_task(source_edit_plan_id="")
+        mock_request = self._make_request(source_edit_plan_id="", template_id="tpl-1")
+
+        # Mock DB query chain: db.query(EditPlanModel).filter(...).order_by(...).first()
+        mock_plan_model = MagicMock()
+        mock_plan_model.id = "plan-found-123"
+
+        mock_db = MagicMock()
+        query_chain = MagicMock()
+        query_chain.filter.return_value = query_chain
+        query_chain.order_by.return_value = query_chain
+        query_chain.first.return_value = mock_plan_model
+        mock_db.query.return_value = query_chain
+
+        mock_gen_repo = MagicMock()
+        mock_gen_repo.count_pending_by_user.return_value = 0
+        mock_gen_repo.count_pending_total.return_value = 0
+
+        self._run_create_task(mock_task, mock_request, mock_db, mock_gen_repo)
+
+        assert mock_task.source_edit_plan_id == "plan-found-123"
+        mock_gen_repo.update.assert_called_once_with(mock_task)
+
+    def test_no_fallback_when_plan_id_already_set(self):
+        """When source_edit_plan_id is already set, fallback should NOT run."""
+        mock_task = self._make_mock_task(source_edit_plan_id="plan-already-set")
+        mock_request = self._make_request(source_edit_plan_id="plan-already-set", template_id="tpl-1")
+
+        mock_db = MagicMock()
+        mock_gen_repo = MagicMock()
+        mock_gen_repo.count_pending_by_user.return_value = 0
+        mock_gen_repo.count_pending_total.return_value = 0
+
+        self._run_create_task(mock_task, mock_request, mock_db, mock_gen_repo)
+
+        assert mock_task.source_edit_plan_id == "plan-already-set"
+        mock_gen_repo.update.assert_not_called()
+
+    def test_fallback_no_match_leaves_plan_id_empty(self):
+        """When no plan matches, source_edit_plan_id stays empty."""
+        mock_task = self._make_mock_task(source_edit_plan_id="")
+        mock_request = self._make_request(source_edit_plan_id="", template_id="tpl-1")
+
+        mock_db = MagicMock()
+        query_chain = MagicMock()
+        query_chain.filter.return_value = query_chain
+        query_chain.order_by.return_value = query_chain
+        query_chain.first.return_value = None  # no matching plan
+        mock_db.query.return_value = query_chain
+
+        mock_gen_repo = MagicMock()
+        mock_gen_repo.count_pending_by_user.return_value = 0
+        mock_gen_repo.count_pending_total.return_value = 0
+
+        self._run_create_task(mock_task, mock_request, mock_db, mock_gen_repo)
+
+        assert mock_task.source_edit_plan_id == ""
+        mock_gen_repo.update.assert_not_called()
+
+    def test_fallback_handles_exception_gracefully(self):
+        """When DB query fails, the fallback should not break the main flow."""
+        mock_task = self._make_mock_task(source_edit_plan_id="")
+        mock_request = self._make_request(source_edit_plan_id="", template_id="tpl-1")
+
+        mock_db = MagicMock()
+        mock_db.query.side_effect = Exception("DB connection error")
+
+        mock_gen_repo = MagicMock()
+        mock_gen_repo.count_pending_by_user.return_value = 0
+        mock_gen_repo.count_pending_total.return_value = 0
+
+        self._run_create_task(mock_task, mock_request, mock_db, mock_gen_repo)
+
+        # Task should still be created (fallback error doesn't break main flow)
+        assert mock_task.source_edit_plan_id == ""
+        mock_gen_repo.update.assert_not_called()

tests/unit/test_unified_render_service.py

@@ -60,6 +60,17 @@ class FakePlan:
     config: dict[str, Any] = field(default_factory=dict)
 
 
+@pytest.fixture(autouse=True)
+def _assume_source_clips_have_audio():
+    """默认假设测试中的视频素材都带音频流。
+
+    新行为：保留视频素材原声（不再无条件丢弃）。需要模拟无音频流的用例
+    自行 patch probe_has_audio=False（如 test_mix_audio_main_no_audio_stream_returns_none）。
+    """
+    with patch("video_processing.render_audio.probe_has_audio", return_value=True):
+        yield
+
+
 def _make_clip(
     clip_id: str,
     clip_type: str = "main",
@@ -976,7 +987,7 @@ class TestAudioMixing:
         assert UnifiedRenderService._clip_effective_duration(clip) == 0.0
 
     def test_mix_audio_single_main_clip(self):
-        """只有 main clip（无独立音频轨）→ 源视频音频被丢弃，返回 None。"""
+        """只有 main clip（无独立音频轨）→ 保留源视频原声，走单轨 concat。"""
         clips = [_make_clip("c1", "main", order=0, duration=5.0)]
         asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
         svc = _make_service(clips, asset_paths)
@@ -990,14 +1001,15 @@ class TestAudioMixing:
             ctx = _make_ctx()
             result = mix_audio(ctx, layers, 5.0)
 
-        # 源视频音频被丢弃，没有独立音频轨 → 无音频
-        assert result is None
-        mock_run.assert_not_called()
+        # 保留源视频原声
+        assert result is not None
+        mock_run.assert_called_once()
+        assert "asset_c1.mp4" in " ".join(mock_run.call_args[0][0])
 
     def test_mix_audio_multi_main_clips(self):
-        """多个独立音频轨用 concat 拼接（main 图层音频被丢弃）。"""
+        """main 原声与独立音频轨通过 amix 混音。"""
         clips = [
-            _make_clip("c1", "main", order=0, duration=3.0),  # 源视频音频被丢弃
+            _make_clip("c1", "main", order=0, duration=3.0),
             _make_clip("tts1", "main", order=0, duration=3.0, config={"role": "audio"}),
             _make_clip("tts2", "main", order=1, duration=2.0, config={"role": "audio"}),
         ]
@@ -1021,15 +1033,17 @@ class TestAudioMixing:
         mock_run.assert_called_once()
         cmd = mock_run.call_args[0][0]
         cmd_str = " ".join(cmd)
-        # 2 个独立音频轨 concat
-        assert "concat=n=2:v=0:a=1" in cmd_str
-        # main 的 c1 不参与音频（源视频杂音被丢弃）
-        assert "asset_c1.mp4" not in cmd_str
+        # 原声 + 2 个独立音频轨 → amix 混音（3 路输入）
+        assert "amix=inputs=3" in cmd_str
+        # main 原声 c1 参与混音
+        assert "asset_c1.mp4" in cmd_str
+        assert "asset_tts1.mp4" in cmd_str
+        assert "asset_tts2.mp4" in cmd_str
 
     def test_mix_audio_with_independent_audio_track(self):
-        """独立音频轨生效；main 图层源视频音频被丢弃。"""
+        """main 原声与独立音频轨通过 amix 混音。"""
         clips = [
-            _make_clip("c1", "main", order=0, duration=5.0),  # 源视频音频被丢弃
+            _make_clip("c1", "main", order=0, duration=5.0),
             _make_clip(
                 "bgm1",
                 "main",
@@ -1057,9 +1071,9 @@ class TestAudioMixing:
         mock_run.assert_called_once()
         cmd = mock_run.call_args[0][0]
         cmd_str = " ".join(cmd)
-        # main 的源视频 c1 不参与音频
-        assert "asset_c1.mp4" not in cmd_str
-        # 独立音频轨 bgm1 作为最终音频生效
+        # main 原声 c1 与独立音频轨 bgm1 都参与 amix
+        assert "amix=inputs=2" in cmd_str
+        assert "asset_c1.mp4" in cmd_str
         assert "asset_bgm1.mp4" in cmd_str
 
     def test_mix_with_independent_audio_amix(self):
@@ -1122,7 +1136,7 @@ class TestAudioMixing:
         assert result is None
 
     def test_mix_audio_background_not_used_as_main(self):
-        """main/background 图层音频都被丢弃，只有独立音频轨参与混音。"""
+        """background 图层不参与主音频；main 原声与独立音频轨混音。"""
         clips = [
             _make_clip("bg1", "background", order=0, duration=5.0),
             _make_clip("c1", "main", order=0, duration=5.0),
@@ -1148,14 +1162,15 @@ class TestAudioMixing:
         mock_run.assert_called_once()
         cmd = mock_run.call_args[0][0]
         cmd_str = " ".join(cmd)
-        # 源视频（background + main）音频都被丢弃
+        # background 图层不参与主音频
         assert "asset_bg1.mp4" not in cmd_str
-        assert "asset_c1.mp4" not in cmd_str
-        # 只有独立音频轨
+        # main 原声 + 独立音频轨都参与
+        assert "amix=inputs=2" in cmd_str
+        assert "asset_c1.mp4" in cmd_str
         assert "asset_tts1.mp4" in cmd_str
 
     def test_mix_audio_main_priority_over_broll(self):
-        """main/broll 图层的源视频音频都被丢弃，只使用独立音频轨。"""
+        """main 图层优先作为主音频，broll 不参与；与独立音频轨 amix。"""
         clips = [
             _make_clip("b1", "b_roll", order=0, duration=5.0),
             _make_clip("c1", "main", order=0, duration=5.0),
@@ -1181,13 +1196,14 @@ class TestAudioMixing:
         mock_run.assert_called_once()
         cmd = mock_run.call_args[0][0]
         cmd_str = " ".join(cmd)
-        # main 和 broll 的源视频音频都被丢弃
-        assert "asset_c1.mp4" not in cmd_str
+        # main 原声 c1 优先参与；broll b1 不参与主音频
+        assert "amix=inputs=2" in cmd_str
+        assert "asset_c1.mp4" in cmd_str
         assert "asset_b1.mp4" not in cmd_str
         assert "asset_tts1.mp4" in cmd_str
 
     def test_mix_audio_broll_used_when_no_main(self):
-        """broll 图层源视频音频也被丢弃；无独立音频轨 → 返回 None。"""
+        """无 main 图层时 broll 原声作为主音频。"""
         clips = [_make_clip("b1", "b_roll", order=0, duration=5.0)]
         asset_paths = {"asset_b1.mp4": Path("/tmp/asset_b1.mp4")}
         svc = _make_service(clips, asset_paths)
@@ -1201,14 +1217,15 @@ class TestAudioMixing:
             ctx = _make_ctx()
             result = mix_audio(ctx, layers, 5.0)
 
-        # 源视频音频被丢弃，无独立音频轨 → 无音频
-        assert result is None
-        mock_run.assert_not_called()
+        # 保留 broll 原声
+        assert result is not None
+        mock_run.assert_called_once()
+        assert "asset_b1.mp4" in " ".join(mock_run.call_args[0][0])
 
     def test_mix_audio_single_clip_truncated_to_video_duration(self):
-        """单独立音频轨截断到 video_duration（video_duration < clip有效时长）。"""
+        """主音频截断到 video_duration（video_duration < clip有效时长）。"""
         clips = [
-            _make_clip("c1", "main", order=0, duration=10.0),  # 源视频音频被丢弃
+            _make_clip("c1", "main", order=0, duration=10.0),
             _make_clip("tts1", "main", order=0, duration=10.0, config={"role": "audio"}),
         ]
         asset_paths = {
@@ -1233,8 +1250,8 @@ class TestAudioMixing:
         # 验证截断到 3.0（-t 3.0 或 atrim=0:3.000）
         cmd_str = " ".join(cmd)
         assert "3.000" in cmd_str or "3.0" in cmd_str
-        # 源视频不参与
-        assert "asset_c1.mp4" not in cmd_str
+        # main 原声参与
+        assert "asset_c1.mp4" in cmd_str
 
     def test_merge_audio_video(self):
         """合并音视频命令正确。"""
@@ -1391,11 +1408,11 @@ class TestAudioMixing:
         mock_run.assert_not_called()
 
     def test_mix_audio_partial_clips_no_audio_filtered(self):
-        """main 图层音频全部丢弃；独立音频轨有/无音频时按预期过滤。"""
+        """main 原声正常保留；无音频流的 clip 被过滤。"""
         clips = [
-            _make_clip("c1", "main", order=0, duration=3.0),  # 源视频音频被丢弃
-            _make_clip("c2", "main", order=1, duration=2.0),  # 源视频音频被丢弃
-            _make_clip("tts1", "main", order=0, duration=2.0, config={"role": "audio"}),  # 独立音频轨
+            _make_clip("c1", "main", order=0, duration=3.0),
+            _make_clip("c2", "main", order=1, duration=2.0),
+            _make_clip("tts1", "main", order=0, duration=2.0, config={"role": "audio"}),
         ]
         asset_paths = {
             "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
@@ -1417,16 +1434,15 @@ class TestAudioMixing:
         mock_run.assert_called_once()
         cmd = mock_run.call_args[0][0]
         cmd_str = " ".join(cmd)
-        # 只有独立音频轨 tts1 参与
+        # main 原声 c1/c2 + 独立音频轨 tts1 全部参与
+        assert "asset_c1.mp4" in cmd_str
+        assert "asset_c2.mp4" in cmd_str
         assert "asset_tts1.mp4" in cmd_str
-        # main 的 c1/c2 源视频音频被丢弃
-        assert "asset_c1.mp4" not in cmd_str
-        assert "asset_c2.mp4" not in cmd_str
 
-    def test_mix_audio_all_main_no_audio_but_independent_track(self):
-        """main 图层源视频音频全部丢弃；仅独立音频轨生效，走 concat 单轨路径。"""
+    def test_mix_audio_main_plus_independent_amix(self):
+        """main 原声与独立音频轨 amix 混音。"""
         clips = [
-            _make_clip("c1", "main", order=0, duration=5.0),  # 源视频音频被丢弃
+            _make_clip("c1", "main", order=0, duration=5.0),
             _make_clip(
                 "bgm1",
                 "main",
@@ -1454,9 +1470,9 @@ class TestAudioMixing:
         mock_run.assert_called_once()
         cmd = mock_run.call_args[0][0]
         cmd_str = " ".join(cmd)
-        # main 的源视频 c1 不参与音频
-        assert "asset_c1.mp4" not in cmd_str
-        # 只有独立音频轨 bgm1 作为主音频走单轨拼接
+        # main 原声 c1 与独立音频轨 bgm1 都参与 amix
+        assert "amix=inputs=2" in cmd_str
+        assert "asset_c1.mp4" in cmd_str
         assert "asset_bgm1.mp4" in cmd_str
 
     def test_mix_audio_both_no_audio_returns_none(self):
@@ -2116,9 +2132,9 @@ class TestConcatNormalizeAudioFormat:
     """
 
     def test_multi_clip_concat_has_aformat(self):
-        """多独立音频轨 concat 前，每个轨都有 aformat 归一化（main 图层源视频音频被丢弃）。"""
+        """main 原声 + 独立音频轨在 concat/amix 前都有 aformat 归一化。"""
         clips = [
-            _make_clip("c1", "main", order=0, duration=3.0),  # 源视频音频被丢弃
+            _make_clip("c1", "main", order=0, duration=3.0),
             _make_clip("tts1", "main", order=0, duration=3.0, config={"role": "audio"}),
             _make_clip("tts2", "main", order=1, duration=2.0, config={"role": "audio"}),
         ]
@@ -2148,14 +2164,14 @@ class TestConcatNormalizeAudioFormat:
         assert "channel_layouts=stereo" in cmd_str, "声道应统一为 stereo"
         assert "sample_fmts=fltp" in cmd_str, "采样格式应统一为 fltp"
 
-        # 2 个独立音频轨都应有 aformat
+        # main 原声 + 2 个独立音频轨都应有 aformat
         aformat_count = cmd_str.count("aformat=")
-        assert aformat_count >= 2, f"每个独立音频轨都应有 aformat，实际 {aformat_count} 个"
+        assert aformat_count >= 3, f"3 路音频都应有 aformat，实际 {aformat_count} 个"
 
-        # 有 concat
-        assert "concat=n=2:v=0:a=1" in cmd_str
-        # main 源视频不参与
-        assert "asset_c1.mp4" not in cmd_str
+        # amix 3 路输入
+        assert "amix=inputs=3" in cmd_str
+        # main 源视频原声参与
+        assert "asset_c1.mp4" in cmd_str
 
     def test_aformat_before_concat(self):
         """aformat 应在 concat 之前（每个独立音频轨处理链中 aformat 在 concat 之前）。"""
@@ -2190,9 +2206,9 @@ class TestConcatNormalizeAudioFormat:
         assert aformat_before_count >= 2, f"concat 之前每个独立音频轨都应有 aformat，实际 {aformat_before_count} 个"
 
     def test_single_clip_audio_has_normalized_output(self):
-        """单独立音频轨输出也应统一格式（一致性保障）。"""
+        """原声+独立音频轨输出统一格式（一致性保障）。"""
         clips = [
-            _make_clip("c1", "main", order=0, duration=5.0),  # 源视频音频被丢弃
+            _make_clip("c1", "main", order=0, duration=5.0),
             _make_clip("tts1", "main", order=0, duration=5.0, config={"role": "audio"}),
         ]
         asset_paths = {
@@ -2212,21 +2228,20 @@ class TestConcatNormalizeAudioFormat:
 
         assert mock_run.called
         cmd = mock_run.call_args[0][0]
-
-        # 单独立音频轨简单路径应有 -ar 48000 和 -ac 2
-        assert "-ar" in cmd, "单独立音频轨应指定采样率"
-        ar_idx = cmd.index("-ar")
-        assert cmd[ar_idx + 1] == "48000", "采样率应为 48000"
-        assert "-ac" in cmd, "单独立音频轨应指定声道数"
-        ac_idx = cmd.index("-ac")
-        assert cmd[ac_idx + 1] == "2", "声道数应为 2（stereo）"
         cmd_str = " ".join(cmd)
-        assert "asset_c1.mp4" not in cmd_str
+
+        # 原声 + 独立音频轨走 amix：两路都有 aformat 归一化
+        assert "amix=inputs=2" in cmd_str
+        assert cmd_str.count("aformat=") >= 2
+        assert "sample_rates=48000" in cmd_str
+        assert "channel_layouts=stereo" in cmd_str
+        assert "asset_c1.mp4" in cmd_str
+        assert "asset_tts1.mp4" in cmd_str
 
     def test_independent_audio_track_has_aformat(self):
-        """独立音频轨输出也应统一格式（48000Hz + stereo + aac）。"""
+        """原声+独立音频轨输出统一格式（48000Hz + stereo + aac）。"""
         clips = [
-            _make_clip("c1", "main", order=0, duration=5.0),  # 源视频音频被丢弃
+            _make_clip("c1", "main", order=0, duration=5.0),
             _make_clip(
                 "audio1",
                 "main",
@@ -2254,15 +2269,13 @@ class TestConcatNormalizeAudioFormat:
         cmd = mock_run.call_args[0][0]
         cmd_str = " ".join(cmd)
 
-        # 单独立音频轨走 concat 单轨简单路径：-ar 48000 -ac 2
-        assert "-ar" in cmd
-        ar_idx = cmd.index("-ar")
-        assert cmd[ar_idx + 1] == "48000"
-        assert "-ac" in cmd
-        ac_idx = cmd.index("-ac")
-        assert cmd[ac_idx + 1] == "2"
-        # main 源视频 c1 不参与音频
-        assert "asset_c1.mp4" not in cmd_str
+        # 原声 + 独立音频轨走 amix：两路都有 aformat 归一化
+        assert "amix=inputs=2" in cmd_str
+        assert cmd_str.count("aformat=") >= 2
+        assert "sample_rates=48000" in cmd_str
+        assert "channel_layouts=stereo" in cmd_str
+        # main 原声 c1 与独立音频轨 audio1 都参与
+        assert "asset_c1.mp4" in cmd_str
         assert "asset_audio1.mp4" in cmd_str
 
 
@@ -2299,9 +2312,9 @@ class TestConcatNormalizeAudioCodec:
         assert "-b:a" in cmd, "应指定音频码率"
 
     def test_single_clip_output_is_aac(self):
-        """单独立音频轨输出编码为 aac。"""
+        """原声+独立音频轨输出编码为 aac。"""
         clips = [
-            _make_clip("c1", "main", order=0, duration=5.0),  # 源视频音频被丢弃
+            _make_clip("c1", "main", order=0, duration=5.0),
             _make_clip("tts1", "main", order=0, duration=5.0, config={"role": "audio"}),
         ]
         asset_paths = {
@@ -2322,7 +2335,8 @@ class TestConcatNormalizeAudioCodec:
         assert mock_run.called
         cmd = mock_run.call_args[0][0]
         assert "aac" in cmd
-        assert "asset_c1.mp4" not in " ".join(cmd)
+        # main 原声参与
+        assert "asset_c1.mp4" in " ".join(cmd)
 
 
 class TestConcatNormalizeFourItemsComplete:
@@ -2372,7 +2386,7 @@ class TestConcatNormalizeFourItemsComplete:
         assert fps_count >= 3, f"3个 clip 都应有 fps=30，实际 {fps_count} 个"
 
     def test_audio_format_normalized(self):
-        """[3/4] 音频格式：3 个独立音频轨 concat 前都有 aformat 归一化（main 图层源视频音频被丢弃）。"""
+        """[3/4] 音频格式：3 个 main 原声 + 3 个独立音频轨都有 aformat 归一化。"""
         clips = self._make_one_take_clips() + [
             _make_clip("tts1", "main", order=10, duration=5.0, config={"role": "audio"}),
             _make_clip("tts2", "main", order=11, duration=4.0, config={"role": "audio"}),
@@ -2396,13 +2410,13 @@ class TestConcatNormalizeFourItemsComplete:
         cmd_str = " ".join(cmd)
 
         aformat_count = cmd_str.count("aformat=")
-        assert aformat_count >= 3, f"3个独立音频轨都应有 aformat，实际 {aformat_count} 个"
+        assert aformat_count >= 6, f"6 路音频（3原声+3独立轨）都应有 aformat，实际 {aformat_count} 个"
         assert "sample_rates=48000" in cmd_str
         assert "channel_layouts=stereo" in cmd_str
         assert "sample_fmts=fltp" in cmd_str
-        # main 图层源视频不参与
+        # main 图层原声参与
         for i in range(1, 4):
-            assert f"asset_c{i}.mp4" not in cmd_str
+            assert f"asset_c{i}.mp4" in cmd_str
 
     def test_audio_codec_aac(self):
         """[4/4] 音频编码：输出为 aac（仅独立音频轨参与）。"""
@@ -2427,6 +2441,7 @@ class TestConcatNormalizeFourItemsComplete:
         assert mock_run.called
         cmd = mock_run.call_args[0][0]
         assert "aac" in cmd
+        # 3 个 main 原声 concat 后再与独立轨 amix
         assert "concat=n=3:v=0:a=1" in " ".join(cmd)
 
 

tests/unit/test_worker_generate_video_task_binding.py

@@ -6,28 +6,63 @@ _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.
+
+NOTE: CI conftest may mock Celery so that @celery_app.task does NOT return
+a fully functional Task/PromiseProxy object.  Tests therefore use multiple
+defensive strategies: source-code inspection, __wrapped__.__func__ chain
+traversal, and direct attribute checks.
 """
 
 from __future__ import annotations
 
 import inspect
+import re
 
 
-def test_generate_video_task_registered_under_expected_name():
+def _get_original_function(generate_video):
+    """Walk the __wrapped__ chain to find the original function object."""
+    obj = generate_video
+    seen = set()
+    while hasattr(obj, "__wrapped__"):
+        obj_id = id(obj)
+        if obj_id in seen:
+            break
+        seen.add(obj_id)
+        obj = obj.__wrapped__
+        # __wrapped__ may be a bound method — unwrap to the underlying function
+        if hasattr(obj, "__func__"):
+            return obj.__func__
+    return obj
+
+
+def test_generate_video_task_has_bind_true():
+    """The decorator must use bind=True — verified via the original function's
+    first parameter being 'self' (bind=True convention)."""
     from worker_app.tasks.generation import generate_video
 
-    # Celery task object exposes its registered name
-    assert generate_video.name == "worker.generate_video"
+    original = _get_original_function(generate_video)
+    sig = inspect.signature(original)
+    params = list(sig.parameters)
+    assert params[0] == "self", f"bind=True requires 'self' as first param, got {params}"
 
 
 def test_generate_video_task_signature_has_task_id():
+    """The original generate_video function must accept task_id as a parameter."""
     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)
+    original = _get_original_function(generate_video)
+    sig = inspect.signature(original)
     params = list(sig.parameters)
-    assert params[0] == "task_id", f"expected task_id as first param, got {params}"
+    assert "task_id" in params, f"expected 'task_id' in params, got {params}"
+
+
+def test_generate_video_preserves_original_function():
+    """The original function wrapped by @celery_app.task must be named
+    'generate_video' — not '_sync_task_config_to_plan'."""
+    from worker_app.tasks.generation import generate_video
+
+    original = _get_original_function(generate_video)
+    assert original.__name__ == "generate_video", f"expected __name__='generate_video', got '{original.__name__}'"
 
 
 def test_sync_task_config_to_plan_is_plain_function():