diff --git a/apps/worker/services/tts_service_factory.py b/apps/worker/services/tts_service_factory.py
index d25aad513..a69f170ec 100755
--- a/apps/worker/services/tts_service_factory.py
+++ b/apps/worker/services/tts_service_factory.py
@@ -35,7 +35,20 @@ def get_tts_service(provider: str | None = None, **kwargs) -> TtsService:
         ValueError: 不支持的供应商
     """
     if provider is None:
-        provider = os.environ.get("TTS_PROVIDER", "mock")
+        provider = os.environ.get("TTS_PROVIDER", "")
+
+    if not provider:
+        # 自动检测：配置了 CosyVoice API Key 则默认用 cosyvoice，否则用 mock
+        try:
+            from packages.shared.config import get_shared_settings
+
+            settings = get_shared_settings()
+            if getattr(settings, "cosyvoice_api_key", ""):
+                provider = "cosyvoice"
+            else:
+                provider = "mock"
+        except Exception:
+            provider = "mock"
 
     provider = provider.lower()
 
@@ -45,6 +58,13 @@ def get_tts_service(provider: str | None = None, **kwargs) -> TtsService:
             from packages.adapters.tts.mock_tts_service import MockTtsService
 
             _PROVIDERS["mock"] = MockTtsService
+        elif provider in ("cosyvoice", "aliyun", "dashscope"):
+            from packages.adapters.tts.cosyvoice_tts_service import CosyVoiceTtsService
+
+            _PROVIDERS["cosyvoice"] = CosyVoiceTtsService
+            _PROVIDERS["aliyun"] = CosyVoiceTtsService
+            _PROVIDERS["dashscope"] = CosyVoiceTtsService
+            provider = "cosyvoice"
         else:
             logger.warning("未知 TTS provider: %s，回退到 mock", provider)
             from packages.adapters.tts.mock_tts_service import MockTtsService
diff --git a/apps/worker/worker_app/tasks/edit_plan_generation.py b/apps/worker/worker_app/tasks/edit_plan_generation.py
old mode 100644
new mode 100755
index 0826dd723..cd1db6ace
--- a/apps/worker/worker_app/tasks/edit_plan_generation.py
+++ b/apps/worker/worker_app/tasks/edit_plan_generation.py
@@ -469,6 +469,87 @@ def _render_with_legacy(
         except Exception as sub_err:
             logger.warning("legacy 标题/字幕叠加失败（不影响主流程）: plan_id=%s err=%s", plan_id, sub_err)
 
+    # ── TTS 配音混音（legacy 引擎补齐） ────────────────────────────────
+    tts_cfg = plan_config.get("tts", {}) or {}
+    tts_enabled = tts_cfg.get("enabled", False) and bool(tts_cfg.get("text", "").strip())
+
+    if tts_enabled and output_path.exists() and duration > 0:
+        try:
+            from packages.domain.tts_config import TtsConfig
+
+            tts_config = TtsConfig.parse(tts_cfg)
+            if tts_config.enabled and tts_config.text.strip():
+                from apps.worker.services.tts_service_factory import get_tts_service
+
+                tts_service = get_tts_service()
+                voiceover_path = tmpdir_path / f"voiceover_{plan_id}.wav"
+
+                # 生成配音音频
+                audio_path = tts_service.synthesize(
+                    text=tts_config.text,
+                    voice_id=tts_config.voice_id,
+                    speed=tts_config.speed,
+                    pitch=tts_config.pitch,
+                    output_path=voiceover_path,
+                )
+
+                if audio_path and audio_path.exists() and audio_path.stat().st_size > 0:
+                    from video_processing.ffmpeg_utils import run_ffmpeg
+
+                    mixed_path = tmpdir_path / f"{plan_id}_with_voiceover.mp4"
+
+                    # 混音：配音音量按配置调整
+                    voice_volume = max(0.0, min(1.0, tts_config.volume))
+
+                    if tts_config.overlap_mode == "mix":
+                        # 混音模式：原音 + 配音混合
+                        filter_complex = (
+                            f"[0:a]volume=1.0[a0];"
+                            f"[1:a]volume={voice_volume:.2f}[a1];"
+                            f"[a0][a1]amix=inputs=2:duration=first:dropout_transition=0[aout]"
+                        )
+                    else:
+                        # replace 模式：配音替换原音
+                        filter_complex = f"[1:a]volume={voice_volume:.2f}[aout]"
+
+                    run_ffmpeg(
+                        [
+                            "ffmpeg",
+                            "-y",
+                            "-i",
+                            str(output_path),
+                            "-i",
+                            str(audio_path),
+                            "-filter_complex",
+                            filter_complex,
+                            "-map",
+                            "0:v",
+                            "-map",
+                            "[aout]",
+                            "-c:v",
+                            "copy",
+                            "-c:a",
+                            "aac",
+                            "-b:a",
+                            "128k",
+                            "-shortest",
+                            str(mixed_path),
+                        ],
+                        timeout=1800,
+                    )
+
+                    if mixed_path.exists() and mixed_path.stat().st_size > 0:
+                        output_path = mixed_path
+                        file_size = mixed_path.stat().st_size
+                        logger.info(
+                            "legacy TTS 配音混音完成: plan_id=%s voice_id=%s mode=%s",
+                            plan_id,
+                            tts_config.voice_id,
+                            tts_config.overlap_mode,
+                        )
+        except Exception as tts_err:
+            logger.warning("legacy TTS 配音混音失败（不影响主流程）: plan_id=%s err=%s", plan_id, tts_err)
+
     # 上传到 OSS
     storage_key = f"rendered/{plan_id}/output.mp4"
     output_url = upload_to_oss(output_path, storage_key)
diff --git a/packages/adapters/tts/cosyvoice_tts_service.py b/packages/adapters/tts/cosyvoice_tts_service.py
new file mode 100755
index 000000000..8abd0d84d
--- /dev/null
+++ b/packages/adapters/tts/cosyvoice_tts_service.py
@@ -0,0 +1,249 @@
+"""CosyVoice TTS 服务适配器.
+
+将 CosyVoiceService 包装为 TtsService 接口，供统一渲染管道的 TtsEngine 使用。
+支持阿里云百炼 CosyVoice 真实音色合成。
+"""
+
+from __future__ import annotations
+
+import logging
+import tempfile
+from pathlib import Path
+from urllib.parse import urlparse
+
+from packages.ports.tts_service import TtsError, TtsService
+
+logger = logging.getLogger(__name__)
+
+
+class CosyVoiceTtsService(TtsService):
+    """CosyVoice TTS 服务适配器.
+
+    包装 CosyVoiceService，实现 TtsService 接口。
+    合成流程：调用 CosyVoice API → 获取音频 URL → 下载到本地 → （可选）转码为目标格式
+    """
+
+    def __init__(
+        self,
+        api_key: str = "",
+        base_url: str = "",
+        model: str = "",
+        sample_rate: int = 24000,
+        format: str = "mp3",
+        ffmpeg_bin: str = "ffmpeg",
+    ) -> None:
+        """初始化 CosyVoice TTS 服务.
+
+        Args:
+            api_key: DashScope API Key，为空时从配置读取
+            base_url: DashScope API Base URL
+            model: 语音合成模型
+            sample_rate: 默认采样率
+            format: 默认输出格式
+            ffmpeg_bin: ffmpeg 可执行文件路径（用于转码）
+        """
+        # 延迟导入避免循环依赖
+        from packages.application.cosyvoice_service import CosyVoiceService
+
+        self._service = CosyVoiceService(
+            api_key=api_key,
+            base_url=base_url,
+            model=model,
+        )
+        self._default_sample_rate = sample_rate
+        self._default_format = format
+        self._ffmpeg_bin = ffmpeg_bin
+
+    @property
+    def provider_name(self) -> str:
+        return "cosyvoice"
+
+    def synthesize(
+        self,
+        text: str,
+        *,
+        voice_id: str = "",
+        speed: float = 1.0,
+        pitch: float = 0.0,
+        output_path: Path | None = None,
+        sample_rate: int = 22050,
+        format: str = "wav",
+    ) -> Path:
+        """调用 CosyVoice 合成语音并下载到本地.
+
+        Args:
+            text: 输入文本
+            voice_id: 音色 ID（CosyVoice 音色名，如 longxiaochun_v3）
+            speed: 语速 (0.5 ~ 2.0)
+            pitch: 语调（半音，-12 ~ 12）— CosyVoice 原生不支持，用 ffmpeg 后处理实现
+            output_path: 输出文件路径（None 则自动生成）
+            sample_rate: 采样率
+            format: 输出格式 (wav/mp3)
+
+        Returns:
+            输出音频文件路径
+        """
+        if not text.strip():
+            raise TtsError("文本不能为空")
+
+        if not voice_id:
+            voice_id = "longxiaochun_v3"  # 默认音色
+
+        # 语速边界
+        speed = max(0.5, min(2.0, speed))
+
+        # 输出路径
+        if output_path is None:
+            suffix = f".{format}"
+            tmp = tempfile.NamedTemporaryFile(suffix=suffix, delete=False)
+            tmp.close()
+            output_path = Path(tmp.name)
+
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+
+        try:
+            # 1. 调用 CosyVoice 合成（默认 mp3 格式，兼容性最好）
+            result = self._service.synthesize_speech(
+                text=text,
+                voice_id=voice_id,
+                sample_rate=sample_rate or self._default_sample_rate,
+                format="mp3",  # 先下mp3，后面按需转码
+                speed=speed,
+            )
+
+            if not result.audio_url:
+                raise TtsError("CosyVoice 未返回音频 URL")
+
+            # 2. 下载音频文件
+            downloaded = self._download_audio(result.audio_url, output_path.parent / "_cosyvoice_tmp.mp3")
+
+            if not downloaded.exists() or downloaded.stat().st_size == 0:
+                raise TtsError("音频下载失败或文件为空")
+
+            # 3. 如需转码（wav）或 pitch 调整，用 ffmpeg 处理
+            need_transcode = (format != "mp3") or abs(pitch) > 0.01
+
+            if need_transcode:
+                self._post_process(downloaded, output_path, format=format, pitch=pitch, sample_rate=sample_rate)
+            else:
+                # 直接移动文件
+                downloaded.rename(output_path)
+
+            if not output_path.exists() or output_path.stat().st_size == 0:
+                raise TtsError("输出文件为空或不存在")
+
+            return output_path
+
+        except TtsError:
+            raise
+        except Exception as e:
+            logger.error("CosyVoice TTS 合成失败: %s", e)
+            raise TtsError(f"CosyVoice TTS 合成失败: {e}") from e
+
+    def estimate_duration(self, text: str, *, speed: float = 1.0) -> float:
+        """估算音频时长（秒）.
+
+        CosyVoice 不返回预估时长，按中文语速经验值估算：
+        - 正常语速约 4 字/秒
+        """
+        if not text:
+            return 0.0
+        char_count = len([c for c in text if not c.isspace()])
+        if char_count == 0:
+            return 0.0
+        base_duration = char_count / 4.0  # 4 字/秒
+        return base_duration / max(0.1, speed)
+
+    def available_voices(self) -> list[str]:
+        """支持的音色列表."""
+        from packages.domain.preset_voices import get_preset_voices
+
+        return [v.voice_id for v in get_preset_voices()]
+
+    def _download_audio(self, url: str, save_path: Path) -> Path:
+        """下载音频文件.
+
+        Args:
+            url: 音频 URL
+            save_path: 保存路径
+
+        Returns:
+            保存路径
+        """
+        import httpx
+
+        parsed = urlparse(url)
+        if parsed.scheme not in ("http", "https"):
+            raise TtsError(f"不支持的音频 URL scheme: {parsed.scheme}")
+
+        save_path.parent.mkdir(parents=True, exist_ok=True)
+
+        with httpx.Client(timeout=120.0) as client:
+            with client.stream("GET", url) as response:
+                response.raise_for_status()
+                with open(save_path, "wb") as f:
+                    for chunk in response.iter_bytes():
+                        f.write(chunk)
+
+        return save_path
+
+    def _post_process(
+        self,
+        input_path: Path,
+        output_path: Path,
+        *,
+        format: str = "wav",
+        pitch: float = 0.0,
+        sample_rate: int = 22050,
+    ) -> None:
+        """后处理：转码 + pitch 调整.
+
+        Args:
+            input_path: 输入文件路径
+            output_path: 输出文件路径
+            format: 输出格式
+            pitch: 语调偏移（半音）
+            sample_rate: 输出采样率
+        """
+        import subprocess
+
+        # 构建滤镜
+        filter_parts = []
+
+        # pitch 调整：通过 asetrate 实现
+        if abs(pitch) > 0.01:
+            pitch_factor = 2 ** (pitch / 12)
+            new_rate = int(sample_rate * pitch_factor)
+            filter_parts.append(f"asetrate={new_rate}")
+            filter_parts.append(f"aresample={sample_rate}")
+
+        filter_str = ",".join(filter_parts) if filter_parts else None
+
+        # 编码参数
+        if format == "mp3":
+            codec_args = ["-acodec", "libmp3lame", "-b:a", "128k"]
+        else:  # wav
+            codec_args = ["-acodec", "pcm_s16le"]
+
+        command = [
+            self._ffmpeg_bin,
+            "-y",
+            "-i",
+            str(input_path),
+        ]
+
+        if filter_str:
+            command.extend(["-af", filter_str])
+
+        command.extend(codec_args)
+        command.extend(["-ar", str(sample_rate), "-ac", "1", str(output_path)])
+
+        result = subprocess.run(
+            command,
+            capture_output=True,
+            text=True,
+            timeout=60,
+        )
+
+        if result.returncode != 0:
+            raise TtsError(f"音频后处理失败: {result.stderr[-500:]}")
diff --git a/tests/unit/test_cosyvoice_tts_adapter.py b/tests/unit/test_cosyvoice_tts_adapter.py
new file mode 100755
index 000000000..b41e97734
--- /dev/null
+++ b/tests/unit/test_cosyvoice_tts_adapter.py
@@ -0,0 +1,174 @@
+"""CosyVoice TTS 适配器单元测试."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from packages.domain.tts_config import TtsConfig
+
+
+class TestTtsConfig:
+    """TTS 配置解析测试."""
+
+    def test_parse_full_config(self):
+        """完整配置解析."""
+        cfg = TtsConfig.parse(
+            {
+                "enabled": True,
+                "voice_id": "longxiaochun_v3",
+                "text": "你好世界",
+                "speed": 1.2,
+                "pitch": 2.0,
+                "volume": 0.7,
+                "align_mode": "full",
+                "overlap_mode": "mix",
+            }
+        )
+        assert cfg.enabled is True
+        assert cfg.voice_id == "longxiaochun_v3"
+        assert cfg.text == "你好世界"
+        assert cfg.speed == 1.2
+        assert cfg.pitch == 2.0
+        assert cfg.volume == 0.7
+        assert cfg.align_mode == "full"
+        assert cfg.overlap_mode == "mix"
+
+    def test_parse_disabled(self):
+        """禁用状态."""
+        cfg = TtsConfig.parse({"enabled": False})
+        assert cfg.enabled is False
+
+    def test_parse_none(self):
+        """空配置."""
+        cfg = TtsConfig.parse(None)
+        assert cfg.enabled is False
+
+    def test_parse_empty_text_still_enabled(self):
+        """有 enabled 但无 text，配置仍然有效（调用方判断是否有文本）."""
+        cfg = TtsConfig.parse({"enabled": True, "voice_id": "test"})
+        assert cfg.enabled is True
+        assert cfg.text == ""
+
+    def test_speed_clamp(self):
+        """语速边界钳制."""
+        cfg = TtsConfig.parse({"enabled": True, "speed": 3.0})
+        assert cfg.speed == 2.0
+
+        cfg2 = TtsConfig.parse({"enabled": True, "speed": 0.1})
+        assert cfg2.speed == 0.5
+
+    def test_volume_clamp(self):
+        """音量边界钳制."""
+        cfg = TtsConfig.parse({"enabled": True, "volume": 2.0})
+        assert cfg.volume == 1.0
+
+    def test_invalid_align_mode(self):
+        """无效对齐模式回退到默认."""
+        cfg = TtsConfig.parse({"enabled": True, "align_mode": "invalid"})
+        assert cfg.align_mode == "full"
+
+
+class TestCosyVoiceTtsAdapter:
+    """CosyVoice TTS 适配器测试."""
+
+    def test_import_ok(self):
+        """适配器能正常导入."""
+        from packages.adapters.tts.cosyvoice_tts_service import CosyVoiceTtsService
+
+        assert CosyVoiceTtsService is not None
+
+    def test_provider_name(self):
+        """provider_name 属性."""
+        from packages.adapters.tts.cosyvoice_tts_service import CosyVoiceTtsService
+
+        # 用 mock 替换底层 CosyVoiceService
+        with patch("packages.application.cosyvoice_service.CosyVoiceService"):
+            svc = CosyVoiceTtsService()
+            assert svc.provider_name == "cosyvoice"
+
+    def test_available_voices(self):
+        """可用音色列表来自预设音色."""
+        from packages.adapters.tts.cosyvoice_tts_service import CosyVoiceTtsService
+
+        with patch("packages.application.cosyvoice_service.CosyVoiceService"):
+            svc = CosyVoiceTtsService()
+            voices = svc.available_voices()
+            assert len(voices) > 0
+            assert "longxiaochun_v3" in voices
+            assert "longxiaoxia_v3" in voices
+
+    def test_estimate_duration(self):
+        """时长估算."""
+        from packages.adapters.tts.cosyvoice_tts_service import CosyVoiceTtsService
+
+        with patch("packages.application.cosyvoice_service.CosyVoiceService"):
+            svc = CosyVoiceTtsService()
+            d = svc.estimate_duration("你好世界", speed=1.0)
+            assert d > 0
+            # 4 个字，4 字/秒 = 1 秒
+            assert abs(d - 1.0) < 0.1
+
+    def test_estimate_duration_speed(self):
+        """语速影响时长估算."""
+        from packages.adapters.tts.cosyvoice_tts_service import CosyVoiceTtsService
+
+        with patch("packages.application.cosyvoice_service.CosyVoiceService"):
+            svc = CosyVoiceTtsService()
+            d_normal = svc.estimate_duration("你好世界", speed=1.0)
+            d_fast = svc.estimate_duration("你好世界", speed=2.0)
+            assert d_fast < d_normal
+            assert abs(d_fast - d_normal / 2.0) < 0.01
+
+    def test_synthesize_empty_text_raises(self):
+        """空文本抛出异常."""
+        from packages.adapters.tts.cosyvoice_tts_service import CosyVoiceTtsService
+        from packages.ports.tts_service import TtsError
+
+        with patch("packages.application.cosyvoice_service.CosyVoiceService"):
+            svc = CosyVoiceTtsService()
+            with pytest.raises(TtsError, match="文本不能为空"):
+                svc.synthesize("  ")
+
+
+class TestTtsServiceFactory:
+    """TTS 服务工厂测试."""
+
+    def test_mock_provider(self):
+        """mock provider 正常."""
+        from apps.worker.services.tts_service_factory import get_tts_service
+
+        svc = get_tts_service("mock")
+        assert svc.provider_name == "mock"
+
+    def test_cosyvoice_provider(self):
+        """cosyvoice provider 注册正常."""
+        from apps.worker.services.tts_service_factory import get_tts_service
+
+        with patch("packages.application.cosyvoice_service.CosyVoiceService"):
+            svc = get_tts_service("cosyvoice")
+            assert svc.provider_name == "cosyvoice"
+
+    def test_aliyun_alias(self):
+        """aliyun 别名映射到 cosyvoice."""
+        from apps.worker.services.tts_service_factory import get_tts_service
+
+        with patch("packages.application.cosyvoice_service.CosyVoiceService"):
+            svc = get_tts_service("aliyun")
+            assert svc.provider_name == "cosyvoice"
+
+    def test_dashscope_alias(self):
+        """dashscope 别名映射到 cosyvoice."""
+        from apps.worker.services.tts_service_factory import get_tts_service
+
+        with patch("packages.application.cosyvoice_service.CosyVoiceService"):
+            svc = get_tts_service("dashscope")
+            assert svc.provider_name == "cosyvoice"
+
+    def test_unknown_fallback_to_mock(self):
+        """未知 provider 回退到 mock."""
+        from apps.worker.services.tts_service_factory import get_tts_service
+
+        svc = get_tts_service("unknown_provider")
+        assert svc.provider_name == "mock"