merge: develop(FFmpeg超时保护) into stream-copy分支

当输入片段与输出参数完全一致（h264/yuv420p/同分辨率/同帧率/无字幕/无特效）时，
走 FFmpeg stream copy 不重编码，性能提升 10 倍以上（典型场景 20s → 1-2s）。

核心改动：
1. probe_video_info 增强：返回编码/像素格式/音频信息（用于 copy 条件判断）
2. _can_use_stream_copy：7项条件检查（编码/分辨率/帧率/像素格式/字幕/trim等）
3. _try_render_stream_copy：stream copy 执行，失败自动回退到重编码
4. 单clip直通场景先尝试 stream copy，不满足或失败再回退带滤镜的直通渲染
5. trim 用 -ss/-t 实现，无需滤镜，copy 模式下也能用

安全保障：
- 条件不满足自动跳过，不影响现有渲染质量
- FFmpeg 失败自动回退到重编码，不影响成功率
- 输出文件为空时判定失败并清理损坏文件

测试：9个新增单测 + 现有75个统一渲染测试，全绿 ✅
（含条件判断、成功路径、失败回退、完整渲染流程等场景）

.gitea/workflows/ci-cd.yml

@@ -158,72 +158,57 @@ jobs:
             python3 scripts/check_migration_safety.py --allow-medium-risk
           fi
 
-  unit-tests:
-    name: Unit Tests
-    runs-on: host
-    timeout-minutes: 8
-
-    env:
-      USE_IN_MEMORY_DB: "true"
-
-    steps:
-      - name: Checkout code
+      - name: Debug coverage paths
         shell: sh
-        env:
-          GITHUB_TOKEN: ${{ github.token }}
         run: |
-          set -eu
+          set +e
+          echo "=== PWD ==="
+          pwd
+          echo "=== check source dirs ==="
+          ls -d apps/api/app packages
+          echo "=== python import check ==="
           python3 - <<'PY'
-          import io, os, tarfile, time, urllib.request, urllib.error
-          url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-          request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-          last_err = None
-          for attempt in range(5):
-              try:
-                  with urllib.request.urlopen(request, timeout=120) as response:
-                      archive = response.read()
-                  break
-              except urllib.error.HTTPError as e:
-                  last_err = e
-                  if e.code >= 500 and attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-              except Exception as e:
-                  last_err = e
-                  if attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-          else:
-              raise last_err
-          with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-              root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-              for member in tar.getmembers():
-                  name = member.name
-                  if name == root_prefix[:-1]:
-                      continue
-                  if name.startswith(root_prefix):
-                      member.name = name[len(root_prefix):]
-                  if member.name:
-                      tar.extract(member, '.')
+          import sys, os
+          os.environ["PYTHONPATH"] = f"{os.getcwd()}/apps/api:{os.getcwd()}"
+          sys.path.insert(0, f"{os.getcwd()}/apps/api")
+          sys.path.insert(0, os.getcwd())
+          print(f"cwd: {os.getcwd()}")
+          print(f"sys.path[:5]: {sys.path[:5]}")
+          try:
+              import app
+              print(f"app.__file__: {app.__file__}")
+          except Exception as e:
+              print(f"import app failed: {e}")
+          try:
+              import packages
+              print(f"packages.__file__: {packages.__file__}")
+          except Exception as e:
+              print(f"import packages failed: {e}")
+          PY
+          echo "=== coverage debug ==="
+          python3 - <<'PY'
+          import os, sys
+          sys.path.insert(0, f"{os.getcwd()}/apps/api")
+          sys.path.insert(0, os.getcwd())
+          import coverage
+          cov = coverage.Coverage(source=["apps/api/app", "packages"])
+          print(f"source: {cov.config.source}")
+          for src in cov.config.source or []:
+              abspath = os.path.abspath(src)
+              print(f"  {src} -> {abspath} exists={os.path.exists(src)}")
+              if os.path.isdir(src):
+                  pyfiles = []
+                  for root, dirs, files in os.walk(src):
+                      for f in files:
+                          if f.endswith('.py'):
+                              pyfiles.append(os.path.join(root, f))
+                  print(f"    .py files: {len(pyfiles)}")
           PY
 
