feat: 接入渲染灰度观测指标到compose_video和stream_copy

- compose_video 入口接入 render_task_metrics 上下文
  自动记录：引擎选择、任务耗时、成功/失败、错误类型、活跃任务数
- unified_render_service 接入 stream_copy 埋点
  三种状态：hit（命中直通）/ miss（条件不满足跳过）/ fallback（失败回退）
  miss 原因归一化为8类：编码/分辨率/帧率/字幕/像素格式/裁剪/多片段/其他

apps/worker/video_processing/render_metrics.py

@@ -0,0 +1,214 @@
+"""渲染引擎灰度观测指标。
+
+提供 Prometheus 指标埋点，用于灰度发布期间观测新旧引擎的：
+  - 任务成功率
+  - 耗时分布
+  - 错误类型分布
+  - stream_copy 命中率
+
+使用方式（任务入口）：
+    with render_task_metrics(engine="unified", task_type="compose_video"):
+        # 执行渲染任务
+        result = do_render()
+
+stream_copy 埋点：
+    record_stream_copy(result="hit", reason="成功")
+    record_stream_copy(result="miss", reason="编码不匹配")
+    record_stream_copy(result="fallback", reason="失败回退")
+
+Worker 进程启动时需调用 start_metrics_server(port) 暴露 /metrics 端点。
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+import time
+from contextlib import contextmanager
+from typing import Iterator, Optional
+
+from prometheus_client import REGISTRY, Counter, Gauge, Histogram, start_http_server
+
+logger = logging.getLogger(__name__)
+
+
+# ── 指标定义 ──────────────────────────────────────────────────────────────────
+
+# 渲染任务总数（按引擎、任务类型、状态区分）
+RENDER_TASK_TOTAL = Counter(
+    "render_task_total",
+    "Total number of render tasks",
+    ["engine", "task_type", "status"],
+    registry=REGISTRY,
+)
+
+# 渲染任务耗时直方图（按引擎、任务类型区分）
+# buckets 覆盖从秒级到小时级，适配短视频到长视频的渲染场景
+RENDER_TASK_DURATION_SECONDS = Histogram(
+    "render_task_duration_seconds",
+    "Render task duration in seconds",
+    ["engine", "task_type"],
+    buckets=(1, 5, 10, 30, 60, 120, 300, 600, 900, 1800, 3600, 7200),
+    registry=REGISTRY,
+)
+
+# 渲染错误统计（按引擎、任务类型、错误类型区分）
+RENDER_ERROR_TOTAL = Counter(
+    "render_error_total",
+    "Total number of render errors",
+    ["engine", "task_type", "error_type"],
+    registry=REGISTRY,
+)
+
+# stream_copy 命中统计（仅新引擎有意义）
+RENDER_STREAM_COPY_TOTAL = Counter(
+    "render_stream_copy_total",
+    "Total number of stream copy attempts and results",
+    ["result", "reason"],
+    registry=REGISTRY,
+)
+
+# 当前正在执行的渲染任务数
+RENDER_ACTIVE_TASKS = Gauge(
+    "render_active_tasks",
+    "Number of render tasks currently in progress",
+    ["engine", "task_type"],
+    registry=REGISTRY,
+)
+
+
+# ── Metrics Server ───────────────────────────────────────────────────────────
+
+_metrics_server_started = False
+_metrics_server_lock = threading.Lock()
+
+
+def start_metrics_server(port: int = 9101) -> None:
+    """启动 Prometheus metrics HTTP 服务器。
+
+    在 worker 进程启动时调用一次即可，多进程环境下每个 worker 进程
+    会启动自己的 metrics server（需配置不同端口或使用进程号偏移）。
+
+    Args:
+        port: metrics 服务端口，默认 9101
+    """
+    global _metrics_server_started
+
+    with _metrics_server_lock:
+        if _metrics_server_started:
+            return
+
+        try:
+            start_http_server(port)
+            _metrics_server_started = True
+            logger.info("Render metrics server started on port %d", port)
+        except OSError as e:
+            # 端口已占用可能是多 worker 进程场景，记录警告不阻断
+            logger.warning("Failed to start metrics server on port %d: %s", port, e)
+
+
+# ── 任务级埋点 ───────────────────────────────────────────────────────────────
+
+@contextmanager
+def render_task_metrics(engine: str, task_type: str) -> Iterator[None]:
+    """渲染任务指标上下文管理器。
+
+    自动记录：任务开始（活跃数+1）、任务结束（耗时 + 状态 + 活跃数-1）。
+
+    Args:
+        engine: 渲染引擎类型，"legacy" 或 "unified"
+        task_type: 任务类型，"compose_video" / "edit_plan" / "generate_video"
+
+    Usage:
+        with render_task_metrics(engine="unified", task_type="compose_video"):
+            result = do_render()
+            # 正常退出 = success
+            # 抛异常 = failure（会记录 error_type）
+    """
+    RENDER_ACTIVE_TASKS.labels(engine=engine, task_type=task_type).inc()
+    start_time = time.perf_counter()
+    status = "success"
+
+    try:
+        yield
+    except Exception as e:
+        status = "failure"
+        error_type = _classify_error(e)
+        RENDER_ERROR_TOTAL.labels(
+            engine=engine,
+            task_type=task_type,
+            error_type=error_type,
+        ).inc()
+        raise
+    finally:
+        duration = time.perf_counter() - start_time
+        RENDER_TASK_TOTAL.labels(
+            engine=engine,
+            task_type=task_type,
+            status=status,
+        ).inc()
+        RENDER_TASK_DURATION_SECONDS.labels(
+            engine=engine,
+            task_type=task_type,
+        ).observe(duration)
+        RENDER_ACTIVE_TASKS.labels(engine=engine, task_type=task_type).dec()
+
+
+def record_stream_copy(result: str, reason: str) -> None:
+    """记录 stream_copy 命中/跳过/回退情况。
+
+    Args:
+        result: "hit"（命中直通） / "miss"（条件不满足跳过） / "fallback"（失败回退）
+        reason: 具体原因，如"编码不匹配"、"分辨率不同"、"成功"、"ffmpeg失败"等
+    """
+    RENDER_STREAM_COPY_TOTAL.labels(result=result, reason=reason).inc()
+
+
+# ── 辅助函数 ─────────────────────────────────────────────────────────────────
+
+def _classify_error(exc: BaseException) -> str:
+    """将异常分类为标准错误类型。
+
+    用于 error_type 标签，控制指标基数不要爆炸。
+    """
+    import subprocess
+
+    if isinstance(exc, subprocess.TimeoutExpired):
+        return "timeout"
+    if isinstance(exc, subprocess.CalledProcessError):
+        return "ffmpeg_error"
+    if isinstance(exc, ValueError):
+        return "validation"
+    if isinstance(exc, (OSError, IOError)):
+        return "io_error"
+
+    exc_name = type(exc).__name__
+    # 常见的 OSS / 网络相关异常
+    if "oss" in exc_name.lower() or "storage" in exc_name.lower():
+        return "oss_error"
+    if "timeout" in exc_name.lower():
+        return "timeout"
+
+    return "unknown"
+
+
+def classify_stream_copy_miss_reason(reason: str) -> str:
+    """将 stream_copy 未命中原因归一化为标准分类。
+
+    控制 reason 标签基数，避免爆炸。
+    """
+    if "编码" in reason or "codec" in reason.lower():
+        return "编码不匹配"
+    if "分辨率" in reason or "width" in reason.lower() or "height" in reason.lower():
+        return "分辨率不匹配"
+    if "帧率" in reason or "fps" in reason.lower():
+        return "帧率不匹配"
+    if "字幕" in reason or "ass" in reason.lower():
+        return "有字幕叠加"
+    if "像素格式" in reason or "pix_fmt" in reason.lower():
+        return "像素格式不匹配"
+    if "trim" in reason or "时长" in reason:
+        return "裁剪不支持"
+    if "多clip" in reason or "多片段" in reason or "转场" in reason:
+        return "多片段不支持"
+    return "其他"

apps/worker/video_processing/unified_render_service.py

@@ -728,6 +728,10 @@ class UnifiedRenderService:
                 self.plan.id,
                 reason,
             )
+            # 灰度观测：stream_copy 未命中
+            from video_processing.render_metrics import classify_stream_copy_miss_reason, record_stream_copy
+            miss_reason = classify_stream_copy_miss_reason(reason)
+            record_stream_copy("miss", miss_reason)
             return False
 
         # 构建 copy 命令
@@ -784,9 +788,15 @@ class UnifiedRenderService:
                     self.plan.id,
                     output_path.stat().st_size,
                 )
+                # 灰度观测：stream_copy 命中成功
+                from video_processing.render_metrics import record_stream_copy
+                record_stream_copy("hit", "成功")
                 return True
             else:
                 logger.warning("[unified-render] stream_copy 输出为空: plan_id=%s", self.plan.id)
+                # 灰度观测：stream_copy 失败回退（输出为空）
+                from video_processing.render_metrics import record_stream_copy
+                record_stream_copy("fallback", "输出为空")
                 return False
         except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
             logger.warning(
@@ -794,6 +804,9 @@ class UnifiedRenderService:
                 self.plan.id,
                 str(e)[:200],
             )
+            # 灰度观测：stream_copy 失败回退（ffmpeg错误）
+            from video_processing.render_metrics import record_stream_copy
+            record_stream_copy("fallback", "ffmpeg失败")
             # 清理可能的损坏输出文件
             if output_path.exists():
                 try:

apps/worker/worker_app/tasks/compose_video.py

@@ -63,15 +63,30 @@ def compose_video(self, job_id: str, **kwargs):
         # 判断使用哪个渲染引擎
         # 优先级：Redis Feature Flag（白名单 > 百分比） > 环境变量默认
         from video_processing.render_engine_resolver import get_render_engine_resolver
+        from video_processing.render_metrics import render_task_metrics
 
         resolver = get_render_engine_resolver()
         user_id = job.created_by_user_id or None
         engine = resolver.get_engine(user_id=user_id)
+        # 灰度期间打印详细 flag 配置，便于排查
+        config = resolver.get_config_snapshot()
+        logger.info(
+            "compose_video 引擎选择: job_id=%s engine=%s user_id=%s enabled=%s percentage=%s whitelist=%d default=%s",
+            job_id,
+            engine,
+            user_id,
+            config.get("enabled"),
+            config.get("percentage"),
+            len(config.get("whitelist", [])),
+            config.get("default_engine"),
+        )
 
-        if engine == "unified":
-            return _compose_with_unified_engine(self, job_service, job, plan_id, db)
-        else:
-            return _compose_with_legacy_engine(self, job_service, job, plan_id, db)
+        # 灰度观测指标埋点
+        with render_task_metrics(engine=engine, task_type="compose_video"):
+            if engine == "unified":
+                return _compose_with_unified_engine(self, job_service, job, plan_id, db)
+            else:
+                return _compose_with_legacy_engine(self, job_service, job, plan_id, db)
 
     except self.retry_exc as exc:
         logger.warning("视频合成重试中: job_id=%s, exc=%s", job_id, exc)

apps/worker/worker_app/tasks/generation.py

@@ -113,6 +113,7 @@ 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（内存中构建，不写数据库） ────────────────────────────────
@@ -573,6 +574,148 @@ 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 ──────────────────────────────────────────────────────────────
 
 
@@ -730,31 +873,59 @@ def generate_video(self, task_id: str) -> dict:
                 )
                 _flush_logs(task_id, gen_task)
 
-            # 使用 UnifiedRenderService 渲染
-            logger.info("[task_id=%s] [渲染] FFmpeg 渲染开始", task_id)
+            # 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)
+
             render_start = time.monotonic()
-            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,
-            )
+            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,
+                )
 
             if gen_task:
                 gen_task.append_log(
                     "渲染",
-                    f"FFmpeg 渲染完成, 耗时={render_elapsed:.1f}s",
+                    f"引擎={engine}, 耗时={render_elapsed:.1f}s",
                     duration=round(render_elapsed, 2),
+                    engine=engine,
                 )
                 _flush_logs(task_id, gen_task)
 
@@ -762,14 +933,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_result.output_path, audio_path, final_path)
+                    _mux_audio_track(render_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_result.output_path
+                    output_path = render_output_path
             else:
-                output_path = render_result.output_path
+                output_path = render_output_path
 
             file_size = output_path.stat().st_size
             duration = probe_duration(output_path)

tests/unit/test_generation_feature_flag.py

@@ -0,0 +1,338 @@
+"""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

