feat(ci): 新增发布与灰度部署脚本

- release.sh: 一键发布脚本（打tag + 触发CI构建 + 灰度提示）
- gray_deploy.sh: 灰度发布脚本（canary容器 + nginx权重切流）
- rollback_gray.sh: 灰度回滚脚本（nginx回滚 + canary清理）
- 更新 rollback.sh 占位符为生产环境适配版

灰度范围：API + Web（Nginx upstream 权重）
Worker 暂不支持灰度（队列消费无法按比例切流）

.gitea/workflows/ci-cd.yml

@@ -1047,6 +1047,52 @@ jobs:
     needs: build-production-runtime-images
 
     steps:
+    - name: Checkout code
+      shell: sh
+      env:
+        GITHUB_TOKEN: ${{ github.token }}
+      run: |
+        set -eu
+        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, '.')
+        PY
+
     - name: Install SSH client
       shell: sh
       run: |
@@ -1081,10 +1127,16 @@ jobs:
           key_path="$HOME/.ssh/xiaoxia_runtime_builder"
           echo "Using key: $key_path (home key)"
         elif [ -n "${PRODUCTION_SSH_KEY:-}" ]; then
-          key_path="$HOME/.ssh/id_ed25519"
+          key_path="$HOME/.ssh/production_deploy_key"
           printf '%s\n' "$PRODUCTION_SSH_KEY" > "$key_path"
           chmod 600 "$key_path"
           echo "Using key from PRODUCTION_SSH_KEY secret"
+        elif [ -f "$HOME/.ssh/id_ed25519" ]; then
+          key_path="$HOME/.ssh/id_ed25519"
+          echo "Using key: $key_path (default id_ed25519)"
+        elif [ -f /root/.ssh/id_ed25519 ]; then
+          key_path="/root/.ssh/id_ed25519"
+          echo "Using key: $key_path (root id_ed25519)"
         else
           echo "ERROR: No SSH key available"
           ls -la ~/.ssh/ 2>/dev/null || true

alembic/versions/040_add_playback_speed_to_edit_plan_clips.py

@@ -0,0 +1,29 @@
+"""add playback_speed to edit_plan_clips
+
+Revision ID: 040_playback_speed
+Revises: 039_transition_duration
+Create Date: 2026-07-14 10:00:00.000000
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "040_playback_speed"
+down_revision = "039_transition_duration"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "edit_plan_clips",
+        sa.Column("playback_speed", sa.Float(), nullable=False, server_default="1.0"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("edit_plan_clips", "playback_speed")

apps/api/app/api/routes/edit_plans.py

@@ -211,6 +211,7 @@ class _PlanClipItem(BaseModel):
     duration: float
     transition_effect: str
     transition_duration: float
+    playback_speed: float = 1.0
     status: str
     config: Optional[dict[str, Any]] = None
     created_at: datetime

apps/api/app/services/edit_plan_service.py

@@ -282,6 +282,7 @@ class EditPlanService:
         duration: float = 0.0,
         transition_effect: str = "cut",
         transition_duration: float = 0.0,
+        playback_speed: float = 1.0,
         config: Optional[dict[str, Any]] = None,
     ) -> EditPlanClip:
         """创建片段
@@ -303,6 +304,7 @@ class EditPlanService:
             duration=duration,
             transition_effect=transition_effect,
             transition_duration=transition_duration,
+            playback_speed=playback_speed,
             config=config,
         )
         created = self._clip_repo.create(clip)
@@ -327,6 +329,7 @@ class EditPlanService:
         duration: Optional[float] = None,
         transition_effect: Optional[str] = None,
         transition_duration: Optional[float] = None,
+        playback_speed: Optional[float] = None,
         config: Optional[dict[str, Any]] = None,
     ) -> EditPlanClip:
         """更新片段
@@ -336,6 +339,15 @@ class EditPlanService:
         """
         existing = self.get_clip_or_raise(clip_id)
 