-      - name: Install dependencies
-        shell: sh
-        run: |
-          set -eu
-          python3 -m pip install -q -r requirements-base.txt
-          python3 -m pip install -q -r requirements.txt
-          python3 -m pip install -q -r requirements-dev.txt
-          pytest --version
-
-      - name: Run unit tests with coverage
+      - name: Run unit tests
         shell: sh
+        env:
+          USE_IN_MEMORY_DB: "true"
         run: |
           set -eu
           PYTHONPATH="$PWD/apps/api:$PWD" python3 -m coverage run \
@@ -235,16 +220,16 @@ jobs:
           python3 -m coverage xml -o coverage.xml
           python3 -m coverage report --fail-under=60 > /dev/null
 
-      - name: CI failure notification
-        if: failure()
+      - name: Build summary
+        if: github.ref == 'refs/heads/develop' || github.ref == 'refs/heads/main'
         shell: sh
-        env:
-          GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }}
-          CI_WEBHOOK_URL: ${{ secrets.CI_WEBHOOK_URL }}
         run: |
-          set +e
-          FAILED_JOB="Unit Tests" python3 scripts/ci_notify_failure.py
-
+          set -eu
+          echo "Build completed successfully!"
+          echo "Branch: ${GITHUB_REF_NAME}"
+          echo "Commit: ${GITHUB_SHA}"
+          # 输出最终覆盖率
+          python3 scripts/ci_coverage_summary.py
 
   integration-tests:
     name: Integration Tests
@@ -593,7 +578,7 @@ jobs:
     timeout-minutes: 30
     needs: [validate, frontend-lint]
 
-    if: github.event_name == 'push' && (github.ref_name == 'main' || github.ref_name == 'develop')
+    if: github.ref_name == 'main' || github.ref_name == 'develop' || startsWith(github.ref_name, 'feature/')
 
     steps:
     - name: Checkout code

apps/worker/worker_app/tasks/generation.py

@@ -113,7 +113,6 @@ from video_processing.oss_helpers import (
     get_signed_download_url,
     upload_to_oss,
 )
-from video_processing.render_engine_resolver import ENGINE_LEGACY, ENGINE_UNIFIED
 from video_processing.unified_render_service import UnifiedRenderService
 
 # ── 虚拟 Plan / Clip（内存中构建，不写数据库） ────────────────────────────────
@@ -574,148 +573,6 @@ def _validate_template_exists(template_id: str) -> None:
         session.close()
 
 
