apps/api/app/api/router.py

@@ -8,6 +8,7 @@ from app.api.routes.dashboard import router as dashboard_router
 from app.api.routes.duplication import router as duplication_router
 from app.api.routes.edit_plans import router as edit_plans_router
 from app.api.routes.edit_templates import router as edit_templates_router
+from app.api.routes.feature_flags import router as feature_flags_router
 from app.api.routes.generated_videos import router as generated_videos_router
 from app.api.routes.generation_tasks import router as generation_tasks_router
 from app.api.routes.health import router as health_check_router
@@ -151,3 +152,7 @@ api_router.include_router(
     prefix="/tts",
     tags=["TTS"],
 )
+api_router.include_router(
+    feature_flags_router,
+    tags=["Internal"],
+)

apps/api/app/api/routes/feature_flags.py

@@ -0,0 +1,195 @@
+"""Feature Flag 内部管理接口。
+
+通过内部 API Key 鉴权，支持查看和修改 Feature Flag 配置。
+主要用于灰度发布期间的动态开关控制。
+
+API:
+  GET  /api/v1/internal/feature-flags          - 列出所有 flag
+  GET  /api/v1/internal/feature-flags/{name}   - 查看单个 flag
+  PUT  /api/v1/internal/feature-flags/{name}   - 设置 flag 配置
+  DELETE /api/v1/internal/feature-flags/{name} - 删除 flag
+
+鉴权：X-API-Key header，走内部 API Key 验证
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from app.api.routes.auth import _verify_internal_api_key
+from app.config import settings
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from pydantic import BaseModel, Field
+
+from packages.adapters.redis.feature_flag_store import (
+    FEATURE_FLAG_REDIS_PREFIX,
+    FeatureFlagConfig,
+    RedisFeatureFlagStore,
+)
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/internal/feature-flags", tags=["Internal"])
+
+# 允许管理的 flag 白名单（防止误操作其他系统 flag）
+ALLOWED_FLAGS = {
+    "render_engine",
+}
+
+
+def _get_feature_flag_store() -> RedisFeatureFlagStore:
+    """获取 Feature Flag 存储实例。"""
+    return RedisFeatureFlagStore(redis_url=settings.REDIS_URL)
+
+
+class FeatureFlagUpdateRequest(BaseModel):
+    """Feature Flag 更新请求体。"""
+
+    enabled: bool = Field(..., description="是否启用")
+    percentage: int = Field(0, ge=0, le=100, description="灰度百分比 (0-100)")
+    whitelist: list[str] = Field(default_factory=list, description="白名单列表（如 user_id）")
+
+
+class FeatureFlagResponse(BaseModel):
+    """Feature Flag 响应。"""
+
+    name: str
+    enabled: bool
+    percentage: int
+    whitelist: list[str]
+
+    @classmethod
+    def from_config(cls, config: FeatureFlagConfig) -> "FeatureFlagResponse":
+        return cls(
+            name=config.name,
+            enabled=config.enabled,
+            percentage=config.percentage,
+            whitelist=sorted(config.whitelist),
+        )
+
+
+class FeatureFlagCheckResponse(BaseModel):
+    """Flag 激活检查响应。"""
+
+    name: str
+    active: bool
+    identifier: Optional[str] = None
+
+
+def _validate_flag_name(name: str) -> None:
+    """校验 flag 名称是否在允许列表中。"""
+    if name not in ALLOWED_FLAGS:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Unsupported flag: {name}. Allowed: {sorted(ALLOWED_FLAGS)}",
+        )
+
+
+@router.get("", response_model=list[FeatureFlagResponse])
+async def list_feature_flags(
+    _: bool = Depends(_verify_internal_api_key),
+    store: RedisFeatureFlagStore = Depends(_get_feature_flag_store),
+):
+    """列出所有 Feature Flag。"""
+    try:
+        flags = store.list_all()
+        # 同时返回预定义的 flag（即使未设置也显示默认值）
+        result = []
+        for name in sorted(ALLOWED_FLAGS):
+            config = flags.get(name) or FeatureFlagConfig(name=name, enabled=False)
+            result.append(FeatureFlagResponse.from_config(config))
+        # 加上已存在但不在白名单中的 flag（只读展示）
+        for name, config in flags.items():
+            if name not in ALLOWED_FLAGS:
+                result.append(FeatureFlagResponse.from_config(config))
+        return sorted(result, key=lambda x: x.name)
+    except Exception as exc:
+        logger.error("Failed to list feature flags: %s", exc)
+        raise HTTPException(status_code=500, detail=f"Failed to list flags: {exc}")
+
+
+@router.get("/{name}", response_model=FeatureFlagResponse)
+async def get_feature_flag(
+    name: str,
+    _: bool = Depends(_verify_internal_api_key),
+    store: RedisFeatureFlagStore = Depends(_get_feature_flag_store),
+):
+    """获取单个 Feature Flag 配置。"""
+    try:
+        config = store.get(name)
+        return FeatureFlagResponse.from_config(config)
+    except Exception as exc:
+        logger.error("Failed to get feature flag %s: %s", name, exc)
+        raise HTTPException(status_code=500, detail=f"Failed to get flag: {exc}")
+
+
+@router.get("/{name}/check", response_model=FeatureFlagCheckResponse)
+async def check_feature_flag(
+    name: str,
+    identifier: Optional[str] = Query(None, description="标识符，如 user_id"),
+    _: bool = Depends(_verify_internal_api_key),
+    store: RedisFeatureFlagStore = Depends(_get_feature_flag_store),
+):
+    """检查某个标识符是否命中 Feature Flag。"""
+    try:
+        active = store.is_active(name, identifier=identifier)
+        return FeatureFlagCheckResponse(name=name, active=active, identifier=identifier)
+    except Exception as exc:
+        logger.error("Failed to check feature flag %s: %s", name, exc)
+        raise HTTPException(status_code=500, detail=f"Failed to check flag: {exc}")
+
+
+@router.put("/{name}", response_model=FeatureFlagResponse)
+async def update_feature_flag(
+    name: str,
+    request: FeatureFlagUpdateRequest,
+    _: bool = Depends(_verify_internal_api_key),
+    store: RedisFeatureFlagStore = Depends(_get_feature_flag_store),
+):
+    """更新 Feature Flag 配置。
+
+    只允许修改 ALLOWED_FLAGS 列表中的 flag。
+    """
+    _validate_flag_name(name)
+
+    try:
+        config = FeatureFlagConfig(
+            name=name,
+            enabled=request.enabled,
+            percentage=request.percentage,
+            whitelist=set(request.whitelist),
+        )
+        store.set(config)
+        logger.info(
+            "Feature flag updated: name=%s enabled=%s percentage=%d whitelist=%d",
+            name,
+            config.enabled,
+            config.percentage,
+            len(config.whitelist),
+        )
+        return FeatureFlagResponse.from_config(config)
+    except Exception as exc:
+        logger.error("Failed to update feature flag %s: %s", name, exc)
+        raise HTTPException(status_code=500, detail=f"Failed to update flag: {exc}")
+
+
+@router.delete("/{name}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_feature_flag(
+    name: str,
+    _: bool = Depends(_verify_internal_api_key),
+    store: RedisFeatureFlagStore = Depends(_get_feature_flag_store),
+):
+    """删除 Feature Flag。
+
+    只允许删除 ALLOWED_FLAGS 列表中的 flag。
+    """
+    _validate_flag_name(name)
+
+    try:
+        deleted = store.delete(name)
+        logger.info("Feature flag deleted: name=%s deleted=%s", name, deleted)
+        return None
+    except Exception as exc:
+        logger.error("Failed to delete feature flag %s: %s", name, exc)
+        raise HTTPException(status_code=500, detail=f"Failed to delete flag: {exc}")

apps/worker/video_processing/__init__.py

@@ -6,6 +6,7 @@
 from . import dedup_helpers, ffmpeg_utils, oss_helpers
 from .processor import VideoProcessor, VideoResult
 from .render_adapter import RenderAdapter, RenderAdapterResult
+from .render_engine_resolver import RenderEngineResolver, get_render_engine_resolver
 from .unified_render_service import RenderResult, UnifiedRenderService
 
 __all__ = [
@@ -18,4 +19,6 @@ __all__ = [
     "RenderResult",
     "RenderAdapter",
     "RenderAdapterResult",
+    "RenderEngineResolver",
+    "get_render_engine_resolver",
 ]

apps/worker/video_processing/render_engine_resolver.py

@@ -0,0 +1,204 @@
+"""渲染引擎 Feature Flag 解析器。
+
+封装渲染引擎选择逻辑，支持：
+- 环境变量作为默认值（RENDER_ENGINE=legacy/unified）
+- Redis Feature Flag 运行时覆盖（白名单 + 百分比 + 全局开关）
+- 定时刷新，支持热更新不重启 worker
+
+使用方式：
+    resolver = RenderEngineResolver(redis_url="redis://...", default_engine="legacy")
+    engine = resolver.get_engine(user_id="user123")
+    # engine: "legacy" 或 "unified"
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from typing import Optional
+
+from packages.adapters.redis.feature_flag_store import (
+    FeatureFlagConfig,
+    FeatureFlagStore,
+    InMemoryFeatureFlagStore,
+    RedisFeatureFlagStore,
+)
+
+logger = logging.getLogger(__name__)
+
+# Feature Flag 名称常量
+FLAG_RENDER_ENGINE = "render_engine"
+
+# 引擎常量
+ENGINE_LEGACY = "legacy"
+ENGINE_UNIFIED = "unified"
+VALID_ENGINES = {ENGINE_LEGACY, ENGINE_UNIFIED}
+
+
+class RenderEngineResolver:
+    """渲染引擎选择器。
+
+    判定逻辑（从高到低）：
+    1. Redis flag 白名单匹配 → unified
+    2. Redis flag 百分比命中 → unified
+    3. Redis flag 全局开启（100%）→ unified
+    4. 环境变量默认值 → legacy / unified
+
+    当 Redis 不可用时，自动降级到环境变量默认值，不影响业务。
+    """
+
+    def __init__(
+        self,
+        default_engine: str = ENGINE_LEGACY,
+        redis_url: Optional[str] = None,
+        refresh_interval: float = 30.0,
+        store: Optional[FeatureFlagStore] = None,
+    ) -> None:
+        """
+        Args:
+            default_engine: 环境变量默认的引擎名（legacy / unified）
+            redis_url: Redis 连接 URL，传 None 时使用内存实现（测试用）
+            refresh_interval: Redis flag 配置刷新间隔（秒）
+            store: 直接传入 store 实例（测试用，优先级高于 redis_url）
+        """
+        self._default_engine = default_engine.lower() if default_engine else ENGINE_LEGACY
+        if self._default_engine not in VALID_ENGINES:
+            logger.warning(
+                "Invalid default engine '%s', fallback to '%s'",
+                self._default_engine,
+                ENGINE_LEGACY,
+            )
+            self._default_engine = ENGINE_LEGACY
+
+        if store is not None:
+            self._store = store
+        elif redis_url:
+            self._store = RedisFeatureFlagStore(redis_url=redis_url)
+        else:
+            self._store = InMemoryFeatureFlagStore()
+            logger.info("No Redis configured, using in-memory feature flag store")
+
+        self._refresh_interval = refresh_interval
+        self._lock = threading.Lock()
+        self._cached_config: Optional[FeatureFlagConfig] = None
+        self._last_refresh: float = 0.0
+
+    def _maybe_refresh(self) -> None:
+        """惰性刷新配置，超过刷新间隔时从存储重新读取。"""
+        import time
+
+        now = time.time()
+        if now - self._last_refresh < self._refresh_interval:
+            return
+
+        try:
+            config = self._store.get(FLAG_RENDER_ENGINE)
+            with self._lock:
+                self._cached_config = config
+                self._last_refresh = now
+        except Exception as exc:
+            logger.warning("Failed to refresh render engine flag: %s", exc)
+            # 刷新失败时保留旧缓存，不中断业务
+            if self._cached_config is None:
+                # 首次就读失败，设一个默认值
+                with self._lock:
+                    self._cached_config = FeatureFlagConfig(name=FLAG_RENDER_ENGINE)
+                    self._last_refresh = now
+
+    def _get_config(self) -> FeatureFlagConfig:
+        """获取当前 flag 配置（带缓存）。"""
+        if self._cached_config is None:
+            self._maybe_refresh()
+        else:
+            self._maybe_refresh()
+        return self._cached_config or FeatureFlagConfig(name=FLAG_RENDER_ENGINE)
+
+    def get_engine(self, user_id: Optional[str] = None) -> str:
+        """获取当前应该使用的渲染引擎。
+
+        Args:
+            user_id: 用户ID，用于白名单匹配和百分比哈希。
+                    传 None 时只看全局开关。
+
+        Returns:
+            "legacy" 或 "unified"
+        """
+        config = self._get_config()
+
+        # 全局关闭 → 用默认值
+        if not config.enabled:
+            return self._default_engine
+
+        # 白名单匹配 / 百分比命中 → unified
+        if config.is_active(user_id):
+            return ENGINE_UNIFIED
+
+        # 未命中灰度 → 用默认值
+        return self._default_engine
+
+    def should_use_unified(self, user_id: Optional[str] = None) -> bool:
+        """便捷方法：是否应该使用统一渲染引擎。"""
+        return self.get_engine(user_id) == ENGINE_UNIFIED
+
+    def force_refresh(self) -> None:
+        """强制立即刷新配置（用于管理接口修改后立即生效）。"""
+        self._last_refresh = 0.0
+        if isinstance(self._store, RedisFeatureFlagStore):
+            self._store.invalidate_cache(FLAG_RENDER_ENGINE)
+        self._maybe_refresh()
+
+    def get_config_snapshot(self) -> dict:
+        """获取当前配置快照（用于管理接口展示）。"""
+        config = self._get_config()
+        return {
+            "flag_name": FLAG_RENDER_ENGINE,
+            "default_engine": self._default_engine,
+            "enabled": config.enabled,
+            "percentage": config.percentage,
+            "whitelist": sorted(config.whitelist),
+            "refresh_interval": self._refresh_interval,
+            "last_refresh": self._last_refresh,
+        }
+
+    def set_flag(self, config: FeatureFlagConfig) -> None:
+        """设置 flag 配置（管理接口用）。"""
+        config.name = FLAG_RENDER_ENGINE
+        self._store.set(config)
+        self.force_refresh()
+
+
+# 全局单例
+_resolver: Optional[RenderEngineResolver] = None
+_resolver_lock = threading.Lock()
+
+
+def get_render_engine_resolver() -> RenderEngineResolver:
+    """获取全局单例（基于 worker 配置）。"""
+    global _resolver
+    if _resolver is not None:
+        return _resolver
+
+    with _resolver_lock:
+        if _resolver is not None:
+            return _resolver
+
+        try:
+            from worker_app.core.config import get_settings
+
+            settings = get_settings()
+            redis_url = getattr(settings, "redis_url", None) or getattr(settings, "broker_url", None)
+            default = getattr(settings, "render_engine", ENGINE_LEGACY)
+            _resolver = RenderEngineResolver(
+                default_engine=default,
+                redis_url=redis_url,
+            )
+            logger.info(
+                "RenderEngineResolver initialized: default=%s, redis=%s",
+                default,
+                bool(redis_url),
+            )
+        except Exception as exc:
+            logger.warning("Failed to init RenderEngineResolver from settings: %s", exc)
+            _resolver = RenderEngineResolver(default_engine=ENGINE_LEGACY)
+
+    return _resolver