+        # 速度边界钳制
+        if playback_speed is not None:
+            if playback_speed <= 0:
+                playback_speed = 1.0
+            elif playback_speed < 0.25:
+                playback_speed = 0.25
+            elif playback_speed > 4.0:
+                playback_speed = 4.0
+
         updated = EditPlanClip(
             id=existing.id,
             plan_id=existing.plan_id,
@@ -352,6 +364,7 @@ class EditPlanService:
             transition_duration=(
                 transition_duration if transition_duration is not None else existing.transition_duration
             ),
+            playback_speed=playback_speed if playback_speed is not None else existing.playback_speed,
             status=existing.status,
             config=config if config is not None else existing.config,
             created_at=existing.created_at,

apps/web/e2e/core-generation.spec.ts

@@ -251,6 +251,9 @@ test.describe("Core generation flow", () => {
     await expect(page.locator(".xx-products-page")).toBeVisible({
       timeout: 15_000,
     });
+
+    // 清理所有路由，避免页面关闭时飞地API请求导致测试报错
+    await page.unrouteAll({ behavior: "ignoreErrors" });
   });
 
   test("generation task API creates and lists tasks", async ({ request }) => {

apps/worker/video_processing/render_audio.py

@@ -23,6 +23,7 @@ from typing import TYPE_CHECKING
 
 from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_has_audio, run_ffmpeg
 from video_processing.reverse_engine import ReverseConfig, ReverseEngine
+from video_processing.speed_engine import SpeedEngine
 
 if TYPE_CHECKING:
     from video_processing.unified_render_service import RenderLayer, ResolvedClip
@@ -225,53 +226,118 @@ def concat_main_audio(
         clip = clips[0]
         effective_duration = clip_effective_duration(clip)
         trim_start = getattr(clip, "start_time", 0) or 0
-        # 最终时长：取 clip 有效时长和视频总时长的较小值
-        # （视频总时长由主图层决定，但单 clip 场景下两者应该一致，仍做保护）
-        final_duration = effective_duration
+        speed = getattr(clip, "playback_speed", 1.0) or 1.0
+        if not isinstance(speed, (int, float)) or speed <= 0:
+            speed = 1.0
+
+        # 调速后时长
+        adjusted_duration = effective_duration / speed if abs(speed - 1.0) >= 1e-6 else effective_duration
+        # 最终时长：取调速后时长和视频总时长的较小值
+        final_duration = adjusted_duration
         if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration):
             final_duration = video_duration
 
         # 音频倒放
         reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
-        af_filters = []
-        if reverse_config.enabled and reverse_config.reverse_audio:
-            reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration)
-            if reverse_filter:
-                af_filters.append(reverse_filter)
+        has_reverse = reverse_config.enabled and reverse_config.reverse_audio
+        has_speed = abs(speed - 1.0) >= 1e-6
 
-        command = [
-            FFMPEG_BIN,
-            "-y",
-            "-i",
-            str(clip.local_path),
-            "-vn",
-            "-acodec",
-            "aac",
-            "-b:a",
-            "128k",
-        ]
-        if trim_start > 0:
-            command.extend(["-ss", f"{trim_start:.3f}"])
-        if af_filters:
-            command.extend(["-af", ",".join(af_filters)])
-        if final_duration > 0:
-            command.extend(["-t", f"{final_duration:.3f}"])
-        command.append(str(output_path))
-        run_ffmpeg(command)
+        if not has_speed and not has_reverse:
+            # 无调速无倒放：简单命令行，-ss 裁剪更高效
+            command = [
+                FFMPEG_BIN,
+                "-y",
+                "-i",
+                str(clip.local_path),
+                "-vn",
+                "-acodec",
+                "aac",
+                "-b:a",
+                "128k",
+            ]
+            if trim_start > 0:
+                command.extend(["-ss", f"{trim_start:.3f}"])
+            if final_duration > 0:
+                command.extend(["-t", f"{final_duration:.3f}"])
+            command.append(str(output_path))
+            run_ffmpeg(command)
+        else:
+            # 有调速或倒放：用 filter_complex
+            speed_engine = SpeedEngine()
+            audio_filters = []
+            if effective_duration > 0:
+                audio_filters.append(f"atrim=start={trim_start:.3f}:duration={effective_duration:.3f}")
+                audio_filters.append("asetpts=PTS-STARTPTS")
+
+            # 音频调速
+            if has_speed:
+                from video_processing.speed_engine import SpeedConfig
+
+                config = SpeedConfig(speed=float(speed))
+                config.clamp()
+                atempo_filter = speed_engine.build_audio_filter(config)
+                if atempo_filter:
+                    audio_filters.append(atempo_filter)
+
+            # 音频倒放
+            if has_reverse:
+                reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration)
+                if reverse_filter:
+                    audio_filters.append(reverse_filter)
+
+            filter_parts: list[str] = [f"[0:a]{','.join(audio_filters)}[outa]"]
+            if video_duration > 0 and final_duration < adjusted_duration:
+                filter_parts.append(f"[outa]atrim=0:{final_duration:.3f}[final_audio]")
+                final_label = "final_audio"
+            else:
+                final_label = "outa"
+
+            filter_complex = ";".join(filter_parts)
+            command = [
+                FFMPEG_BIN,
+                "-y",
+                "-i",
+                str(clip.local_path),
+                "-filter_complex",
+                filter_complex,
+                "-map",
+                f"[{final_label}]",
+                "-acodec",
+                "aac",
+                "-b:a",
+                "128k",
+                str(output_path),
+            ]
+            run_ffmpeg(command)
         return
 
     # 多 clip，用 filter_complex concat
     input_args: list[str] = []
     filter_parts: list[str] = []
+    speed_engine = SpeedEngine()
 
     for i, clip in enumerate(clips):
         input_args.extend(["-i", str(clip.local_path)])
         effective_duration = clip_effective_duration(clip)
         trim_start = getattr(clip, "start_time", 0) or 0
+        speed = getattr(clip, "playback_speed", 1.0) or 1.0
+        if not isinstance(speed, (int, float)) or speed <= 0:
+            speed = 1.0
+
         audio_filters: list[str] = []
         if effective_duration > 0:
             audio_filters.append(f"atrim=start={trim_start:.3f}:duration={effective_duration:.3f}")
             audio_filters.append("asetpts=PTS-STARTPTS")
+
+        # 音频调速 — atempo 多级串联
+        if abs(speed - 1.0) >= 1e-6:
+            from video_processing.speed_engine import SpeedConfig
+
+            config = SpeedConfig(speed=float(speed))
+            config.clamp()
+            atempo_filter = speed_engine.build_audio_filter(config)
+            if atempo_filter:
+                audio_filters.append(atempo_filter)
         else:
             audio_filters.append("asetpts=PTS-STARTPTS")
 

apps/worker/video_processing/speed_engine.py

@@ -0,0 +1,167 @@
+"""视频调速引擎 — 基于 FFmpeg setpts + atempo 的速度调整能力。
+
+支持：
+- 0.25x ~ 4x 变速范围
+- 视频调速（setpts）
+- 音频调速（atempo，多级串联处理超范围值）
+- 音调修正（pitch_correct，默认开启）
+- 边界自动钳制，不阻断渲染
+"""
+
+from dataclasses import dataclass
+from typing import Optional
+
+# ─── 常量 ───────────────────────────────────────────────
+MIN_SPEED = 0.25
+MAX_SPEED = 4.0
+DEFAULT_SPEED = 1.0
+
+# atempo 单级有效范围
+_ATEMPO_MIN = 0.5
+_ATEMPO_MAX = 2.0
+
+
+@dataclass
+class SpeedConfig:
+    """调速配置。
+
+    Attributes:
+        speed: 播放速度，0.25~4.0，1.0 为原速
+        pitch_correct: 是否保持音调（默认 True，用 atempo 时间拉伸算法）
+    """
+
+    speed: float = DEFAULT_SPEED
+    pitch_correct: bool = True
+
+    @classmethod
+    def parse(cls, data: Optional[dict]) -> "SpeedConfig":
+        """从 dict 解析配置，无效值回退到默认。"""
+        if not data or not isinstance(data, dict):
+            return cls()
+
+        speed = data.get("speed", DEFAULT_SPEED)
+        if not isinstance(speed, (int, float)):
+            speed = DEFAULT_SPEED
+
+        pitch_correct = data.get("pitch_correct", True)
+        if not isinstance(pitch_correct, bool):
+            pitch_correct = True
+
+        config = cls(speed=float(speed), pitch_correct=pitch_correct)
+        config.clamp()
+        return config
+
+    def clamp(self) -> None:
+        """将速度钳制到合法范围。"""
+        if self.speed <= 0:
+            self.speed = DEFAULT_SPEED
+        elif self.speed < MIN_SPEED:
+            self.speed = MIN_SPEED
+        elif self.speed > MAX_SPEED:
+            self.speed = MAX_SPEED
+
+    @property
+    def is_original(self) -> bool:
+        """是否原速（无需调速）。"""
+        return abs(self.speed - 1.0) < 1e-6
+
+
+class SpeedEngine:
+    """调速引擎 — 生成 FFmpeg 调速滤镜链。
+
+    用法：
+        engine = SpeedEngine()
+        video_filter = engine.build_video_filter(config)
+        audio_filter = engine.build_audio_filter(config)
+        new_duration = engine.adjust_duration(duration, config)
+    """
+
+    def build_video_filter(self, config: SpeedConfig) -> str:
+        """生成视频调速滤镜字符串。
+
+        返回 setpts 滤镜表达式，原速时返回空字符串。
+        """
+        if config.is_original:
+            return ""
+        # setpts=PTS/speed — speed>1 加速，speed<1 减速
+        return f"setpts=PTS/{config.speed:.4f}"
+
+    def build_audio_filter(self, config: SpeedConfig) -> str:
+        """生成音频调速滤镜字符串。
+
+        atempo 单级范围 0.5~2.0，超出范围时自动多级串联：
+        - 0.25x → atempo=0.5,atempo=0.5
+        - 4x → atempo=2.0,atempo=2.0
+        - 0.3x → atempo=0.5,atempo=0.6
+        - 3x → atempo=2.0,atempo=1.5
+
+        原速时返回空字符串。
+        """
+        if config.is_original:
+            return ""
+
+        speed = config.speed
+        stages: list[float] = self._split_atempo_stages(speed)
+        return ",".join(f"atempo={s:.4f}" for s in stages)
+
+    @staticmethod
+    def _split_atempo_stages(speed: float) -> list[float]:
+        """将速度拆分为多级 atempo 串联，每级都在 [0.5, 2.0] 范围内。"""
+        if _ATEMPO_MIN <= speed <= _ATEMPO_MAX:
+            return [speed]
+
+        stages: list[float] = []
+        remaining = speed
+
+        # 加速场景（speed > 2.0）
+        if speed > _ATEMPO_MAX:
+            while remaining > _ATEMPO_MAX:
+                stages.append(_ATEMPO_MAX)
+                remaining /= _ATEMPO_MAX
+            stages.append(remaining)
+
+        # 减速场景（speed < 0.5）
+        else:
+            while remaining < _ATEMPO_MIN:
+                stages.append(_ATEMPO_MIN)
+                remaining /= _ATEMPO_MIN
+            stages.append(remaining)
+
+        return stages
+
+    def adjust_duration(self, original_duration: float, config: SpeedConfig) -> float:
+        """计算调速后的时长。
+
+        加速 → 时长变短；减速 → 时长变长。
+        """
+        if config.is_original or original_duration <= 0:
+            return original_duration
+        return original_duration / config.speed
+
+    def build_clip_speed_filter(
+        self,
+        speed: float,
+        pitch_correct: bool = True,
+    ) -> tuple[str, str, SpeedConfig]:
+        """便捷方法：从单一 speed 值生成视频+音频滤镜。
+
+        返回 (video_filter, audio_filter, config)。
+        """
+        config = SpeedConfig(speed=speed, pitch_correct=pitch_correct)
+        config.clamp()
+        return (
+            self.build_video_filter(config),
+            self.build_audio_filter(config),
+            config,
+        )
+
+    @staticmethod
+    def resolve_clip_speed(
+        clip_config: dict,
+        global_speed: float = DEFAULT_SPEED,
+    ) -> float:
+        """从 clip config 中解析 playback_speed，0 或缺失则使用全局速度。"""
+        speed = clip_config.get("playback_speed", 0) if clip_config else 0
+        if not isinstance(speed, (int, float)) or speed <= 0:
+            return global_speed
+        return float(speed)

apps/worker/video_processing/unified_render_service.py

@@ -45,6 +45,7 @@ from video_processing.pip_engine import PiPConfig, PiPEngine, PiPLayerConfig
 from video_processing.render_audio import RenderContext, merge_audio_video, mix_audio
 from video_processing.render_subtitles import generate_ass_subtitles
 from video_processing.reverse_engine import ReverseConfig, ReverseEngine
+from video_processing.speed_engine import SpeedConfig, SpeedEngine
 from video_processing.sticker_engine import StickerEngine, parse_stickers_from_config
 from video_processing.subtitle_generator import generate_ass_from_timeline
 from video_processing.transition_engine import TransitionEngine
@@ -73,6 +74,7 @@ class ResolvedClip:
     duration: float = 0.0  # 0 表示使用素材完整时长
     transition_effect: str = "cut"
     transition_duration: float = 0.0  # 0 表示使用全局默认值
+    playback_speed: float = 1.0  # 0 或 1.0 表示原速
     config: dict[str, Any] = field(default_factory=dict)
 
     # 运行时填充
@@ -186,6 +188,7 @@ class UnifiedRenderService:
         self.asr_service = asr_service
         self.bgm_path = bgm_path
         self._transition_engine = TransitionEngine(default_duration=transition_duration)
+        self._speed_engine = SpeedEngine()
 
     def render(self) -> RenderResult:
         """执行渲染，返回 RenderResult.
@@ -493,7 +496,7 @@ class UnifiedRenderService:
         if not main_layer or not main_layer.clips:
             return 0.0
 
-        total = sum(UnifiedRenderService._clip_effective_duration(c) for c in main_layer.clips)
+        total = sum(UnifiedRenderService._clip_adjusted_duration(c) for c in main_layer.clips)
 
         # 减去转场重叠时间（粗略估算）
         n_clips = len(main_layer.clips)
@@ -1197,6 +1200,7 @@ class UnifiedRenderService:
                 duration=final_duration,
                 transition_effect=clip.transition_effect or "cut",
                 transition_duration=getattr(clip, "transition_duration", 0.0) or 0.0,
+                playback_speed=getattr(clip, "playback_speed", 1.0) or 1.0,
                 config=clip_config,
                 actual_duration=actual_duration,
                 trim_config=effective_trim,
@@ -1294,6 +1298,11 @@ class UnifiedRenderService:
                     filters.append(f"trim=duration={effective_duration:.3f}")
                 filters.append("setpts=PTS-STARTPTS")
 
+            # 调速 — 基于 setpts 改变播放速度
+            speed = UnifiedRenderService._clip_speed(clip)
+            if abs(speed - 1.0) >= 1e-6:
+                filters.append(f"setpts=PTS/{speed:.4f}")
+
             # 倒放滤镜（在 trim 之后、scale 之前应用）
             reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
             if reverse_config.enabled and reverse_config.reverse_video:
@@ -1351,8 +1360,8 @@ class UnifiedRenderService:
         for layer in layers:
             layer_clip_indices = [all_clips.index(c) for c in layer.clips]
             layer_labels = [preprocessed_labels[i] for i in layer_clip_indices]
-            # 使用 trim 后的有效时长，与 Step 1 的 trim=duration 保持一致
-            layer_durations = [UnifiedRenderService._clip_effective_duration(all_clips[i]) for i in layer_clip_indices]
+            # 使用调速后的实际时长，与 Step 1 的调速处理保持一致
+            layer_durations = [UnifiedRenderService._clip_adjusted_duration(all_clips[i]) for i in layer_clip_indices]
             layer_transitions = [all_clips[i].transition_effect for i in layer_clip_indices]
             layer_transition_durations = [all_clips[i].transition_duration for i in layer_clip_indices]
 
@@ -1597,7 +1606,7 @@ class UnifiedRenderService:
 
     @staticmethod
     def _clip_effective_duration(clip: ResolvedClip) -> float:
-        """计算 clip 的有效时长."""
+        """计算 clip 的有效时长（原速 trim 后时长）."""
         if clip.duration > 0:
             return min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration
         return clip.actual_duration if clip.actual_duration > 0 else 0.0
@@ -1694,3 +1703,20 @@ class UnifiedRenderService:
         )
 
         return new_filter, new_input_args
+
+    @staticmethod
+    def _clip_speed(clip: ResolvedClip) -> float:
+        """获取 clip 的播放速度，无效值回退到 1.0."""
+        speed = getattr(clip, "playback_speed", 1.0)
+        if not isinstance(speed, (int, float)) or speed <= 0:
+            return 1.0
+        return float(speed)
+
+    @staticmethod
+    def _clip_adjusted_duration(clip: ResolvedClip) -> float:
+        """计算调速后的 clip 实际时长（用于拼接计算）."""
+        base = UnifiedRenderService._clip_effective_duration(clip)
+        speed = UnifiedRenderService._clip_speed(clip)
+        if abs(speed - 1.0) < 1e-6:
+            return base
+        return base / speed

docs/schema-metadata-snapshot.json

@@ -866,6 +866,14 @@
           "type": "FLOAT",
           "unique": false
         },
+        {
+          "index": false,
+          "name": "playback_speed",
+          "nullable": false,
+          "primary_key": false,
+          "type": "FLOAT",
+          "unique": false
+        },
         {
           "index": true,
           "name": "status",

packages/adapters/sqlalchemy_impl/edit_plan_clip_repository.py

@@ -55,6 +55,7 @@ class SQLAlchemyEditPlanClipRepository:
             duration=clip.duration,
             transition_effect=clip.transition_effect,
             transition_duration=clip.transition_duration,
+            playback_speed=clip.playback_speed,
             status=clip.status,
             config=clip.config,
         )
@@ -78,6 +79,7 @@ class SQLAlchemyEditPlanClipRepository:
         model.duration = clip.duration
         model.transition_effect = clip.transition_effect
         model.transition_duration = clip.transition_duration
+        model.playback_speed = clip.playback_speed
         model.status = clip.status
         model.config = clip.config
         model.updated_at = clip.updated_at
@@ -123,6 +125,7 @@ class SQLAlchemyEditPlanClipRepository:
             duration=model.duration or 0.0,
             transition_effect=model.transition_effect or "cut",
             transition_duration=getattr(model, "transition_duration", 0.0) or 0.0,
+            playback_speed=model.playback_speed or 1.0,
             status=EditPlanClipStatus(model.status) if model.status else EditPlanClipStatus.PENDING,
             config=model.config or {},
             created_at=model.created_at,

packages/adapters/sqlalchemy_impl/models.py

@@ -197,6 +197,7 @@ class EditPlanClipModel(Base):
     duration = Column(Float, nullable=False, default=0.0)
     transition_effect = Column(String(20), nullable=False, default="cut")
     transition_duration = Column(Float, nullable=False, default=0.0)
+    playback_speed = Column(Float, nullable=False, default=1.0)
     status = Column(String(20), nullable=False, default="pending", index=True)
     config = Column(JSON, nullable=False, default=dict)
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))

packages/domain/edit_plan_clip.py

@@ -51,6 +51,7 @@ class EditPlanClip:
     duration: float = 0.0
     transition_effect: str = "cut"
     transition_duration: float = 0.0  # 0 表示使用全局默认值
+    playback_speed: float = 1.0  # 0 或 1.0 表示原速，范围 0.25~4.0
     status: EditPlanClipStatus = EditPlanClipStatus.PENDING
     config: dict[str, Any] = field(default_factory=dict)
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@@ -70,6 +71,7 @@ class EditPlanClip:
         duration: float = 0.0,
         transition_effect: str = "cut",
         transition_duration: float = 0.0,
+        playback_speed: float = 1.0,
         config: dict[str, Any] | None = None,
     ) -> EditPlanClip:
         """创建剪辑计划片段"""
@@ -81,6 +83,13 @@ class EditPlanClip:
             raise ValueError("start_time 不能为负数")
         if duration < 0:
             raise ValueError("duration 不能为负数")
+        # 速度边界钳制
+        if playback_speed <= 0:
+            playback_speed = 1.0
+        elif playback_speed < 0.25:
+            playback_speed = 0.25
+        elif playback_speed > 4.0:
+            playback_speed = 4.0
 
         return cls(
             id=uuid4().hex,
@@ -94,6 +103,7 @@ class EditPlanClip:
             duration=duration,
             transition_effect=transition_effect.strip() or "cut",
             transition_duration=max(0.0, transition_duration),
+            playback_speed=playback_speed,
             status=EditPlanClipStatus.PENDING,
             config=config or {},
         )

scripts/gray_deploy.sh

@@ -0,0 +1,227 @@
+#!/bin/bash
+# 灰度发布脚本：在生产服务器上启动 canary 版本，通过 Nginx 权重切流
+# 用法: ./scripts/gray_deploy.sh <版本号> <灰度百分比>
+#
+# 前提：
+#   - 在生产服务器上执行（或通过 SSH 管道执行）
+#   - 当前已有全量运行的 production 容器
+#   - Nginx 配置在 /etc/nginx/sites-enabled/00-xiaoxia-saas
+#
+# 灰度范围：API + Web（Worker 暂时全量升级，队列消费无法按比例切流）
+
+set -euo pipefail
+
+VERSION="${1:-}"
+GRAY_PCT="${2:-10}"
+
+if [[ -z "$VERSION" ]]; then
+    echo "用法: $0 <版本号> [灰度百分比]"
+    echo "示例: $0 v0.1.130 5"
+    exit 1
+fi
+
+REGISTRY="${REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
+NGINX_CONF="${NGINX_CONF:-/etc/nginx/sites-enabled/00-xiaoxia-saas}"
+ENV_FILE="${ENV_FILE:-/var/lib/xiaoxia-saas-production/.env}"
+GENERATED_DIR="${GENERATED_DIR:-/var/lib/xiaoxia-saas-production/generated}"
+LEGACY_ASSETS_DIR="${LEGACY_ASSETS_DIR:-/var/lib/xiaoxia-saas-production/legacy-assets}"
+
+# Canary 端口（与 production 错开）
+CANARY_API_PORT=18001
+CANARY_WEB_PORT=13002
+
+echo "============================================"
+echo "  灰度发布"
+echo "  新版本: $VERSION"
+echo "  灰度比例: ${GRAY_PCT}%"
+echo "  Canary API 端口: $CANARY_API_PORT"
+echo "  Canary Web 端口: $CANARY_WEB_PORT"
+echo "============================================"
+
+# 1. 检查环境
+if [[ ! -f "$ENV_FILE" ]]; then
+    echo "错误: 环境文件不存在: $ENV_FILE"
+    exit 1
+fi
+
+if [[ ! -f "$NGINX_CONF" ]]; then
+    echo "错误: Nginx 配置不存在: $NGINX_CONF"
+    exit 1
+fi
+
+# 2. 拉取新版本镜像
+echo ""
+echo ">>> 拉取新版本镜像..."
+for component in api web worker; do
+    echo "  拉取 $component:$VERSION ..."
+    docker pull "${REGISTRY}-${component}:${VERSION}" 2>&1 | tail -1
+done
+echo "  ✅ 镜像拉取完成"
+
+# 3. 启动 API Canary
+echo ""
+echo ">>> 启动 API Canary 容器..."
+CANARY_API="xiaoxia-api-canary"
+if docker ps -a --format '{{.Names}}' | grep -q "^${CANARY_API}$"; then
+    echo "  停止旧 canary..."
+    docker rm -f "$CANARY_API" >/dev/null 2>&1
+fi
+
+docker run -d \
+    --name "$CANARY_API" \
+    --env-file "$ENV_FILE" \
+    --network xiaoxia-net-production \
+    -p "127.0.0.1:${CANARY_API_PORT}:8000" \
+    -e APP_ENV=production \
+    -e APP_VERSION="$VERSION-canary" \
+    -e GENERATED_FILES_DIR=/app/generated \
+    -e GENERATED_FILES_URL_PREFIX=/generated-files \
+    -e PUBLIC_API_BASE_URL=https://saas-api.xiaoxiajianji.com \
+    -v "${GENERATED_DIR}:/app/generated" \
+    --restart unless-stopped \
+    --cpus 1 \
+    --memory 1g \
+    --health-cmd "python -c \"import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=5)\"" \
+    --health-interval 30s \
+    --health-timeout 10s \
+    --health-retries 3 \
+    --health-start-period 40s \
+    --log-driver json-file \
+    --log-opt max-size=50m \
+    --log-opt max-file=3 \
+    "${REGISTRY}-api:${VERSION}" >/dev/null
+
+echo "  ✅ API Canary 已启动（端口 $CANARY_API_PORT）"
+
+# 4. 启动 Web Canary
+echo ""
+echo ">>> 启动 Web Canary 容器..."
+CANARY_WEB="xiaoxia-web-canary"
+if docker ps -a --format '{{.Names}}' | grep -q "^${CANARY_WEB}$"; then
+    echo "  停止旧 canary..."
+    docker rm -f "$CANARY_WEB" >/dev/null 2>&1
+fi
+
+LEGACY_VOLUME=""
+if [[ -d "$LEGACY_ASSETS_DIR" ]] && [[ -n "$(ls -A "$LEGACY_ASSETS_DIR" 2>/dev/null)" ]]; then
+    LEGACY_VOLUME="-v ${LEGACY_ASSETS_DIR}:/usr/share/nginx/html/assets-legacy/assets:ro"
+fi
+
+docker run -d \
+    --name "$CANARY_WEB" \
+    --network xiaoxia-net-production \
+    -p "127.0.0.1:${CANARY_WEB_PORT}:80" \
+    --restart unless-stopped \
+    --cpus 0.5 \
+    --memory 256m \
+    $LEGACY_VOLUME \
+    --health-cmd "wget --spider -q http://127.0.0.1:80" \
+    --health-interval 30s \
+    --health-timeout 5s \
+    --health-retries 3 \
+    --health-start-period 10s \
+    --log-driver json-file \
+    --log-opt max-size=50m \
+    --log-opt max-file=3 \
+    "${REGISTRY}-web:${VERSION}" >/dev/null
+
+echo "  ✅ Web Canary 已启动（端口 $CANARY_WEB_PORT）"
+
+# 5. 等待健康检查
+echo ""
+echo ">>> 等待 Canary 容器健康..."
+for i in $(seq 1 40); do
+    api_healthy=$(docker inspect --format='{{.State.Health.Status}}' "$CANARY_API" 2>/dev/null || echo "starting")
+    web_healthy=$(docker inspect --format='{{.State.Health.Status}}' "$CANARY_WEB" 2>/dev/null || echo "starting")
+    
+    if [[ "$api_healthy" == "healthy" && "$web_healthy" == "healthy" ]]; then
+        echo "  ✅ API + Web Canary 均健康（用时 ${i}s）"
+        break
+    fi
+    
+    if [[ "$api_healthy" == "unhealthy" ]]; then
+        echo "  ❌ API Canary 健康检查失败"
+        docker logs --tail 30 "$CANARY_API"
+        exit 1
+    fi
+    if [[ "$web_healthy" == "unhealthy" ]]; then
+        echo "  ❌ Web Canary 健康检查失败"
+        docker logs --tail 20 "$CANARY_WEB"
+        exit 1
+    fi
+    
+    sleep 3
+done
+
+# 6. 更新 Nginx 配置 - 添加 upstream 权重
+echo ""
+echo ">>> 更新 Nginx 权重（稳定: $((100-GRAY_PCT))% / 灰度: ${GRAY_PCT}%）..."
+
+# 备份
+BAK_FILE="${NGINX_CONF}.bak.gray.$(date +%Y%m%d%H%M%S)"
+cp "$NGINX_CONF" "$BAK_FILE"
+echo "  已备份: $BAK_FILE"
+
+# 生成 upstream 块
+UPSTREAM_BLOCK="
+# Gray release upstreams（自动生成 - gray_deploy.sh）
+upstream saas_api_backend {
+    server 127.0.0.1:8001 weight=$((100-GRAY_PCT));
+    server 127.0.0.1:${CANARY_API_PORT} weight=${GRAY_PCT};
+}
+
+upstream saas_web_backend {
+    server 127.0.0.1:3002 weight=$((100-GRAY_PCT));
+    server 127.0.0.1:${CANARY_WEB_PORT} weight=${GRAY_PCT};
+}
+"
+
+# 在文件最前面插入 upstream 块
+TMP_CONF=$(mktemp)
+{
+    echo "$UPSTREAM_BLOCK"
+    cat "$NGINX_CONF"
+} > "$TMP_CONF"
+
+# 替换 proxy_pass 指向 upstream
+# API: proxy_pass http://127.0.0.1:8001 -> proxy_pass http://saas_api_backend
+sed -i 's|proxy_pass http://127\.0\.0\.1:8001|proxy_pass http://saas_api_backend|g' "$TMP_CONF"
+# Web: proxy_pass http://127.0.0.1:3002/ -> proxy_pass http://saas_web_backend/
+sed -i 's|proxy_pass http://127\.0\.0\.1:3002/|proxy_pass http://saas_web_backend/|g' "$TMP_CONF"
+
+# 测试配置
+mv "$TMP_CONF" "$NGINX_CONF"
+if ! nginx -t 2>&1; then
+    echo "  ❌ Nginx 配置测试失败，回滚..."
+    cp "$BAK_FILE" "$NGINX_CONF"
+    nginx -t
+    exit 1
+fi
+
+nginx -s reload
+echo "  ✅ Nginx 已 reload，灰度生效"
+
+# 7. 验证灰度流量
+echo ""
+echo ">>> 验证灰度流量..."
+gray_hits=0
+total_hits=20
+for i in $(seq 1 $total_hits); do
+    resp=$(curl -s -o /dev/null -w "%{http_code}" -H "X-Gray-Test: 1" http://127.0.0.1:${CANARY_API_PORT}/health 2>/dev/null || echo "000")
+    if [[ "$resp" == "200" ]]; then
+        gray_hits=$((gray_hits + 1))
+    fi
+    sleep 0.1
+done
+echo "  Canary 健康验证: $gray_hits/$total_hits 请求成功"
+
+echo ""
+echo "============================================"
+echo "  ✅ 灰度发布完成"
+echo "  版本: $VERSION (${GRAY_PCT}%流量)"
+echo "  API: 127.0.0.1:$CANARY_API_PORT"
+echo "  Web: 127.0.0.1:$CANARY_WEB_PORT"
+echo "  Nginx 备份: $BAK_FILE"
+echo "  回滚: ./scripts/rollback.sh"
+echo "  Worker: 暂不灰度（队列消费无法按比例切流）"
+echo "============================================"

scripts/release.sh

@@ -0,0 +1,113 @@
+#!/bin/bash
+# 一键发布脚本：打 tag → 触发 CI 构建 → 可选灰度发布
+# 用法: ./scripts/release.sh v0.1.130 [--gray 5]
+#
+# 说明：
+#   - 打 tag 后 CI 会自动构建镜像并全量部署到生产
+#   - 加 --gray 参数则在构建完成后执行灰度切流（需 SSH 到生产服务器执行）
+#   - 加 --no-deploy 只打 tag 不触发自动部署
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
+
+usage() {
+    echo "用法: $0 <版本号> [--gray 百分比] [--no-deploy]"
+    echo ""
+    echo "示例:"
+    echo "  $0 v0.1.130                  # 打tag + 全量发布（CI自动部署）"
+    echo "  $0 v0.1.130 --gray 5         # 打tag + 5%灰度发布"
+    echo "  $0 v0.1.130 --no-deploy      # 只打tag，不部署"
+    exit 1
+}
+
+VERSION=""
+GRAY_PCT=0
+DEPLOY=true
+
+while [[ $# -gt 0 ]]; do
+    case "$1" in
+        --gray)
+            GRAY_PCT="$2"
+            shift 2
+            ;;
+        --no-deploy)
+            DEPLOY=false
+            shift
+            ;;
+        -h|--help)
+            usage
+            ;;
+        v*)
+            VERSION="$1"
+            shift
+            ;;
+        *)
+            echo "未知参数: $1"
+            usage
+            ;;
+    esac
+done
+
+if [[ -z "$VERSION" ]]; then
+    echo "错误: 请指定版本号（如 v0.1.130）"
+    usage
+fi
+
+echo "============================================"
+echo "  发布版本: $VERSION"
+echo "  灰度比例: ${GRAY_PCT}%"
+echo "  自动部署: $DEPLOY"
+echo "============================================"
+
+cd "$REPO_ROOT"
+
+# 1. 确认分支
+CURRENT_BRANCH=$(git rev-parse --abbrev-ref HEAD)
+if [[ "$CURRENT_BRANCH" != "develop" ]]; then
+    echo "错误: 请在 develop 分支上打 tag"
+    exit 1
+fi
+
+# 2. 拉取最新
+echo ""
+echo ">>> 拉取最新代码..."
+git pull origin develop
+
+# 3. 检查 tag 是否已存在
+if git rev-parse "$VERSION" >/dev/null 2>&1; then
+    echo "警告: tag $VERSION 已存在，跳过打 tag"
+else
+    echo ""
+    echo ">>> 打 tag $VERSION ..."
+    git tag -a "$VERSION" -m "Release $VERSION"
+    git push origin "$VERSION"
+    echo "  ✅ Tag 已推送，CI 将自动构建生产镜像"
+fi
+
+# 4. 部署提示
+if [[ "$DEPLOY" == "true" ]]; then
+    echo ""
+    echo ">>> 构建 & 部署"
+    echo "  CI 会自动执行:"
+    echo "    1. Build Production Runtime Images（约10-15分钟）"
+    echo "    2. Deploy Production（SSH 到生产服务器部署）"
+    echo ""
+    echo "  查看进度: Gitea Actions → 对应 tag 的 run"
+    
+    if [[ "$GRAY_PCT" -gt 0 ]]; then
+        echo ""
+        echo ">>> 灰度发布"
+        echo "  构建部署完成后，在生产服务器上执行:"
+        echo "    cd /var/lib/xiaoxia-saas-production"
+        echo "    ./gray_deploy.sh $VERSION $GRAY_PCT"
+    fi
+fi
+
+echo ""
+echo "============================================"
+echo "  ✅ 发布流程触发完成"
+echo "  版本: $VERSION"
+echo "  灰度: ${GRAY_PCT}%"
+echo "============================================"

scripts/rollback.sh

@@ -0,0 +1,42 @@
+#!/bin/bash
+# 灰度回滚脚本：切回稳定版本流量
+# 用法: ./scripts/rollback.sh [稳定版本号]
+
+set -euo pipefail
+
+STABLE_VERSION="${1:-current}"
+
+echo "=========================================="
+echo "  灰度回滚"
+echo "  切回稳定版本: $STABLE_VERSION"
+echo "=========================================="
+
+# 1. 恢复Nginx全量到稳定版本
+echo ""
+echo ">>> 恢复Nginx全量流量到稳定版本..."
+
+NGINX_CONF="${NGINX_CONF:-/etc/nginx/conf.d/saas-api.conf}"
+if [[ -f "$NGINX_CONF" ]]; then
+    # 找最近的备份
+    LATEST_BAK=$(ls -t "${NGINX_CONF}".bak.* 2>/dev/null | head -1)
+    if [[ -n "$LATEST_BAK" ]]; then
+        cp "$LATEST_BAK" "$NGINX_CONF"
+        echo "  从备份恢复: $LATEST_BAK"
+    else
+        echo "  未找到备份，请手动移除 canary upstream"
+    fi
+    
+    nginx -t && nginx -s reload
+    echo "  ✅ Nginx 已回滚"
+fi
+
+# 2. 停止灰度版本容器（保留30分钟以便排查）
+echo ""
+echo ">>> 灰度版本容器将在30分钟后停止（便于排查问题）"
+echo "    立即停止请执行: docker stop saas-api-canary saas-worker-canary"
+
+echo ""
+echo "=========================================="
+echo "  ✅ 回滚完成"
+echo "  流量已全部切回稳定版本"
+echo "=========================================="

scripts/rollback_gray.sh

@@ -0,0 +1,77 @@
+#!/bin/bash
+# 灰度回滚脚本：切回全量稳定版本，停止 canary 容器
+# 用法: ./scripts/rollback_gray.sh
+
+set -euo pipefail
+
+NGINX_CONF="${NGINX_CONF:-/etc/nginx/sites-enabled/00-xiaoxia-saas}"
+CANARY_API="${CANARY_API:-xiaoxia-api-canary}"
+CANARY_WEB="${CANARY_WEB:-xiaoxia-web-canary}"
+
+echo "============================================"
+echo "  灰度回滚"
+echo "  目标: 全量切回稳定版本"
+echo "============================================"
+
+# 1. 找最近的灰度备份
+echo ""
+echo ">>> 查找最近的灰度备份..."
+LATEST_BAK=$(ls -t "${NGINX_CONF}".bak.gray.* 2>/dev/null | head -1 || true)
+
+if [[ -z "$LATEST_BAK" ]]; then
+    echo "  未找到灰度备份，尝试手动移除 upstream 配置..."
+    
+    # 手动回滚：移除 upstream 块，把 proxy_pass 改回 127.0.0.1
+    TMP_CONF=$(mktemp)
+    
+    # 移除 upstream 块（从 "# Gray release upstreams" 到空行结束）
+    awk '
+        /^# Gray release upstreams/ { skip=1; next }
+        skip && /^$/ && !found_first_empty { found_first_empty=1; next }
+        skip && found_first_empty && /^$/ { skip=0; found_first_empty=0; next }
+        skip { next }
+        { print }
+    ' "$NGINX_CONF" > "$TMP_CONF"
+    
+    # 把 upstream 名改回 IP
+    sed -i 's|proxy_pass http://saas_api_backend|proxy_pass http://127.0.0.1:8001|g' "$TMP_CONF"
+    sed -i 's|proxy_pass http://saas_web_backend/|proxy_pass http://127.0.0.1:3002/|g' "$TMP_CONF"
+    
+    mv "$TMP_CONF" "$NGINX_CONF"
+else
+    echo "  从备份恢复: $LATEST_BAK"
+    cp "$LATEST_BAK" "$NGINX_CONF"
+fi
+
+# 2. 测试并 reload nginx
+echo ""
+echo ">>> Nginx 测试 & reload..."
+if ! nginx -t 2>&1; then
+    echo "  ❌ Nginx 配置测试失败！请检查"
+    exit 1
+fi
+nginx -s reload
+echo "  ✅ Nginx 已回滚，全量切回稳定版本"
+
+# 3. 停止 canary 容器（延迟停止，保留30分钟便于排查）
+echo ""
+echo ">>> Canary 容器将在30分钟后停止（便于排查）"
+echo "    立即停止请执行: docker rm -f $CANARY_API $CANARY_WEB"
+
+# 30分钟后停止（后台执行，不阻塞脚本）
+(
+    sleep 1800
+    for c in "$CANARY_API" "$CANARY_WEB"; do
+        if docker ps --format '{{.Names}}' | grep -q "^${c}$"; then
+            docker stop "$c" >/dev/null 2>&1 && docker rm "$c" >/dev/null 2>&1
+            echo "[$(date)] 已停止 canary 容器: $c"
+        fi
+    done
+) &
+
+echo ""
+echo "============================================"
+echo "  ✅ 灰度回滚完成"
+echo "  流量已全部切回稳定版本"
+echo "  Canary 容器: 30分钟后自动清理"
+echo "============================================"

tests/unit/test_speed_engine.py

@@ -0,0 +1,269 @@
+"""视频调速引擎单元测试."""
+
+import pytest
+from video_processing.speed_engine import (
+    MAX_SPEED,
+    MIN_SPEED,
+    SpeedConfig,
+    SpeedEngine,
+)
+
+# ─── SpeedConfig 解析与校验 ──────────────────────────────────
+
+
+class TestSpeedConfig:
+    def test_default_values(self):
+        config = SpeedConfig()
+        assert config.speed == 1.0
+        assert config.pitch_correct is True
+
+    def test_parse_none(self):
+        config = SpeedConfig.parse(None)
+        assert config.speed == 1.0
+        assert config.pitch_correct is True
+
+    def test_parse_empty_dict(self):
+        config = SpeedConfig.parse({})
+        assert config.speed == 1.0
+
+    def test_parse_valid_speed(self):
+        config = SpeedConfig.parse({"speed": 2.0})
+        assert config.speed == 2.0
+
+    def test_parse_pitch_correct_false(self):
+        config = SpeedConfig.parse({"pitch_correct": False})
+        assert config.pitch_correct is False
+
+    def test_parse_invalid_speed_type(self):
+        config = SpeedConfig.parse({"speed": "fast"})
+        assert config.speed == 1.0
+
+    def test_parse_invalid_pitch_type(self):
+        config = SpeedConfig.parse({"pitch_correct": "yes"})
+        assert config.pitch_correct is True
+
+    def test_clamp_below_min(self):
+        config = SpeedConfig(speed=0.1)
+        config.clamp()
+        assert config.speed == MIN_SPEED
+
+    def test_clamp_zero(self):
+        config = SpeedConfig(speed=0)
+        config.clamp()
+        assert config.speed == 1.0
+
+    def test_clamp_negative(self):
+        config = SpeedConfig(speed=-1.0)
+        config.clamp()
+        assert config.speed == 1.0
+
+    def test_clamp_above_max(self):
+        config = SpeedConfig(speed=10.0)
+        config.clamp()
+        assert config.speed == MAX_SPEED
+
+    def test_clamp_within_range(self):
+        config = SpeedConfig(speed=1.5)
+        config.clamp()
+        assert config.speed == 1.5
+
+    def test_is_original_true(self):
+        config = SpeedConfig(speed=1.0)
+        assert config.is_original is True
+
+    def test_is_original_false(self):
+        config = SpeedConfig(speed=1.5)
+        assert config.is_original is False
+
+    def test_parse_clamps_automatically(self):
+        """parse 方法应该自动调用 clamp."""
+        config = SpeedConfig.parse({"speed": 100.0})
+        assert config.speed == MAX_SPEED
+
+
+# ─── SpeedEngine 视频滤镜 ────────────────────────────────────
+
+
+class TestSpeedEngineVideoFilter:
+    def setup_method(self):
+        self.engine = SpeedEngine()
+
+    def test_original_speed_returns_empty(self):
+        config = SpeedConfig(speed=1.0)
+        assert self.engine.build_video_filter(config) == ""
+
+    def test_double_speed(self):
+        config = SpeedConfig(speed=2.0)
+        result = self.engine.build_video_filter(config)
+        assert "setpts=PTS/2.0" in result
+
+    def test_half_speed(self):
+        config = SpeedConfig(speed=0.5)
+        result = self.engine.build_video_filter(config)
+        assert "setpts=PTS/0.5" in result
+
+    def test_quarter_speed(self):
+        config = SpeedConfig(speed=0.25)
+        result = self.engine.build_video_filter(config)
+        assert "setpts=PTS/0.25" in result
+
+    def test_quad_speed(self):
+        config = SpeedConfig(speed=4.0)
+        result = self.engine.build_video_filter(config)
+        assert "setpts=PTS/4.0" in result
+
+
+# ─── SpeedEngine 音频滤镜（atempo 多级串联） ─────────────────
+
+
+class TestSpeedEngineAudioFilter:
+    def setup_method(self):
+        self.engine = SpeedEngine()
+
+    def test_original_speed_returns_empty(self):
+        config = SpeedConfig(speed=1.0)
+        assert self.engine.build_audio_filter(config) == ""
+
+    def test_double_speed_single_stage(self):
+        """2x 在 atempo 单级范围内，只需一个 atempo."""
+        config = SpeedConfig(speed=2.0)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=2.0000"
+
+    def test_half_speed_single_stage(self):
+        config = SpeedConfig(speed=0.5)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=0.5000"
+
+    def test_quad_speed_two_stages(self):
+        """4x 需要两级 atempo: 2.0 * 2.0."""
+        config = SpeedConfig(speed=4.0)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=2.0000,atempo=2.0000"
+
+    def test_quarter_speed_two_stages(self):
+        """0.25x 需要两级 atempo: 0.5 * 0.5."""
+        config = SpeedConfig(speed=0.25)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=0.5000,atempo=0.5000"
+
+    def test_triple_speed_two_stages(self):
+        """3x: 2.0 * 1.5."""
+        config = SpeedConfig(speed=3.0)
+        result = self.engine.build_audio_filter(config)
+        parts = result.split(",")
+        assert len(parts) == 2
+        assert "atempo=2.0000" in parts
+        assert "atempo=1.5000" in parts
+
+    def test_03_speed_two_stages(self):
+        """0.3x: 0.5 * 0.6."""
+        config = SpeedConfig(speed=0.3)
+        result = self.engine.build_audio_filter(config)
+        parts = result.split(",")
+        assert len(parts) == 2
+        assert "atempo=0.5000" in parts
+        assert "atempo=0.6000" in parts
+
+    def test_split_atempo_inside_range(self):
+        """0.5~2.0 范围内只返回一级."""
+        stages = SpeedEngine._split_atempo_stages(1.5)
+        assert len(stages) == 1
+        assert stages[0] == 1.5
+
+    def test_split_atempo_boundary_min(self):
+        stages = SpeedEngine._split_atempo_stages(0.5)
+        assert len(stages) == 1
+        assert stages[0] == 0.5
+
+    def test_split_atempo_boundary_max(self):
+        stages = SpeedEngine._split_atempo_stages(2.0)
+        assert len(stages) == 1
+        assert stages[0] == 2.0
+
+    def test_split_atempo_product_equals_speed(self):
+        """所有级联的乘积应该等于原速度."""
+        test_cases = [0.25, 0.3, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0, 4.0]
+        for speed in test_cases:
+            stages = SpeedEngine._split_atempo_stages(speed)
+            product = 1.0
+            for s in stages:
+                product *= s
+            assert abs(product - speed) < 1e-6, f"speed={speed}, stages={stages}, product={product}"
+
+    def test_split_atempo_all_in_range(self):
+        """所有级都应该在 0.5~2.0 范围内."""
+        test_cases = [0.25, 0.3, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0, 4.0]
+        for speed in test_cases:
+            stages = SpeedEngine._split_atempo_stages(speed)
+            for s in stages:
+                assert 0.5 <= s <= 2.0, f"speed={speed}, stage={s} out of range"
+
+
+# ─── SpeedEngine 时长计算 ────────────────────────────────────
+
+
+class TestSpeedEngineDuration:
+    def setup_method(self):
+        self.engine = SpeedEngine()
+
+    def test_original_speed_same_duration(self):
+        config = SpeedConfig(speed=1.0)
+        assert self.engine.adjust_duration(10.0, config) == 10.0
+
+    def test_double_speed_half_duration(self):
+        config = SpeedConfig(speed=2.0)
+        assert self.engine.adjust_duration(10.0, config) == 5.0
+
+    def test_half_speed_double_duration(self):
+        config = SpeedConfig(speed=0.5)
+        assert self.engine.adjust_duration(10.0, config) == 20.0
+
+    def test_quad_speed_quarter_duration(self):
+        config = SpeedConfig(speed=4.0)
+        assert self.engine.adjust_duration(10.0, config) == 2.5
+
+    def test_zero_duration(self):
+        config = SpeedConfig(speed=2.0)
+        assert self.engine.adjust_duration(0.0, config) == 0.0
+
+    def test_negative_duration(self):
+        config = SpeedConfig(speed=2.0)
+        assert self.engine.adjust_duration(-1.0, config) == -1.0
+
+
+# ─── SpeedEngine 便捷方法 ────────────────────────────────────
+
+
+class TestSpeedEngineHelper:
+    def setup_method(self):
+        self.engine = SpeedEngine()
+
+    def test_build_clip_speed_filter_original(self):
+        v_f, a_f, cfg = self.engine.build_clip_speed_filter(1.0)
+        assert v_f == ""
+        assert a_f == ""
+        assert cfg.speed == 1.0
+
+    def test_build_clip_speed_filter_2x(self):
+        v_f, a_f, cfg = self.engine.build_clip_speed_filter(2.0)
+        assert "setpts=PTS/2.0" in v_f
+        assert "atempo=2.0" in a_f
+        assert cfg.speed == 2.0
+
+    def test_build_clip_speed_clamped(self):
+        _, _, cfg = self.engine.build_clip_speed_filter(100.0)
+        assert cfg.speed == MAX_SPEED
+
+    def test_resolve_clip_speed_default(self):
+        assert SpeedEngine.resolve_clip_speed({}) == 1.0
+        assert SpeedEngine.resolve_clip_speed(None) == 1.0
+
+    def test_resolve_clip_speed_zero_uses_global(self):
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": 0}, 1.5) == 1.5
+
+    def test_resolve_clip_speed_custom(self):
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": 2.0}) == 2.0
+
+    def test_resolve_clip_speed_invalid_type(self):
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": "fast"}) == 1.0