-# ── 渲染引擎选择 ─────────────────────────────────────────────────────────────
-
-
-def _resolve_render_engine(user_id: str) -> str:
-    """根据 Feature Flag 决定使用哪个渲染引擎。
-
-    Returns:
-        "legacy" 或 "unified"
-    """
-    try:
-        from video_processing.render_engine_resolver import get_render_engine_resolver
-
-        resolver = get_render_engine_resolver()
-        return resolver.get_engine(user_id=user_id)
-    except Exception as exc:
-        logger.warning("获取渲染引擎配置失败，fallback 到 unified: %s", exc)
-        return ENGINE_UNIFIED
-
-
-# ── 旧引擎渲染（FFmpeg filter_complex） ────────────────────────────────────────
-
-
-def _render_with_legacy_engine(
-    task_id: str,
-    virtual_clips: list[_VirtualClip],
-    asset_path_map: dict[str, Path],
-    work_dir: Path,
-    output_path: Path,
-) -> tuple[float, int]:
-    """旧引擎渲染路径：手动构建 FFmpeg filter_complex 命令。
-
-    说明：generate_video 任务使用虚拟 clips（无 EditPlan 数据库记录），
-    因此无法直接复用 VideoComposeService。这里手动构建等价的 filter_complex
-    命令，与旧引擎行为一致（scale → crop → setpts → trim → setpts，
-    无 fps 归一化，保持原帧率）。
-
-    支持模式：one_take / pip / voice_over / voice_pip
-    - 所有模式统一走 concat 滤镜（与旧引擎多片段逻辑一致）
-
-    Returns:
-        (duration_seconds, file_size_bytes)
-    """
-    import subprocess
-
-    main_clips = [
-        c
-        for c in virtual_clips
-        if c.clip_type in ("main", "b_roll", "background")
-        or (c.clip_type == "main" and c.config.get("role") == "b_roll")
-    ]
-    if not main_clips:
-        main_clips = virtual_clips[:1]
-
-    input_args: list[str] = []
-    video_filters: list[str] = []
-    audio_filters: list[str] = []
-
-    for i, clip in enumerate(main_clips):
-        local_path = asset_path_map.get(clip.asset_id)
-        if not local_path:
-            continue
-        input_args.extend(["-i", str(local_path)])
-
-        duration = clip.duration or 0.0
-
-        # 视频滤镜：scale → crop → setpts → trim → setpts（与旧引擎一致）
-        vf = (
-            f"[{i}:v]"
-            f"scale={OUTPUT_WIDTH}:{OUTPUT_HEIGHT}:force_original_aspect_ratio=increase,"
-            f"crop={OUTPUT_WIDTH}:{OUTPUT_HEIGHT},"
-            f"setpts=PTS-STARTPTS,"
-            f"trim=0:{duration:.3f},"
-            f"setpts=PTS-STARTPTS"
-            f"[v{i}]"
-        )
-        video_filters.append(vf)
-
-        # 音频滤镜：atrim → asetpts
-        af = f"[{i}:a]atrim=0:{duration:.3f},asetpts=PTS-STARTPTS[a{i}]"
-        audio_filters.append(af)
-
-    n = len(main_clips)
-
-    if n == 1:
-        video_label = "[v0]"
-        audio_label = "[a0]"
-    else:
-        # concat 视频
-        v_inputs = "".join(f"[v{i}]" for i in range(n))
-        video_filters.append(f"{v_inputs}concat=n={n}:v=1:a=0[outv]")
-        # concat 音频
-        a_inputs = "".join(f"[a{i}]" for i in range(n))
-        audio_filters.append(f"{a_inputs}concat=n={n}:v=0:a=1[outa]")
-        video_label = "[outv]"
-        audio_label = "[outa]"
-
-    # 组装 filter_complex
-    fc_parts = video_filters + audio_filters
-    filter_complex = ";".join(fc_parts)
-
-    command = [
-        FFMPEG_BIN,
-        "-y",
-        *input_args,
-        "-filter_complex",
-        filter_complex,
-        "-map",
-        video_label,
-        "-map",
-        audio_label,
-        "-c:v",
-        "libx264",
-        "-crf",
-        "23",
-        "-preset",
-        "medium",
-        "-c:a",
-        "aac",
-        "-b:a",
-        "192k",
-        "-movflags",
-        "+faststart",
-        str(output_path),
-    ]
-
-    logger.info("[task_id=%s] [渲染] legacy 引擎 FFmpeg 开始: clips=%d", task_id, n)
-    try:
-        run_ffmpeg(command)
-    except subprocess.CalledProcessError as e:
-        logger.error(
-            "[task_id=%s] [渲染] legacy 引擎 FFmpeg 失败: %s\nfilter_complex: %s",
-            task_id,
-            e,
-            filter_complex[:500],
-        )
-        raise
-
-    file_size = output_path.stat().st_size if output_path.exists() else 0
-    duration = probe_duration(output_path)
-    return duration, file_size
-
-
 # ── Celery Task ──────────────────────────────────────────────────────────────
 
 
@@ -873,59 +730,31 @@ def generate_video(self, task_id: str) -> dict:
                 )
                 _flush_logs(task_id, gen_task)
 
-            # 3. 根据 Feature Flag 选择渲染引擎
-            user_id = getattr(gen_task, "created_by_user_id", "") if gen_task else ""
-            engine = _resolve_render_engine(user_id) if user_id else ENGINE_UNIFIED
-            logger.info("[task_id=%s] [渲染] 引擎选择: %s (user_id=%s)", task_id, engine, user_id)
-
+            # 使用 UnifiedRenderService 渲染
+            logger.info("[task_id=%s] [渲染] FFmpeg 渲染开始", task_id)
             render_start = time.monotonic()