apps/worker/worker_app/tasks/compose_video.py

@@ -61,10 +61,12 @@ def compose_video(self, job_id: str, **kwargs):
             return {"status": "error", "message": "Missing plan_id"}
 
         # 判断使用哪个渲染引擎
-        from worker_app.core.config import get_settings as get_worker_settings
+        # 优先级：Redis Feature Flag（白名单 > 百分比） > 环境变量默认
+        from video_processing.render_engine_resolver import get_render_engine_resolver
 
-        worker_settings = get_worker_settings()
-        engine = (worker_settings.render_engine or "legacy").lower()
+        resolver = get_render_engine_resolver()
+        user_id = job.created_by_user_id or None
+        engine = resolver.get_engine(user_id=user_id)
 
         if engine == "unified":
             return _compose_with_unified_engine(self, job_service, job, plan_id, db)
@@ -207,7 +209,12 @@ def _compose_with_unified_engine(task, job_service, job, plan_id: str, db) -> di
     }
     job_service.complete_job(job_id, result=result_data)
 
-    logger.info("视频合成完成(unified): job_id=%s plan_id=%s duration=%.2fs", job_id, plan_id, result.duration)
+    logger.info(
+        "视频合成完成(unified): job_id=%s plan_id=%s duration=%.2fs",
+        job_id,
+        plan_id,
+        result.duration,
+    )
     return {"status": "completed", "job_id": job_id, "result": result_data}
 
 