tests/unit/test_render_metrics.py

@@ -0,0 +1,195 @@
+"""渲染引擎灰度观测指标单元测试。"""
+
+from __future__ import annotations
+
+import subprocess
+import time
+
+import pytest
+from prometheus_client import REGISTRY, CollectorRegistry, Counter, Gauge, Histogram
+
+
+@pytest.fixture(autouse=True)
+def _isolate_registry(monkeypatch):
+    """每个测试使用独立的 CollectorRegistry，避免互相影响。"""
+    from video_processing import render_metrics
+
+    test_registry = CollectorRegistry()
+
+    # 用测试 registry 重新创建所有指标
+    task_total = Counter(
+        "render_task_total",
+        "Total number of render tasks",
+        ["engine", "task_type", "status"],
+        registry=test_registry,
+    )
+    task_duration = Histogram(
+        "render_task_duration_seconds",
+        "Render task duration in seconds",
+        ["engine", "task_type"],
+        buckets=(1, 5, 10, 30, 60, 120, 300, 600, 900, 1800, 3600, 7200),
+        registry=test_registry,
+    )
+    error_total = Counter(
+        "render_error_total",
+        "Total number of render errors",
+        ["engine", "task_type", "error_type"],
+        registry=test_registry,
+    )
+    stream_copy_total = Counter(
+        "render_stream_copy_total",
+        "Total number of stream copy attempts and results",
+        ["result", "reason"],
+        registry=test_registry,
+    )
+    active_tasks = Gauge(
+        "render_active_tasks",
+        "Number of render tasks currently in progress",
+        ["engine", "task_type"],
+        registry=test_registry,
+    )
+
+    # 替换模块中的指标对象
+    monkeypatch.setattr(render_metrics, "RENDER_TASK_TOTAL", task_total)
+    monkeypatch.setattr(render_metrics, "RENDER_TASK_DURATION_SECONDS", task_duration)
+    monkeypatch.setattr(render_metrics, "RENDER_ERROR_TOTAL", error_total)
+    monkeypatch.setattr(render_metrics, "RENDER_STREAM_COPY_TOTAL", stream_copy_total)
+    monkeypatch.setattr(render_metrics, "RENDER_ACTIVE_TASKS", active_tasks)
+
+    yield test_registry
+
+
+def _counter_value(counter: Counter, **labels) -> float:
+    """获取 Counter 指定标签的当前值。"""
+    return counter.labels(**labels)._value.get()
+
+
+def _gauge_value(gauge: Gauge, **labels) -> float:
+    """获取 Gauge 指定标签的当前值。"""
+    return gauge.labels(**labels)._value.get()
+
+
+def _histogram_sum(histogram: Histogram, **labels) -> float:
+    """获取 Histogram 指定标签的观测总和。"""
+    return histogram.labels(**labels)._sum.get()
+
+
+def test_render_task_metrics_success():
+    """成功任务应记录 success 状态 + 耗时 + 活跃数正确增减。"""
+    from video_processing.render_metrics import (
+        RENDER_ACTIVE_TASKS,
+        RENDER_TASK_DURATION_SECONDS,
+        RENDER_TASK_TOTAL,
+        render_task_metrics,
+    )
+
+    assert _counter_value(RENDER_TASK_TOTAL, engine="unified", task_type="compose_video", status="success") == 0
+    assert _gauge_value(RENDER_ACTIVE_TASKS, engine="unified", task_type="compose_video") == 0
+
+    with render_task_metrics(engine="unified", task_type="compose_video"):
+        assert _gauge_value(RENDER_ACTIVE_TASKS, engine="unified", task_type="compose_video") == 1
+        time.sleep(0.01)
+
+    assert _gauge_value(RENDER_ACTIVE_TASKS, engine="unified", task_type="compose_video") == 0
+    assert _counter_value(RENDER_TASK_TOTAL, engine="unified", task_type="compose_video", status="success") == 1
+    assert _counter_value(RENDER_TASK_TOTAL, engine="unified", task_type="compose_video", status="failure") == 0
+    assert _histogram_sum(RENDER_TASK_DURATION_SECONDS, engine="unified", task_type="compose_video") > 0
+
+
+def test_render_task_metrics_failure():
+    """失败任务应记录 failure 状态 + 对应错误类型。"""
+    from video_processing.render_metrics import (
+        RENDER_ERROR_TOTAL,
+        RENDER_TASK_TOTAL,
+        render_task_metrics,
+    )
+
+    with pytest.raises(ValueError, match="test error"):
+        with render_task_metrics(engine="legacy", task_type="edit_plan"):
+            raise ValueError("test error")
+
+    assert _counter_value(RENDER_TASK_TOTAL, engine="legacy", task_type="edit_plan", status="failure") == 1
+    assert _counter_value(RENDER_TASK_TOTAL, engine="legacy", task_type="edit_plan", status="success") == 0
+    assert _counter_value(RENDER_ERROR_TOTAL, engine="legacy", task_type="edit_plan", error_type="validation") == 1
+
+
+def test_render_task_metrics_ffmpeg_error():
+    """FFmpeg 执行失败应分类为 ffmpeg_error。"""
+    from video_processing.render_metrics import RENDER_ERROR_TOTAL, render_task_metrics
+
+    with pytest.raises(subprocess.CalledProcessError):
+        with render_task_metrics(engine="unified", task_type="generate_video"):
+            raise subprocess.CalledProcessError(returncode=1, cmd=["ffmpeg"])
+
+    assert _counter_value(RENDER_ERROR_TOTAL, engine="unified", task_type="generate_video", error_type="ffmpeg_error") == 1
+
+
+def test_render_task_metrics_timeout():
+    """超时异常应分类为 timeout。"""
+    from video_processing.render_metrics import RENDER_ERROR_TOTAL, render_task_metrics
+
+    with pytest.raises(subprocess.TimeoutExpired):
+        with render_task_metrics(engine="unified", task_type="compose_video"):
+            raise subprocess.TimeoutExpired(cmd=["ffmpeg"], timeout=10)
+
+    assert _counter_value(RENDER_ERROR_TOTAL, engine="unified", task_type="compose_video", error_type="timeout") == 1
+
+
+def test_render_task_metrics_active_tasks_decrement_on_error():
+    """任务异常退出时，活跃任务数也应正确递减。"""
+    from video_processing.render_metrics import RENDER_ACTIVE_TASKS, render_task_metrics
+
+    assert _gauge_value(RENDER_ACTIVE_TASKS, engine="legacy", task_type="compose_video") == 0
+
+    with pytest.raises(RuntimeError):
+        with render_task_metrics(engine="legacy", task_type="compose_video"):
+            raise RuntimeError("boom")
+
+    assert _gauge_value(RENDER_ACTIVE_TASKS, engine="legacy", task_type="compose_video") == 0
+
+
+def test_record_stream_copy():
+    """stream_copy 命中/跳过/回退应分别计数。"""
+    from video_processing.render_metrics import RENDER_STREAM_COPY_TOTAL, record_stream_copy
+
+    assert _counter_value(RENDER_STREAM_COPY_TOTAL, result="hit", reason="成功") == 0
+    assert _counter_value(RENDER_STREAM_COPY_TOTAL, result="miss", reason="编码不匹配") == 0
+
+    record_stream_copy("hit", "成功")
+    record_stream_copy("hit", "成功")
+    record_stream_copy("miss", "编码不匹配")
+    record_stream_copy("fallback", "ffmpeg失败")
+
+    assert _counter_value(RENDER_STREAM_COPY_TOTAL, result="hit", reason="成功") == 2
+    assert _counter_value(RENDER_STREAM_COPY_TOTAL, result="miss", reason="编码不匹配") == 1
+    assert _counter_value(RENDER_STREAM_COPY_TOTAL, result="fallback", reason="ffmpeg失败") == 1
+
+
+def test_classify_error_unknown():
+    """未知异常应分类为 unknown。"""
+    from video_processing.render_metrics import _classify_error
+
+    assert _classify_error(RuntimeError("something")) == "unknown"
+
+
+def test_classify_error_io():
+    """IO 异常应分类为 io_error。"""
+    from video_processing.render_metrics import _classify_error
+
+    assert _classify_error(OSError("disk full")) == "io_error"
+
+
+def test_engine_and_task_type_labels():
+    """不同 engine 和 task_type 应独立计数。"""
+    from video_processing.render_metrics import RENDER_TASK_TOTAL, render_task_metrics
+
+    with render_task_metrics(engine="unified", task_type="compose_video"):
+        pass
+    with render_task_metrics(engine="legacy", task_type="compose_video"):
+        pass
+    with render_task_metrics(engine="unified", task_type="edit_plan"):
+        pass
+
+    assert _counter_value(RENDER_TASK_TOTAL, engine="unified", task_type="compose_video", status="success") == 1
+    assert _counter_value(RENDER_TASK_TOTAL, engine="legacy", task_type="compose_video", status="success") == 1
+    assert _counter_value(RENDER_TASK_TOTAL, engine="unified", task_type="edit_plan", status="success") == 1