-            render_output_path = temp_path / f"rendered-{task_id}.mp4"
-
-            if engine == ENGINE_LEGACY:
-                # 旧引擎：filter_complex + concat（保持原帧率，无 fps 归一化）
-                render_duration, render_file_size = _render_with_legacy_engine(
-                    task_id=task_id,
-                    virtual_clips=virtual_clips,
-                    asset_path_map=asset_path_map,
-                    work_dir=temp_path,
-                    output_path=render_output_path,
-                )
-                render_elapsed = time.monotonic() - render_start
-                logger.info(
-                    "[task_id=%s] [渲染] legacy 引擎完成: 耗时=%.1fs, 时长=%.2fs",
-                    task_id,
-                    render_elapsed,
-                    render_duration,
-                )
-            else:
-                # 新引擎：UnifiedRenderService 图层架构
-                logger.info("[task_id=%s] [渲染] unified 引擎 FFmpeg 渲染开始", task_id)
-                render_service = UnifiedRenderService(
-                    plan=virtual_plan,
-                    clips=virtual_clips,
-                    asset_path_map=asset_path_map,
-                    work_dir=temp_path,
-                    output_width=OUTPUT_WIDTH,
-                    output_height=OUTPUT_HEIGHT,
-                    output_fps=int(OUTPUT_FPS),
-                )
-                render_result = render_service.render()
-                render_output_path = render_result.output_path
-                render_duration = render_result.duration
-                render_file_size = render_result.file_size
-                render_elapsed = time.monotonic() - render_start
-                logger.info(
-                    "[task_id=%s] [渲染] unified 引擎完成: 耗时=%.1fs",
-                    task_id,
-                    render_elapsed,
-                )
+            render_service = UnifiedRenderService(
+                plan=virtual_plan,
+                clips=virtual_clips,
+                asset_path_map=asset_path_map,
+                work_dir=temp_path,
+                output_width=OUTPUT_WIDTH,
+                output_height=OUTPUT_HEIGHT,
+                output_fps=int(OUTPUT_FPS),
+            )
+            render_result = render_service.render()
+            render_elapsed = time.monotonic() - render_start
+            logger.info(
+                "[task_id=%s] [渲染] FFmpeg 渲染完成: 耗时=%.1fs",
+                task_id,
+                render_elapsed,
+            )
 
             if gen_task:
                 gen_task.append_log(
                     "渲染",
-                    f"引擎={engine}, 耗时={render_elapsed:.1f}s",
+                    f"FFmpeg 渲染完成, 耗时={render_elapsed:.1f}s",
                     duration=round(render_elapsed, 2),
-                    engine=engine,
                 )
                 _flush_logs(task_id, gen_task)
 
@@ -933,14 +762,14 @@ def generate_video(self, task_id: str) -> dict:
             if audio_path:
                 final_path = temp_path / f"final-{task_id}.mp4"
                 try:
-                    _mux_audio_track(render_output_path, audio_path, final_path)
+                    _mux_audio_track(render_result.output_path, audio_path, final_path)
                     # 混音成功，使用混音后的文件
                     output_path = final_path
                 except Exception as mux_err:
                     logger.warning("[task_id=%s] [混音] 音频混合失败，使用无音频版本: %s", task_id, mux_err)
-                    output_path = render_output_path
+                    output_path = render_result.output_path
             else:
-                output_path = render_output_path
+                output_path = render_result.output_path
 
             file_size = output_path.stat().st_size
             duration = probe_duration(output_path)

scripts/build_release_images.sh

@@ -30,8 +30,7 @@ fi
 # ---- Registry 配置 ----
 REGISTRY="${REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
 CACHE_REGISTRY="${CACHE_REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
-# 主缓存 tag：develop 分支构建时写入，所有分支读取
-CACHE_TAG_PRIMARY="${CACHE_TAG:-develop}"
+CACHE_TAG="${CACHE_TAG:-release}"
 
 API_IMAGE="xiaoxia-saas-api:$VERSION"
 WORKER_IMAGE="xiaoxia-saas-worker:$VERSION"