packages/adapters/redis/__init__.py

@@ -1,3 +1,9 @@
+from packages.adapters.redis.feature_flag_store import (
+    FeatureFlagConfig,
+    FeatureFlagStore,
+    InMemoryFeatureFlagStore,
+    RedisFeatureFlagStore,
+)
 from packages.adapters.redis.session_store import (
     NoopSessionStore,
     RedisConfig,
@@ -5,4 +11,13 @@ from packages.adapters.redis.session_store import (
     get_session_store,
 )
 
-__all__ = ["NoopSessionStore", "RedisConfig", "SessionStore", "get_session_store"]
+__all__ = [
+    "FeatureFlagConfig",
+    "FeatureFlagStore",
+    "InMemoryFeatureFlagStore",
+    "NoopSessionStore",
+    "RedisConfig",
+    "RedisFeatureFlagStore",
+    "SessionStore",
+    "get_session_store",
+]

packages/adapters/redis/feature_flag_store.py

@@ -0,0 +1,259 @@
+"""Feature Flag 存储实现。
+
+支持两种后端：
+- RedisFeatureFlagStore：生产环境使用，支持多实例共享、热更新
+- InMemoryFeatureFlagStore：测试/开发环境使用，纯内存
+
+支持的 Flag 类型：
+- 全局开关（enabled: bool）
+- 白名单（whitelist: Set[str]，如 user_id 列表）
+- 百分比切流（percentage: 0-100，基于标识符哈希取模）
+
+判定优先级：白名单 > 百分比 > 全局开关
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import threading
+import time
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Optional, Set
+
+logger = logging.getLogger(__name__)
+
+# Redis key 前缀
+FEATURE_FLAG_REDIS_PREFIX = "feature_flag:"
+
+
+@dataclass
+class FeatureFlagConfig:
+    """单个 Feature Flag 的配置。"""
+
+    name: str
+    enabled: bool = False
+    percentage: int = 0  # 0-100
+    whitelist: Set[str] = field(default_factory=set)
+
+    def to_dict(self) -> dict:
+        return {
+            "name": self.name,
+            "enabled": self.enabled,
+            "percentage": self.percentage,
+            "whitelist": sorted(self.whitelist),
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "FeatureFlagConfig":
+        return cls(
+            name=data["name"],
+            enabled=bool(data.get("enabled", False)),
+            percentage=int(data.get("percentage", 0)),
+            whitelist=set(data.get("whitelist", [])),
+        )
+
+    def is_active(self, identifier: Optional[str] = None) -> bool:
+        """判断当前 flag 是否激活。
+
+        判定优先级：
+        1. 全局关闭 → False
+        2. 白名单匹配 → True
+        3. 百分比命中 → True
+        4. 其他 → False
+
+        Args:
+            identifier: 用于白名单匹配和百分比哈希的标识符（如 user_id）。
+                       传 None 时只看全局开关 + 百分比（百分比用随机值）。
+        """
+        if not self.enabled:
+            return False
+
+        # 白名单：精确匹配
+        if identifier and identifier in self.whitelist:
+            return True
+
+        # 百分比：0 直接 False，100 直接 True
+        if self.percentage <= 0:
+            # 没有白名单且百分比为0 → 未启用
+            return False
+        if self.percentage >= 100:
+            return True
+
+        # 基于 identifier 做哈希取模，确保同一用户始终落在同一侧
+        if identifier:
+            hash_val = int(
+                hashlib.md5(f"{self.name}:{identifier}".encode("utf-8")).hexdigest(), 16  # nosec B324
+            )  # nosec B324 - 用于哈希取模做百分比切流，非安全用途
+            return (hash_val % 100) < self.percentage
+
+        # 无 identifier 且百分比在 0-100 之间 → 按比例随机（不保证一致性）
+        import random
+
+        return random.randint(0, 99) < self.percentage
+
+
+class FeatureFlagStore(ABC):
+    """Feature Flag 存储抽象接口。"""
+
+    @abstractmethod
+    def get(self, name: str) -> FeatureFlagConfig:
+        """获取指定 flag 的配置，不存在则返回默认配置（关闭状态）。"""
+        ...
+
+    @abstractmethod
+    def set(self, config: FeatureFlagConfig) -> None:
+        """设置 flag 配置。"""
+        ...
+
+    @abstractmethod
+    def delete(self, name: str) -> bool:
+        """删除 flag，返回是否成功删除。"""
+        ...
+
+    @abstractmethod
+    def list_all(self) -> dict[str, FeatureFlagConfig]:
+        """列出所有 flag。"""
+        ...
+
+    def is_active(self, name: str, identifier: Optional[str] = None) -> bool:
+        """便捷方法：判断 flag 是否激活。"""
+        return self.get(name).is_active(identifier)
+
+
+class InMemoryFeatureFlagStore(FeatureFlagStore):
+    """内存实现，用于测试和本地开发。"""
+
+    def __init__(self) -> None:
+        self._flags: dict[str, FeatureFlagConfig] = {}
+        self._lock = threading.Lock()
+
+    def get(self, name: str) -> FeatureFlagConfig:
+        with self._lock:
+            return self._flags.get(name, FeatureFlagConfig(name=name, enabled=False))
+
+    def set(self, config: FeatureFlagConfig) -> None:
+        with self._lock:
+            self._flags[config.name] = config
+
+    def delete(self, name: str) -> bool:
+        with self._lock:
+            if name in self._flags:
+                del self._flags[name]
+                return True
+            return False
+
+    def list_all(self) -> dict[str, FeatureFlagConfig]:
+        with self._lock:
+            return dict(self._flags)
+
+
+class RedisFeatureFlagStore(FeatureFlagStore):
+    """Redis 实现，支持多实例共享配置。
+
+    每个 flag 存在一个独立的 Redis hash key 中：
+      Key: feature_flag:{name}
+      Fields: enabled, percentage, whitelist(JSON array)
+    """
+
+    def __init__(self, redis_url: str, key_prefix: str = FEATURE_FLAG_REDIS_PREFIX) -> None:
+        import redis as redis_lib
+
+        self._redis = redis_lib.from_url(redis_url, decode_responses=True)
+        self._key_prefix = key_prefix
+        # 本地缓存 + TTL，减少 Redis 调用
+        self._cache: dict[str, tuple[FeatureFlagConfig, float]] = {}
+        self._cache_ttl = 5.0  # 秒，默认5秒本地缓存
+        self._lock = threading.Lock()
+
+    def _redis_key(self, name: str) -> str:
+        return f"{self._key_prefix}{name}"
+
+    def _parse_whitelist(self, raw: Optional[str]) -> Set[str]:
+        if not raw:
+            return set()
+        try:
+            data = json.loads(raw)
+            return set(data) if isinstance(data, list) else set()
+        except (json.JSONDecodeError, TypeError):
+            return set()
+
+    def get(self, name: str) -> FeatureFlagConfig:
+        now = time.time()
+
+        # 先查本地缓存
+        with self._lock:
+            cached = self._cache.get(name)
+            if cached and now - cached[1] < self._cache_ttl:
+                return cached[0]
+
+        # 从 Redis 读取
+        try:
+            key = self._redis_key(name)
+            data = self._redis.hgetall(key)
+            if not data:
+                config = FeatureFlagConfig(name=name, enabled=False)
+            else:
+                config = FeatureFlagConfig(
+                    name=name,
+                    enabled=(data.get("enabled", "0") in ("1", "true", "True")),
+                    percentage=int(data.get("percentage", 0)),
+                    whitelist=self._parse_whitelist(data.get("whitelist")),
+                )
+
+            # 写入本地缓存
+            with self._lock:
+                self._cache[name] = (config, now)
+
+            return config
+        except Exception as exc:
+            logger.warning("Failed to get feature flag %s from Redis: %s", name, exc)
+            # Redis 不可用时返回默认值（关闭），不影响业务
+            return FeatureFlagConfig(name=name, enabled=False)
+
+    def set(self, config: FeatureFlagConfig) -> None:
+        key = self._redis_key(config.name)
+        self._redis.hset(
+            key,
+            mapping={
+                "enabled": "1" if config.enabled else "0",
+                "percentage": str(config.percentage),
+                "whitelist": json.dumps(sorted(config.whitelist), ensure_ascii=False),
+            },
+        )
+        # 失效本地缓存
+        with self._lock:
+            self._cache.pop(config.name, None)
+
+    def delete(self, name: str) -> bool:
+        key = self._redis_key(name)
+        result = self._redis.delete(key)
+        with self._lock:
+            self._cache.pop(name, None)
+        return bool(result)
+
+    def list_all(self) -> dict[str, FeatureFlagConfig]:
+        pattern = f"{self._key_prefix}*"
+        result: dict[str, FeatureFlagConfig] = {}
+        try:
+            cursor = 0
+            while True:
+                cursor, keys = self._redis.scan(cursor=cursor, match=pattern, count=100)
+                for key in keys:
+                    name = key[len(self._key_prefix) :]
+                    result[name] = self.get(name)
+                if cursor == 0:
+                    break
+        except Exception as exc:
+            logger.warning("Failed to list feature flags from Redis: %s", exc)
+        return result
+
+    def invalidate_cache(self, name: Optional[str] = None) -> None:
+        """手动失效本地缓存。"""
+        with self._lock:
+            if name:
+                self._cache.pop(name, None)
+            else:
+                self._cache.clear()

tests/unit/test_feature_flag.py

@@ -0,0 +1,424 @@
+"""Feature Flag 单元测试。
+
+测试 FeatureFlagConfig、InMemoryFeatureFlagStore、RenderEngineResolver 的核心逻辑。
+"""
+
+from __future__ import annotations
+
+import time
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from packages.adapters.redis.feature_flag_store import (
+    FeatureFlagConfig,
+    InMemoryFeatureFlagStore,
+)
+
+# ── FeatureFlagConfig 测试 ──────────────────────────────────────────────────
+
+
+class TestFeatureFlagConfig:
+    """FeatureFlagConfig 核心逻辑测试。"""
+
+    def test_default_disabled(self):
+        """默认配置为关闭状态。"""
+        config = FeatureFlagConfig(name="test_flag")
+        assert config.enabled is False
+        assert config.percentage == 0
+        assert config.whitelist == set()
+        assert config.is_active() is False
+        assert config.is_active("user1") is False
+
+    def test_global_enabled_100_percent(self):
+        """100% + 启用 = 全部命中。"""
+        config = FeatureFlagConfig(name="test_flag", enabled=True, percentage=100)
+        assert config.is_active() is True
+        assert config.is_active("user1") is True
+        assert config.is_active("any_user") is True
+
+    def test_global_enabled_0_percent_no_whitelist(self):
+        """启用但 0% 且无白名单 = 不命中。"""
+        config = FeatureFlagConfig(name="test_flag", enabled=True, percentage=0)
+        assert config.is_active() is False
+        assert config.is_active("user1") is False
+
+    def test_whitelist_takes_priority(self):
+        """白名单优先级高于百分比。"""
+        config = FeatureFlagConfig(
+            name="test_flag",
+            enabled=True,
+            percentage=0,
+            whitelist={"user1", "user2"},
+        )
+        assert config.is_active("user1") is True
+        assert config.is_active("user2") is True
+        assert config.is_active("user3") is False
+
+    def test_whitelist_with_percentage(self):
+        """白名单用户即使百分比为0也命中，非白名单按百分比。"""
+        config = FeatureFlagConfig(
+            name="test_flag",
+            enabled=True,
+            percentage=100,  # 100% 所有人命中
+            whitelist={"user1"},
+        )
+        assert config.is_active("user1") is True
+        assert config.is_active("user999") is True  # 100% 命中
+
+    def test_percentage_consistency_same_user(self):
+        """同一用户多次调用结果一致（哈希确定性）。"""
+        config = FeatureFlagConfig(name="test_flag", enabled=True, percentage=50)
+        results = [config.is_active("user_fixed") for _ in range(100)]
+        assert all(r == results[0] for r in results)
+
+    def test_percentage_different_users_distributed(self):
+        """不同用户分布大致符合百分比（统计检验，宽松阈值）。"""
+        config = FeatureFlagConfig(name="test_flag", enabled=True, percentage=50)
+        active_count = sum(1 for i in range(1000) if config.is_active(f"user_{i}"))
+        # 50% 上下浮动 10% 都算合理
+        assert 400 <= active_count <= 600, f"Expected ~500, got {active_count}"
+
+    def test_percentage_boundary_0_and_100(self):
+        """0% 和 100% 的边界情况。"""
+        config_0 = FeatureFlagConfig(name="test", enabled=True, percentage=0)
+        config_100 = FeatureFlagConfig(name="test", enabled=True, percentage=100)
+
+        for i in range(100):
+            assert config_0.is_active(f"user_{i}") is False
+            assert config_100.is_active(f"user_{i}") is True
+
+    def test_disabled_ignores_all_other_settings(self):
+        """关闭时忽略白名单和百分比。"""
+        config = FeatureFlagConfig(
+            name="test_flag",
+            enabled=False,
+            percentage=100,
+            whitelist={"user1"},
+        )
+        assert config.is_active("user1") is False
+        assert config.is_active() is False
+
+    def test_none_identifier_with_percentage(self):
+        """无 identifier 时按随机比例（0% 和 100% 是确定的）。"""
+        config_0 = FeatureFlagConfig(name="test", enabled=True, percentage=0)
+        config_100 = FeatureFlagConfig(name="test", enabled=True, percentage=100)
+        assert config_0.is_active(None) is False
+        assert config_100.is_active(None) is True
+
+    def test_to_dict_and_from_dict(self):
+        """序列化和反序列化对称。"""
+        original = FeatureFlagConfig(
+            name="test_flag",
+            enabled=True,
+            percentage=30,
+            whitelist={"user_a", "user_b", "user_c"},
+        )
+        data = original.to_dict()
+        restored = FeatureFlagConfig.from_dict(data)
+        assert restored.name == original.name
+        assert restored.enabled == original.enabled
+        assert restored.percentage == original.percentage
+        assert restored.whitelist == original.whitelist
+
+    def test_from_dict_with_missing_fields(self):
+        """from_dict 缺失字段时使用默认值。"""
+        config = FeatureFlagConfig.from_dict({"name": "minimal"})
+        assert config.name == "minimal"
+        assert config.enabled is False
+        assert config.percentage == 0
+        assert config.whitelist == set()
+
+
+# ── InMemoryFeatureFlagStore 测试 ───────────────────────────────────────────
+
+
+class TestInMemoryFeatureFlagStore:
+    """内存存储实现测试。"""
+
+    def test_get_nonexistent_returns_default(self):
+        """获取不存在的 flag 返回默认配置（关闭）。"""
+        store = InMemoryFeatureFlagStore()
+        config = store.get("nonexistent")
+        assert config.name == "nonexistent"
+        assert config.enabled is False
+
+    def test_set_and_get(self):
+        """设置后可以读取。"""
+        store = InMemoryFeatureFlagStore()
+        config = FeatureFlagConfig(name="test", enabled=True, percentage=50, whitelist={"u1"})
+        store.set(config)
+
+        got = store.get("test")
+        assert got.enabled is True
+        assert got.percentage == 50
+        assert got.whitelist == {"u1"}
+
+    def test_delete_existing(self):
+        """删除存在的 flag 返回 True。"""
+        store = InMemoryFeatureFlagStore()
+        store.set(FeatureFlagConfig(name="test", enabled=True))
+        assert store.delete("test") is True
+        assert store.get("test").enabled is False
+
+    def test_delete_nonexistent(self):
+        """删除不存在的 flag 返回 False。"""
+        store = InMemoryFeatureFlagStore()
+        assert store.delete("nonexistent") is False
+
+    def test_list_all(self):
+        """列出所有 flag。"""
+        store = InMemoryFeatureFlagStore()
+        store.set(FeatureFlagConfig(name="flag_a", enabled=True))
+        store.set(FeatureFlagConfig(name="flag_b", percentage=10))
+
+        all_flags = store.list_all()
+        assert len(all_flags) == 2
+        assert "flag_a" in all_flags
+        assert "flag_b" in all_flags
+        assert all_flags["flag_a"].enabled is True
+
+    def test_is_active_convenience(self):
+        """is_active 便捷方法。"""
+        store = InMemoryFeatureFlagStore()
+        store.set(FeatureFlagConfig(name="render", enabled=True, percentage=0, whitelist={"vip_user"}))
+        assert store.is_active("render", "vip_user") is True
+        assert store.is_active("render", "normal_user") is False
+        assert store.is_active("nonexistent") is False
+
+
+# ── RenderEngineResolver 测试 ───────────────────────────────────────────────
+
+
+class TestRenderEngineResolver:
+    """渲染引擎选择器测试。"""
+
+    def test_default_legacy_when_flag_disabled(self):
+        """flag 关闭时使用默认引擎（legacy）。"""
+        store = InMemoryFeatureFlagStore()
+        resolver = self._make_resolver(store=store, default="legacy")
+        assert resolver.get_engine() == "legacy"
+        assert resolver.get_engine("user1") == "legacy"
+
+    def test_default_unified_when_flag_disabled(self):
+        """flag 关闭但默认值是 unified 时返回 unified。"""
+        store = InMemoryFeatureFlagStore()
+        resolver = self._make_resolver(store=store, default="unified")
+        assert resolver.get_engine() == "unified"
+
+    def test_whitelist_user_uses_unified(self):
+        """白名单用户走新引擎。"""
+        store = InMemoryFeatureFlagStore()
+        store.set(
+            FeatureFlagConfig(
+                name="render_engine",
+                enabled=True,
+                percentage=0,
+                whitelist={"beta_tester"},
+            )
+        )
+        resolver = self._make_resolver(store=store, default="legacy")
+        assert resolver.get_engine("beta_tester") == "unified"
+        assert resolver.get_engine("normal_user") == "legacy"
+
+    def test_100_percent_all_unified(self):
+        """100% 时所有用户走新引擎。"""
+        store = InMemoryFeatureFlagStore()
+        store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100))
+        resolver = self._make_resolver(store=store, default="legacy")
+        for i in range(50):
+            assert resolver.get_engine(f"user_{i}") == "unified"
+
+    def test_invalid_default_engine_fallback(self):
+        """无效默认值回退到 legacy。"""
+        store = InMemoryFeatureFlagStore()
+        resolver = self._make_resolver(store=store, default="invalid_value")
+        assert resolver.get_engine() == "legacy"
+
+    def test_should_use_unified_helper(self):
+        """should_use_unified 便捷方法。"""
+        store = InMemoryFeatureFlagStore()
+        store.set(
+            FeatureFlagConfig(
+                name="render_engine",
+                enabled=True,
+                percentage=0,
+                whitelist={"user_a"},
+            )
+        )
+        resolver = self._make_resolver(store=store)
+        assert resolver.should_use_unified("user_a") is True
+        assert resolver.should_use_unified("user_b") is False
+
+    def test_config_snapshot(self):
+        """配置快照。"""
+        store = InMemoryFeatureFlagStore()
+        store.set(
+            FeatureFlagConfig(
+                name="render_engine",
+                enabled=True,
+                percentage=30,
+                whitelist={"u1", "u2"},
+            )
+        )
+        resolver = self._make_resolver(store=store)
+        snapshot = resolver.get_config_snapshot()
+        assert snapshot["flag_name"] == "render_engine"
+        assert snapshot["enabled"] is True
+        assert snapshot["percentage"] == 30
+        assert snapshot["whitelist"] == ["u1", "u2"]
+
+    def test_set_flag_updates_config(self):
+        """通过 set_flag 修改后立即生效。"""
+        store = InMemoryFeatureFlagStore()
+        resolver = self._make_resolver(store=store, default="legacy")
+
+        # 初始：关闭
+        assert resolver.get_engine("user1") == "legacy"
+
+        # 开启 100%
+        resolver.set_flag(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100))
+        assert resolver.get_engine("user1") == "unified"
+
+        # 关闭
+        resolver.set_flag(FeatureFlagConfig(name="render_engine", enabled=False))
+        assert resolver.get_engine("user1") == "legacy"
+
+    def test_force_refresh(self):
+        """强制刷新不报错。"""
+        store = InMemoryFeatureFlagStore()
+        resolver = self._make_resolver(store=store)
+        resolver.force_refresh()  # 不抛异常即可
+
+    def test_does_not_affect_in_flight_tasks(self):
+        """
+        热更新不影响在途任务验证：
+        任务开始时确定引擎，中途配置变更不改变当前任务的引擎选择。
+        （这是通过"每次调用 get_engine 时读取当前配置"来保证的，
+        任务开始时调用一次拿到结果，之后不再变化）
+        """
+        store = InMemoryFeatureFlagStore()
+        store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100))
+        resolver = self._make_resolver(store=store, default="legacy")
+
+        # 模拟任务开始时获取引擎
+        engine_at_start = resolver.get_engine("user1")
+        assert engine_at_start == "unified"
+
+        # 任务进行中关闭 flag
+        store.set(FeatureFlagConfig(name="render_engine", enabled=False))
+        resolver.force_refresh()
+
+        # 在途任务持有的 engine_at_start 仍然是 unified（不随配置变化）
+        assert engine_at_start == "unified"
+        # 新任务会拿到 legacy
+        assert resolver.get_engine("user1") == "legacy"
+
+    # ── 辅助方法 ──
+
+    @staticmethod
+    def _make_resolver(store=None, default="legacy"):
+        from apps.worker.video_processing.render_engine_resolver import (
+            RenderEngineResolver,
+        )
+
+        return RenderEngineResolver(
+            default_engine=default,
+            store=store or InMemoryFeatureFlagStore(),
+            refresh_interval=9999,  # 测试时禁用自动刷新
+        )
+
+
+# ── RedisFeatureFlagStore 降级测试（无 Redis 环境） ───────────────────────
+
+
+class TestRedisStoreDegradation:
+    """Redis 不可用时的降级行为测试。"""
+
+    def test_get_returns_default_when_redis_unavailable(self):
+        """Redis 连接失败时返回默认关闭配置，不抛异常。"""
+        import importlib
+
+        from packages.adapters.redis import feature_flag_store as ff_module
+
+        # 模拟 redis 模块不存在的场景不好做，这里直接测试异常捕获逻辑
+        store = ff_module.RedisFeatureFlagStore.__new__(ff_module.RedisFeatureFlagStore)
+        store._redis = MagicMock()
+        store._redis.hgetall.side_effect = ConnectionError("Redis down")
+        store._key_prefix = ff_module.FEATURE_FLAG_REDIS_PREFIX
+        store._cache = {}
+        store._cache_ttl = 5.0
+        import threading
+
+        store._lock = threading.Lock()
+
+        config = store.get("render_engine")
+        assert config.enabled is False
+        assert config.name == "render_engine"
+
+    def test_list_all_returns_empty_on_redis_error(self):
+        """Redis 错误时 list_all 返回空字典。"""
+        import importlib
+
+        from packages.adapters.redis import feature_flag_store as ff_module
+
+        store = ff_module.RedisFeatureFlagStore.__new__(ff_module.RedisFeatureFlagStore)
+        store._redis = MagicMock()
+        store._redis.scan.side_effect = ConnectionError("Redis down")
+        store._key_prefix = ff_module.FEATURE_FLAG_REDIS_PREFIX
+        store._cache = {}
+        store._cache_ttl = 5.0
+        import threading
+
+        store._lock = threading.Lock()
+
+        result = store.list_all()
+        assert result == {}
+
+
+class TestRedisStoreListAll:
+    """RedisFeatureFlagStore list_all 正常路径测试。"""
+
+    def _make_store(self):
+        from packages.adapters.redis import feature_flag_store as ff_module
+
+        store = ff_module.RedisFeatureFlagStore.__new__(ff_module.RedisFeatureFlagStore)
+        store._redis = MagicMock()
+        store._key_prefix = ff_module.FEATURE_FLAG_REDIS_PREFIX
+        store._cache = {}
+        store._cache_ttl = 5.0
+        import threading
+
+        store._lock = threading.Lock()
+        return store
+
+    def test_list_all_scan_with_match_param(self):
+        """list_all 调用 redis.scan 时使用正确的 match 参数名。"""
+        store = self._make_store()
+        prefix = store._key_prefix
+
+        # 模拟 scan 返回 2 个 key，分 2 次游标
+        store._redis.scan.side_effect = [
+            (10, [f"{prefix}render_engine", f"{prefix}other_flag"]),
+            (0, []),
+        ]
+        # 模拟 hgetall 返回配置
+        store._redis.hgetall.return_value = {
+            b"enabled": b"true",
+            b"percentage": b"50",
+            b"whitelist": b'["user1","user2"]',
+        }
+
+        result = store.list_all()
+
+        # 验证 scan 被调用了 2 次（游标遍历）
+        assert store._redis.scan.call_count == 2
+        # 验证参数名是 match（不是 match_pattern）
+        first_call_kwargs = store._redis.scan.call_args_list[0][1]
+        assert "match" in first_call_kwargs
+        assert "match_pattern" not in first_call_kwargs
+        assert first_call_kwargs["match"] == f"{prefix}*"
+        # 验证返回了 2 个 flag
+        assert len(result) == 2
+        assert "render_engine" in result
+        assert "other_flag" in result