@@ -46,7 +45,6 @@ REGISTRY_WEB="${REGISTRY}/xiaoxia-saas-web:$VERSION"
 
 USE_CACHE=0
 USE_PUSH=0
-CACHE_WRITE=0
 
 # 检查 buildx 和 Registry 认证
 if docker buildx version >/dev/null 2>&1; then
@@ -56,10 +54,7 @@ if docker buildx version >/dev/null 2>&1; then
   docker buildx use default 2>/dev/null || true
 fi
 
-# ---- 缓存读写策略（按分支隔离）----
-# 默认只读不写，防止 feature 分支污染主缓存
-# 只有 develop/main 分支才写回缓存
-BRANCH_NAME="${GITHUB_REF_NAME:-${CI_COMMIT_BRANCH:-unknown}}"
+echo "=== Building API image ==="
 if [ "$USE_CACHE" -eq 1 ]; then
   docker buildx build \
     --build-arg APP_VERSION="$VERSION" \
@@ -73,52 +68,6 @@ else
   docker build --pull=false --build-arg APP_VERSION="$VERSION" -f infra/docker/api.Dockerfile -t "$API_IMAGE" -t "$API_LATEST" .
 fi
 
-build_with_cache() {
-  # usage: build_with_cache <image_name> <dockerfile> <extra_args...>
-  IMG_NAME="$1"
-  DOCKERFILE="$2"
-  shift 2
-  EXTRA_ARGS="$*"
-
-  CACHE_FROM="type=registry,ref=${CACHE_REGISTRY}/${IMG_NAME}-cache:${CACHE_TAG_PRIMARY},ignore-error=true"
-
-  if [ "$CACHE_WRITE" -eq 1 ]; then
-    CACHE_TO="type=registry,ref=${CACHE_REGISTRY}/${IMG_NAME}-cache:${CACHE_TAG_PRIMARY},mode=max"
-    echo "  cache: read+write from ${CACHE_REGISTRY}/${IMG_NAME}-cache:${CACHE_TAG_PRIMARY}"
-  else
-    CACHE_TO=""
-    echo "  cache: read-only from ${CACHE_REGISTRY}/${IMG_NAME}-cache:${CACHE_TAG_PRIMARY}"
-  fi
-
-  if [ "$USE_CACHE" -eq 1 ]; then
-    if [ -n "$CACHE_TO" ]; then
-      docker buildx build \
-        $EXTRA_ARGS \
-        --cache-from "$CACHE_FROM" \
-        --cache-to "$CACHE_TO" \
-        -f "$DOCKERFILE" \
-        -t "$IMG_NAME:$VERSION" \
-        --load \
-        .
-    else
-      docker buildx build \
-        $EXTRA_ARGS \
-        --cache-from "$CACHE_FROM" \
-        -f "$DOCKERFILE" \
-        -t "$IMG_NAME:$VERSION" \
-        --load \
-        .
-    fi
-  else
-    docker build --pull=false $EXTRA_ARGS -f "$DOCKERFILE" -t "$IMG_NAME:$VERSION" .
-  fi
-}
-
-echo "=== Building API image ==="
-build_with_cache "api" "infra/docker/api.Dockerfile" \
-  "--build-arg APP_VERSION=$VERSION"
-docker tag "$API_IMAGE" "$API_LATEST"
-
 echo "=== Building Worker image ==="
 if [ "$USE_CACHE" -eq 1 ]; then
   docker buildx build \
@@ -134,16 +83,9 @@ else
 fi
 
 echo "=== Building Web image (with buildx cache) ==="
-# 先构建前端产物（使用持久化 npm 缓存卷）
-NPM_CACHE_VOLUME="xiaoxia-npm-cache"
-if ! docker volume inspect "$NPM_CACHE_VOLUME" >/dev/null 2>&1; then
-  docker volume create "$NPM_CACHE_VOLUME" >/dev/null
-  echo "  Created npm cache volume: $NPM_CACHE_VOLUME"
-fi
-
+# 先构建前端产物
 docker run --rm \
   -v "$PWD:/workspace" \
-  -v "$NPM_CACHE_VOLUME:/workspace/apps/web/node_modules" \
   -w /workspace/apps/web \
   docker.m.daocloud.io/library/node:20 \
   sh -lc "npm ci && npm run build"

tests/unit/test_ffmpeg_timeout_protection.py

@@ -79,7 +79,7 @@ class TestProbeVideoInfoTimeout:
         """正常情况应解析 ffprobe JSON 输出。"""
         fake_output = """
         {
-            "streams": [{"width": 1920, "height": 1080, "codec_type": "video", "r_frame_rate": "30/1", "duration": "10.5"}],
+            "streams": [{"width": 1920, "height": 1080, "r_frame_rate": "30/1", "duration": "10.5"}],
             "format": {"duration": "10.5"}
         }
         """

tests/unit/test_generation_feature_flag.py

@@ -1,338 +0,0 @@
-"""generate_video 任务 Feature Flag 灰度引擎选择单元测试.
-
-覆盖：
-- _resolve_render_engine 正常返回 unified / legacy
-- Feature Flag 不可用时 fallback 到 unified
-- 白名单 / 百分比 / 全局开关各场景
-- _render_with_legacy_engine 命令构建与输出验证
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-from datetime import datetime, timezone
-from types import ModuleType
-from typing import Any
-from unittest.mock import MagicMock, patch
-
-os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
-os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
-
-from pathlib import Path
-
-import pytest
-
-# ── Mock worker 模块以避免数据库连接 ──────────────────────────────────────────
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "worker"))
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
-
-_mock_db_mod = ModuleType("worker_app.db")
-_mock_db_mod.SessionLocal = MagicMock()
-sys.modules.setdefault("worker_app.db", _mock_db_mod)
-
-_mock_celery_mod = ModuleType("worker_app.celery_app")
-_mock_celery_app = MagicMock()
-_mock_celery_app.task = lambda **kwargs: lambda fn: fn
-_mock_celery_mod.celery_app = _mock_celery_app
-sys.modules.setdefault("worker_app.celery_app", _mock_celery_mod)
-
-# Mock worker_app.core.config 避免 settings 加载
-_mock_config_mod = ModuleType("worker_app.core.config")
-_mock_settings = MagicMock()
-_mock_settings.redis_url = None
-_mock_settings.render_engine = "unified"
-_mock_config_mod.get_settings = lambda: _mock_settings
-sys.modules.setdefault("worker_app.core", ModuleType("worker_app.core"))
-sys.modules.setdefault("worker_app.core.config", _mock_config_mod)
-
-
-# ── 测试用数据类 ──────────────────────────────────────────────────────────────
-
-
-class _TestClip:
-    def __init__(self, asset_id, duration=30.0, clip_type="main", config=None, order=0):
-        self.id = f"clip_{asset_id}"
-        self.plan_id = "test-plan"
-        self.clip_type = clip_type
-        self.order = order
-        self.asset_id = asset_id
-        self.duration = duration
-        self.config = config or {}
-        self.start_time = 0.0
-        self.transition_effect = "cut"
-
-
-# ── RenderEngineResolver 基础行为测试 ───────────────────────────────────────
-
-
-def test_resolver_unified_when_enabled_100_percent():
-    """flag 全局开启（percentage=100）时，返回 unified。"""
-    from video_processing.render_engine_resolver import RenderEngineResolver
-
-    from packages.adapters.redis.feature_flag_store import (
-        FeatureFlagConfig,
-        InMemoryFeatureFlagStore,
-    )
-
-    store = InMemoryFeatureFlagStore()
-    store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100))
-    resolver = RenderEngineResolver(default_engine="legacy", store=store)
-
-    assert resolver.get_engine(user_id="user-123") == "unified"
-
-
-def test_resolver_legacy_when_flag_disabled():
-    """flag 全局关闭时，返回默认引擎 legacy。"""
-    from video_processing.render_engine_resolver import RenderEngineResolver
-
-    from packages.adapters.redis.feature_flag_store import (
-        FeatureFlagConfig,
-        InMemoryFeatureFlagStore,
-    )
-
-    store = InMemoryFeatureFlagStore()
-    store.set(FeatureFlagConfig(name="render_engine", enabled=False, percentage=100))
-    resolver = RenderEngineResolver(default_engine="legacy", store=store)
-
-    assert resolver.get_engine(user_id="user-123") == "legacy"
-
-
-def test_resolver_whitelist_overrides_percentage_0():
-    """白名单用户即使 percentage=0 也走 unified。"""
-    from video_processing.render_engine_resolver import RenderEngineResolver
-
-    from packages.adapters.redis.feature_flag_store import (
-        FeatureFlagConfig,
-        InMemoryFeatureFlagStore,
-    )
-
-    store = InMemoryFeatureFlagStore()
-    store.set(
-        FeatureFlagConfig(
-            name="render_engine",
-            enabled=True,
-            percentage=0,
-            whitelist={"user-vip"},
-        )
-    )
-    resolver = RenderEngineResolver(default_engine="legacy", store=store)
-
-    assert resolver.get_engine(user_id="user-vip") == "unified"
-    assert resolver.get_engine(user_id="user-other") == "legacy"
-
-
-def test_resolver_percentage_0_all_legacy():
-    """percentage=0 且无白名单时，全部走 legacy。"""
-    from video_processing.render_engine_resolver import RenderEngineResolver
-
-    from packages.adapters.redis.feature_flag_store import (
-        FeatureFlagConfig,
-        InMemoryFeatureFlagStore,
-    )
-
-    store = InMemoryFeatureFlagStore()
-    store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=0))
-    resolver = RenderEngineResolver(default_engine="legacy", store=store)
-
-    for i in range(50):
-        assert resolver.get_engine(user_id=f"user-{i}") == "legacy"
-
-
-def test_resolver_default_unified_when_flag_off():
-    """默认引擎设为 unified 且 flag 关闭时，返回 unified。"""
-    from video_processing.render_engine_resolver import RenderEngineResolver
-
-    from packages.adapters.redis.feature_flag_store import (
-        FeatureFlagConfig,
-        InMemoryFeatureFlagStore,
-    )
-
-    store = InMemoryFeatureFlagStore()
-    store.set(FeatureFlagConfig(name="render_engine", enabled=False, percentage=0))
-    resolver = RenderEngineResolver(default_engine="unified", store=store)
-
-    assert resolver.get_engine(user_id="user-123") == "unified"
-
-
-# ── _render_with_legacy_engine 集成测试 ──────────────────────────────────────
-
-
-def test_legacy_engine_single_clip_keeps_original_fps():
-    """单 clip 场景：输出保持原帧率（不做 fps 归一化），分辨率缩放正确。"""
-    import subprocess
-    import tempfile
-
-    from video_processing.ffmpeg_utils import probe_video_info
-
-    from apps.worker.worker_app.tasks.generation import _render_with_legacy_engine
-
-    with tempfile.TemporaryDirectory() as tmpdir:
-        tmp_path = Path(tmpdir)
-        input_path = tmp_path / "input.mp4"
-        output_path = tmp_path / "output.mp4"
-
-        # 生成 1 秒 30fps 测试视频（带音频）
-        subprocess.run(
-            [
-                "ffmpeg",
-                "-y",
-                "-f",
-                "lavfi",
-                "-i",
-                "color=c=red:s=640x360:d=1:r=30",
-                "-f",
-                "lavfi",
-                "-i",
-                "anullsrc=r=44100:cl=stereo:d=1",
-                "-c:v",
-                "libx264",
-                "-pix_fmt",
-                "yuv420p",
-                "-c:a",
-                "aac",
-                "-shortest",
-                str(input_path),
-            ],
-            check=True,
-            capture_output=True,
-        )
-
-        clip = _TestClip(asset_id="asset-1", duration=1.0)
-        asset_path_map = {"asset-1": input_path}
-
-        duration, file_size = _render_with_legacy_engine(
-            task_id="test-task",
-            virtual_clips=[clip],
-            asset_path_map=asset_path_map,
-            work_dir=tmp_path,
-            output_path=output_path,
-        )
-
-        assert output_path.exists()
-        assert file_size > 0
-        assert duration > 0
-
-        # 旧引擎保持原帧率（30fps），不做 fps 归一化
-        info = probe_video_info(str(output_path))
-        assert abs(info.get("fps", 0) - 30.0) < 0.5
-        assert info.get("width") == 1280
-        assert info.get("height") == 720
-
-
-def test_legacy_engine_two_clips_concat_duration():
-    """多 clip 场景：concat 后时长为两片段之和。"""
-    import subprocess
-    import tempfile
-
-    from video_processing.ffmpeg_utils import probe_duration
-
-    from apps.worker.worker_app.tasks.generation import _render_with_legacy_engine
-
-    with tempfile.TemporaryDirectory() as tmpdir:
-        tmp_path = Path(tmpdir)
-        input1 = tmp_path / "input1.mp4"
-        input2 = tmp_path / "input2.mp4"
-        output_path = tmp_path / "output.mp4"
-
-        for idx, inp in enumerate([input1, input2]):
-            color = "red" if idx == 0 else "blue"
-            subprocess.run(
-                [
-                    "ffmpeg",
-                    "-y",
-                    "-f",
-                    "lavfi",
-                    "-i",
-                    f"color=c={color}:s=640x360:d=1:r=30",
-                    "-f",
-                    "lavfi",
-                    "-i",
-                    "anullsrc=r=44100:cl=stereo:d=1",
-                    "-c:v",
-                    "libx264",
-                    "-pix_fmt",
-                    "yuv420p",
-                    "-c:a",
-                    "aac",
-                    "-shortest",
-                    str(inp),
-                ],
-                check=True,
-                capture_output=True,
-            )
-
-        clip1 = _TestClip(asset_id="asset-1", duration=1.0, clip_type="main", order=0)
-        clip2 = _TestClip(asset_id="asset-2", duration=1.0, clip_type="main", order=1)
-        asset_path_map = {"asset-1": input1, "asset-2": input2}
-
-        duration, file_size = _render_with_legacy_engine(
-            task_id="test-task",
-            virtual_clips=[clip1, clip2],
-            asset_path_map=asset_path_map,
-            work_dir=tmp_path,
-            output_path=output_path,
-        )
-
-        assert output_path.exists()
-        assert file_size > 0
-        assert abs(duration - 2.0) < 0.2
-
-
-def test_legacy_engine_broll_mode_supported():
-    """b_roll 类型的 clip 也被正确识别为主图层并渲染。"""
-    import subprocess
-    import tempfile
-
-    from apps.worker.worker_app.tasks.generation import _render_with_legacy_engine
-
-    with tempfile.TemporaryDirectory() as tmpdir:
-        tmp_path = Path(tmpdir)
-        input_path = tmp_path / "input.mp4"
-        output_path = tmp_path / "output.mp4"
-
-        subprocess.run(
-            [
-                "ffmpeg",
-                "-y",
-                "-f",
-                "lavfi",
-                "-i",
-                "color=c=green:s=640x360:d=1:r=30",
-                "-f",
-                "lavfi",
-                "-i",
-                "anullsrc=r=44100:cl=stereo:d=1",
-                "-c:v",
-                "libx264",
-                "-pix_fmt",
-                "yuv420p",
-                "-c:a",
-                "aac",
-                "-shortest",
-                str(input_path),
-            ],
-            check=True,
-            capture_output=True,
-        )
-
-        clip = _TestClip(
-            asset_id="asset-1",
-            duration=1.0,
-            clip_type="main",
-            config={"role": "b_roll"},
-        )
-        asset_path_map = {"asset-1": input_path}
-
-        duration, file_size = _render_with_legacy_engine(
-            task_id="test-task",
-            virtual_clips=[clip],
-            asset_path_map=asset_path_map,
-            work_dir=tmp_path,
-            output_path=output_path,
-        )
-
-        assert output_path.exists()
-        assert file_size > 0
-        assert duration > 0