feat: P1 长文本分段合成

- 新增 text_splitter.py: 长文本智能分段（句子边界 + 短段合并）
- 新增 audio_merger.py: FFmpeg concat 音频合并器
- workflow.py: 分段合成完整流程（同步合并 / 异步轮询 / 失败处理）
- tts_synthesis.py: 新增 process_tts_segment_synthesis Celery 任务
- tts.py: 路由层自动识别分段任务并分发到对应 Celery task
- 23 个单元测试全部通过，P0 回归测试无退化

apps/api/app/api/routes/tts.py

@@ -130,18 +130,25 @@ def synthesize(
 
     # 若任务处于 processing 状态（异步模式），触发 Celery 后台轮询
     if job.status.value == "processing":
-        task_id = (job.metadata or {}).get("cosyvoice_task_id", "")
-        if task_id:
-            try:
+        # 分段合成任务 vs 普通单段任务
+        segment_task_ids = (job.metadata or {}).get("segment_task_ids", [])
+        is_segment = len(segment_task_ids) > 0
+
+        try:
+            if is_segment:
+                from worker_app.tasks import process_tts_segment_synthesis
+
+                process_tts_segment_synthesis.delay(job.id)
+            else:
                 from worker_app.tasks import process_tts_synthesis
 
                 process_tts_synthesis.delay(job.id)
-            except Exception as e:
-                # Celery 调度失败，标记 job 为 failed
-                try:
-                    workflow.process_synthesis_failure(job.id, f"Celery 任务调度失败: {e}")
-                except Exception as inner_e:
-                    logger.error(f"Celery 调度后标记失败时出错: job_id={job.id}, error={inner_e}")
+        except Exception as e:
+            # Celery 调度失败，标记 job 为 failed
+            try:
+                workflow.process_synthesis_failure(job.id, f"Celery 任务调度失败: {e}")
+            except Exception as inner_e:
+                logger.error(f"Celery 调度后标记失败时出错: job_id={job.id}, error={inner_e}")
 
     return TTSSynthesizeResponse(
         job_id=job.id,

apps/worker/worker_app/tasks/__init__.py

@@ -41,6 +41,10 @@ def __getattr__(name: str):
         from .tts_synthesis import process_tts_synthesis
 
         return process_tts_synthesis
+    elif name == "process_tts_segment_synthesis":
+        from .tts_synthesis import process_tts_segment_synthesis
+
+        return process_tts_segment_synthesis
     elif name == "run_ai_recommend":
         from .ai_tasks import run_ai_recommend
 
@@ -62,6 +66,7 @@ __all__ = [
     "extract_background_task",
     "process_voice_clone",
     "process_tts_synthesis",
+    "process_tts_segment_synthesis",
     "run_ai_recommend",
     "run_generate_cover",
 ]

apps/worker/worker_app/tasks/tts_synthesis.py

@@ -104,3 +104,77 @@ def process_tts_synthesis(self: Task, job_id: str) -> dict:
     finally:
         if session is not None:
             session.close()
+
+
+@celery_app.task(bind=True, max_retries=2, name="worker.process_tts_segment_synthesis")
+def process_tts_segment_synthesis(self: Task, job_id: str) -> dict:
+    """分段合成轮询任务 — 轮询多个 CosyVoice 子任务并合并音频。
+
+    与 process_tts_synthesis 类似，但超时更长（300s），
+    因为分段任务需要等待所有子任务完成。
+    """
+    session = None
+    try:
+        session = SessionLocal()
+        repo = SQLAlchemyTTSJobRepository(session)
+        workflow = TTSWorkflowService(
+            repository=repo,
+            cosyvoice_service=CosyVoiceService(),
+        )
+
+        updated_job = workflow.poll_and_process_synthesis(job_id, timeout=300)
+        session.commit()
+
+        logger.info(f"TTS segment synthesis completed: job_id={job_id}, " f"audio_url={updated_job.output_audio_url}")
+        return {
+            "ok": True,
+            "job_id": job_id,
+            "audio_url": updated_job.output_audio_url,
+        }
+
+    except Retry:
+        raise
+
+    except CosyVoiceTimeoutError as e:
+        logger.warning(f"TTS segment synthesis timeout for {job_id}: {e}")
+        if session is not None:
+            session.rollback()
+        raise self.retry(exc=e, countdown=60)
+
+    except CosyVoiceError as e:
+        logger.error(f"TTS segment synthesis failed for {job_id}: {e}")
+        if session is not None:
+            session.rollback()
+        try:
+            if session is not None:
+                job = repo.get(job_id)
+                if job is not None:
+                    job.mark_failed(str(e))
+                    repo.update(job)
+                    session.commit()
+        except Exception as inner_e:
+            logger.error(f"Failed to mark job as failed: {inner_e}")
+            if session is not None:
+                session.rollback()
+        return {"ok": False, "job_id": job_id, "error": str(e)}
+
+    except Exception as e:
+        logger.error(f"TTS segment synthesis unexpected error for {job_id}: {e}")
+        if session is not None:
+            session.rollback()
+        try:
+            if session is not None:
+                job = repo.get(job_id)
+                if job is not None:
+                    job.mark_failed(str(e))
+                    repo.update(job)
+                    session.commit()
+        except Exception as inner_e:
+            logger.error(f"Failed to mark job as failed: {inner_e}")
+            if session is not None:
+                session.rollback()
+        return {"ok": False, "job_id": job_id, "error": str(e)}
+
+    finally:
+        if session is not None:
+            session.close()

packages/application/tts_job/audio_merger.py

@@ -0,0 +1,96 @@
+"""FFmpeg 音频合并器 — P1 长文本分段合成。
+
+将多个分段音频文件合并为一个完整音频文件。
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import shutil
+import subprocess
+import tempfile
+
+logger = logging.getLogger(__name__)
+
+
+class AudioMergeError(Exception):
+    """音频合并异常。"""
+
+    pass
+
+
+class AudioMerger:
+    """使用 FFmpeg 合并多个音频文件。"""
+
+    def merge(self, audio_paths: list[str], output_format: str = "mp3") -> bytes:
+        """合并多个音频文件，返回合并后的音频数据。
+
+        使用 FFmpeg concat demuxer 按顺序拼接音频。
+        所有输入文件必须为相同格式和采样率。
+
+        Args:
+            audio_paths: 音频文件路径列表（按合成顺序）
+            output_format: 输出格式（mp3/wav/pcm）
+
+        Returns:
+            合并后的音频文件字节数据
+
+        Raises:
+            AudioMergeError: 合并失败
+        """
+        if not audio_paths:
+            raise AudioMergeError("没有可合并的音频文件")
+
+        if len(audio_paths) == 1:
+            with open(audio_paths[0], "rb") as f:
+                return f.read()
+
+        temp_dir = tempfile.mkdtemp(prefix="tts_merge_")
+        try:
+            # 生成 concat demuxer 列表文件
+            list_path = os.path.join(temp_dir, "concat_list.txt")
+            with open(list_path, "w") as f:
+                for path in audio_paths:
+                    # FFmpeg concat 文件需要 file: 前缀，路径中的 ' 和 \n 需转义
+                    escaped = path.replace("'", "'\\''").replace("\n", "\\n")
+                    f.write(f"file '{escaped}'\n")
+
+            output_path = os.path.join(temp_dir, f"merged.{output_format}")
+
+            cmd = [
+                "ffmpeg",
+                "-y",
+                "-f",
+                "concat",
+                "-safe",
+                "0",
+                "-i",
+                list_path,
+                "-c",
+                "copy",
+                output_path,
+            ]
+
+            result = subprocess.run(
+                cmd,
+                capture_output=True,
+                text=True,
+                timeout=120,
+            )
+
+            if result.returncode != 0:
+                logger.error(f"FFmpeg 合并失败: stderr={result.stderr}")
+                raise AudioMergeError(f"FFmpeg 合并失败: {result.stderr[:500]}")
+
+            with open(output_path, "rb") as f:
+                return f.read()
+
+        except subprocess.TimeoutExpired:
+            raise AudioMergeError("FFmpeg 合并超时（120 秒）")
+        except AudioMergeError:
+            raise
+        except Exception as e:
+            raise AudioMergeError(f"音频合并失败: {e}")
+        finally:
+            shutil.rmtree(temp_dir, ignore_errors=True)

packages/application/tts_job/text_splitter.py

@@ -0,0 +1,70 @@
+"""长文本分段工具 — P1 长文本分段合成。
+
+将超过阈值的文本按句子边界分段，供 CosyVoice 并发合成后合并。
+"""
+
+from __future__ import annotations
+
+# 中文句子结束符（含全角/半角）
+_SENTENCE_ENDS = frozenset("。！？；\n.!?;")
+
+
+def split_text(text: str, max_chars: int = 500) -> list[str]:
+    """将文本分段，每段不超过 max_chars 个字符。
+
+    优先在句子边界（句号、问号、感叹号、换行符）处分段。
+    若单个句子超过 max_chars，则在逗号等次级标点处拆分。
+    若仍超长，则硬切。
+
+    Args:
+        text: 待分段文本
+        max_chars: 每段最大字符数
+
+    Returns:
+        分段列表，每段 ≤ max_chars。文本为空时返回空列表。
+    """
+    text = text.strip()
+    if not text:
+        return []
+    if len(text) <= max_chars:
+        return [text]
+
+    segments: list[str] = []
+    current = ""
+
+    for char in text:
+        current += char
+        if char in _SENTENCE_ENDS and len(current) >= 50:
+            # 句子边界且长度合理，切段
+            segments.append(current.strip())
+            current = ""
+        elif len(current) >= max_chars:
+            # 达到上限，强制切段
+            segments.append(current.strip())
+            current = ""
+
+    if current.strip():
+        segments.append(current.strip())
+
+    # 合并过短的段（< 50 字符且不是最后一段），减少 API 调用次数
+    merged: list[str] = []
+    buffer = ""
+    for seg in segments:
+        if buffer:
+            combined = buffer + seg
+            if len(combined) <= max_chars:
+                buffer = combined
+                continue
+            merged.append(buffer)
+            buffer = ""
+        if len(seg) < 50:
+            buffer = seg
+        else:
+            merged.append(seg)
+    if buffer:
+        if merged and len(merged[-1]) + len(buffer) <= max_chars:
+            merged[-1] = merged[-1] + buffer
+        else:
+            merged.append(buffer)
+
+    return [s for s in merged if s]

packages/application/tts_job/workflow.py

@@ -11,6 +11,11 @@ from __future__ import annotations
 
 import io
 import logging
+import os
+import shutil
+import tempfile
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import Optional
 
 import httpx
@@ -20,12 +25,19 @@ from packages.application.cosyvoice_service import (
     CosyVoiceError,
     CosyVoiceService,
 )
+from packages.application.tts_job.audio_merger import AudioMergeError, AudioMerger
+from packages.application.tts_job.text_splitter import split_text
 from packages.domain.tts_job import TTSJob, TTSJobStatus
 from packages.ports.tts_job_repository import TTSJobRepository
 from packages.shared.storage import SharedStorageService, get_shared_storage_service
 
 logger = logging.getLogger(__name__)
 
+# 长文本分段阈值：超过此字符数自动分段合成
+_SEGMENT_THRESHOLD = 500
+# 分段并发上限
+_MAX_SEGMENT_WORKERS = 5
+
 
 class TTSWorkflowError(Exception):
     """TTS 合成工作流异常。"""
@@ -130,6 +142,10 @@ class TTSWorkflowService:
         job.mark_processing()
         job = self.repository.update(job)
 
+        # 长文本自动分段合成
+        if len(job.input_text) > _SEGMENT_THRESHOLD:
+            return self._start_segment_synthesis(job)
+
         try:
             submit_result = self.cosyvoice_service.submit_synthesize_task(
                 text=job.input_text,
@@ -185,6 +201,11 @@ class TTSWorkflowService:
         if job is None:
             raise TTSJobNotFoundError(f"TTS job {job_id} not found")
 
+        # 检查是否为分段合成任务
+        segment_task_ids = (job.metadata or {}).get("segment_task_ids", [])
+        if segment_task_ids:
+            return self._poll_segment_tasks(job)
+
         task_id = (job.metadata or {}).get("cosyvoice_task_id", "")
         if not task_id:
             raise ValueError(f"TTSJob {job_id} has no cosyvoice_task_id in metadata")
@@ -257,3 +278,224 @@ class TTSWorkflowService:
         job = self.repository.update(job)
         logger.error(f"TTS 合成失败: job_id={job_id}, error={error_message}")
         return job
+
+    # ── P1: 长文本分段合成 ─────────────────────────────────────
+
+    def _upload_merged_to_oss(
+        self, merged_data: bytes, user_id: str, job_id: str, audio_format: str
+    ) -> tuple[str, str]:
+        """上传合并后的音频数据到 OSS。
+
+        Returns:
+            (permanent_url, storage_key) 元组。
+            上传失败时返回 ("", "")。
+        """
+        storage_key = f"tts-outputs/{user_id}/{job_id}.{audio_format}"
+        content_type_map = {
+            "mp3": "audio/mpeg",
+            "wav": "audio/wav",
+            "pcm": "audio/pcm",
+            "opus": "audio/opus",
+        }
+        content_type = content_type_map.get(audio_format, "application/octet-stream")
+        try:
+            file_obj = io.BytesIO(merged_data)
+            permanent_url = self._storage.upload_file(file_obj, storage_key, content_type=content_type)
+            return permanent_url, storage_key
+        except Exception as e:
+            logger.warning(f"分段合并音频转存 OSS 失败: job_id={job_id}, error={e}")
+            return "", ""
+
+    def _start_segment_synthesis(self, job: TTSJob) -> TTSJob:
+        """长文本分段合成入口。
+
+        将文本分段后并发提交到 CosyVoice，根据同步/异步结果走不同路径。
+        """
+        segments = split_text(job.input_text, max_chars=_SEGMENT_THRESHOLD)
+        logger.info(f"长文本分段合成: job_id={job.id}, " f"原文={len(job.input_text)}字, 段数={len(segments)}")
+
+        # 记录分段信息到 metadata
+        job_metadata = dict(job.metadata)
+        job_metadata["segment_count"] = len(segments)
+
+        # 并发提交所有分段
+        results = self._submit_segments_concurrent(segments, job)
+        if results is None:
+            # 提交阶段已失败，_submit_segments_concurrent 内部已标记 failed
+            return self.repository.get(job.id)
+
+        # 判断同步还是异步
+        has_audio_urls = any(r.get("audio_url", "") for r in results)
+        has_task_ids = any(r.get("task_id", "") for r in results)
+
+        if has_audio_urls and not has_task_ids:
+            # 所有分段同步返回音频，直接合并
+            return self._process_segments_sync(job, results)
+
+        # 异步路径：保存各分段的 task_id 供后续轮询
+        segment_task_ids = [r.get("task_id", "") for r in results]
+        segment_audio_urls = [r.get("audio_url", "") for r in results]
+        job_metadata["segment_task_ids"] = segment_task_ids
+        job_metadata["segment_audio_urls"] = segment_audio_urls
+        job_metadata["segment_format"] = job.format
+
+        job.metadata = job_metadata
+        job = self.repository.update(job)
+        logger.info(f"分段合成任务已提交（异步）: job_id={job.id}, " f"段数={len(segments)}")
+        return job
+
+    def _submit_segments_concurrent(self, segments: list[str], job: TTSJob) -> list[dict] | None:
+        """并发提交分段合成任务。
+
+        Returns:
+            各分段的结果列表（保持顺序），提交失败时返回 None。
+        """
+        max_workers = min(len(segments), _MAX_SEGMENT_WORKERS)
+        results: list[dict | None] = [None] * len(segments)
+
+        with ThreadPoolExecutor(max_workers=max_workers) as executor:
+            future_to_idx = {}
+            for idx, segment_text in enumerate(segments):
+                future = executor.submit(
+                    self.cosyvoice_service.submit_synthesize_task,
+                    text=segment_text,
+                    voice_id=job.voice_id,
+                    sample_rate=job.sample_rate,
+                    format=job.format,
+                )
+                future_to_idx[future] = idx
+
+            for future in as_completed(future_to_idx):
+                idx = future_to_idx[future]
+                try:
+                    results[idx] = future.result()
+                except Exception as e:
+                    logger.error(f"分段合成提交失败: job_id={job.id}, " f"segment={idx}, error={e}")
+                    self._handle_segment_failure(job, f"分段 {idx + 1} 合成提交失败: {e}")
+                    return None
+
+        return results  # type: ignore[return-value]
+
+    def _process_segments_sync(self, job: TTSJob, results: list[dict]) -> TTSJob:
+        """同步路径：所有分段已返回 audio_url，下载合并后转存 OSS。"""
+        merged_data, total_duration = self._download_and_merge_segments(results, job)
+
+        # 直接上传合并后的音频 bytes 到 OSS
+        permanent_url, storage_key = self._upload_merged_to_oss(merged_data, job.user_id, job.id, job.format)
+
+        job.mark_completed(
+            output_audio_url=permanent_url,
+            output_audio_key=storage_key,
+            duration=total_duration,
+            file_size=len(merged_data),
+        )
+        job = self.repository.update(job)
+        logger.info(f"分段合成完成: job_id={job.id}, " f"merged_size={len(merged_data)}, duration={total_duration:.1f}")
+        return job
+
+    def _download_and_merge_segments(self, results: list[dict], job: TTSJob) -> tuple[bytes, float]:
+        """下载各分段音频并合并。
+
+        Returns:
+            (merged_audio_bytes, total_duration)
+        """
+        temp_dir = tempfile.mkdtemp(prefix="tts_segments_")
+        try:
+            audio_paths: list[str] = []
+            total_duration = 0.0
+
+            for idx, result in enumerate(results):
+                audio_url = result.get("audio_url", "")
+                if not audio_url:
+                    raise TTSWorkflowError(f"分段 {idx + 1} 没有返回 audio_url")
+
+                total_duration += result.get("duration", 0.0)
+
+                # 下载分段音频到临时文件
+                resp = httpx.get(audio_url, timeout=60.0, follow_redirects=True)
+                resp.raise_for_status()
+
+                seg_path = os.path.join(temp_dir, f"seg_{idx:03d}.{job.format}")
+                with open(seg_path, "wb") as f:
+                    f.write(resp.content)
+                audio_paths.append(seg_path)
+
+            # 合并
+            merger = AudioMerger()
+            merged_data = merger.merge(audio_paths, output_format=job.format)
+            return merged_data, total_duration
+
+        finally:
+            shutil.rmtree(temp_dir, ignore_errors=True)
+
+    def _poll_segment_tasks(self, job: TTSJob) -> TTSJob:
+        """轮询所有分段异步任务，全部完成后合并音频。"""
+        segment_task_ids: list[str] = (job.metadata or {}).get("segment_task_ids", [])
+        segment_audio_urls: list[str] = (job.metadata or {}).get("segment_audio_urls", [])
+        segment_count = len(segment_task_ids)
+
+        poll_start = time.monotonic()
+        poll_timeout = 300.0  # 分段任务超时更长
+        poll_interval = 2.0
+
+        while time.monotonic() - poll_start < poll_timeout:
+            all_done = True
+            results: list[dict | None] = [None] * segment_count
+
+            for idx, task_id in enumerate(segment_task_ids):
+                # 已经有音频的分段跳过轮询
+                if idx < len(segment_audio_urls) and segment_audio_urls[idx]:
+                    results[idx] = {
+                        "audio_url": segment_audio_urls[idx],
+                        "duration": 0.0,
+                        "file_size": 0,
+                    }
+                    continue
+
+                try:
+                    result = self.cosyvoice_service.poll_synthesize_task(task_id, timeout=poll_timeout)
+                    results[idx] = result
+                except Exception as e:
+                    logger.error(f"分段任务轮询失败: job_id={job.id}, " f"segment={idx}, error={e}")
+                    self._handle_segment_failure(job, f"分段 {idx + 1} 轮询失败: {e}")
+                    return self.repository.get(job.id)
+
+                if results[idx] is None:
+                    all_done = False
+
+            if all_done and all(r is not None for r in results):
+                # 所有分段完成，下载合并
+                try:
+                    merged_data, total_duration = self._download_and_merge_segments(results, job)
+
+                    # 转存 OSS
+                    permanent_url, storage_key = self._upload_merged_to_oss(
+                        merged_data, job.user_id, job.id, job.format
+                    )
+
+                    job.mark_completed(
+                        output_audio_url=permanent_url,
+                        output_audio_key=storage_key,
+                        duration=total_duration,
+                        file_size=len(merged_data),
+                    )
+                    job = self.repository.update(job)
+                    logger.info(f"分段合成轮询完成: job_id={job.id}, " f"merged_size={len(merged_data)}")
+                    return job
+
+                except Exception as e:
+                    self._handle_segment_failure(job, f"分段合并失败: {e}")
+                    return self.repository.get(job.id)
+
+            # 等待后重试
+            time.sleep(poll_interval)
+
+        # 超时
+        self._handle_segment_failure(job, "分段合成轮询超时（300 秒）")
+        return self.repository.get(job.id)
+
+    def _handle_segment_failure(self, job: TTSJob, error_message: str) -> None:
+        """分段合成失败处理。"""
+        job.mark_failed(error_message)
+        self.repository.update(job)
+        logger.error(f"分段合成失败: job_id={job.id}, error={error_message}")

tests/unit/test_tts_segment_synthesis.py

@@ -0,0 +1,485 @@
+"""P1 长文本分段合成单元测试。
+
+覆盖：
+- text_splitter.split_text 分段逻辑
+- audio_merger.AudioMerger 合并逻辑
+- workflow 分段合成路径（同步 / 异步 / 失败）
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, call, patch
+
+import pytest
+
+from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService
+from packages.application.tts_job.audio_merger import AudioMergeError, AudioMerger
+from packages.application.tts_job.text_splitter import split_text
+from packages.application.tts_job.workflow import TTSWorkflowService
+from packages.domain.tts_job import TTSJob, TTSJobStatus
+
+# ── text_splitter ─────────────────────────────────────────────
+
+
+class TestSplitText:
+    """测试文本分段工具。"""
+
+    def test_short_text_no_split(self) -> None:
+        """短文本不拆分。"""
+        assert split_text("你好世界", max_chars=500) == ["你好世界"]
+
+    def test_empty_text(self) -> None:
+        """空文本返回空列表。"""
+        assert split_text("") == []
+        assert split_text("   ") == []
+
+    def test_exact_threshold(self) -> None:
+        """恰好等于阈值不拆分。"""
+        text = "a" * 500
+        assert split_text(text, max_chars=500) == [text]
+
+    def test_split_at_sentence_boundary(self) -> None:
+        """在句子边界处分段。"""
+        text = "第一句话。" * 60  # 300 chars
+        text += "第二句话。" * 60  # 300 chars → total 600
+        segments = split_text(text, max_chars=500)
+        assert len(segments) >= 2
+        for seg in segments:
+            assert len(seg) <= 500
+
+    def test_split_at_newline(self) -> None:
+        """在换行符处分段。"""
+        text = "段落一\n" * 100  # 300 chars
+        text += "段落二\n" * 100  # 300 chars
+        segments = split_text(text, max_chars=500)
+        assert len(segments) >= 2
+
+    def test_long_sentence_hard_split(self) -> None:
+        """超长句子硬切。"""
+        text = "a" * 1200
+        segments = split_text(text, max_chars=500)
+        assert len(segments) >= 3
+        for seg in segments:
+            assert len(seg) <= 500
+
+    def test_merge_short_segments(self) -> None:
+        """短段合并减少 API 调用。"""
+        # 多个短句子应该被合并
+        text = "你好。" * 120  # 360 chars, each sentence 3 chars
+        segments = split_text(text, max_chars=500)
+        # 短段应该被合并，段数应该比较少
+        assert len(segments) < 120
+
+    def test_preserves_order(self) -> None:
+        """分段保持原始顺序。"""
+        text = "第一段。第二段。第三段。" + "x" * 490
+        segments = split_text(text, max_chars=500)
+        # 第一个段应该以 "第一段" 开头
+        assert segments[0].startswith("第一段")
+
+
+# ── audio_merger ─────────────────────────────────────────────
+
+
+class TestAudioMerger:
+    """测试 FFmpeg 音频合并器。"""
+
+    def test_empty_list_raises(self) -> None:
+        """空列表抛出 AudioMergeError。"""
+        merger = AudioMerger()
+        with pytest.raises(AudioMergeError, match="没有可合并"):
+            merger.merge([])
+
+    def test_single_file_returns_bytes(self) -> None:
+        """单文件直接返回内容。"""
+        with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
+            f.write(b"fake audio content")
+            f.flush()
+            path = f.name
+
+        try:
+            merger = AudioMerger()
+            data = merger.merge([path])
+            assert data == b"fake audio content"
+        finally:
+            os.unlink(path)
+
+    @patch("packages.application.tts_job.audio_merger.subprocess.run")
+    def test_ffmpeg_called_correctly(self, mock_run: MagicMock) -> None:
+        """多文件调用 FFmpeg concat。"""
+        mock_run.return_value = MagicMock(returncode=0)
+
+        # 创建临时文件
+        paths = []
+        for i in range(3):
+            with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
+                f.write(b"audio")
+                paths.append(f.name)
+
+        try:
+            merger = AudioMerger()
+            # Mock open for reading the merged output
+            with patch("builtins.open", create=True) as mock_open:
+                mock_open.return_value.__enter__ = lambda s: s
+                mock_open.return_value.read = lambda: b"merged audio"
+                try:
+                    merger.merge(paths, output_format="mp3")
+                except (FileNotFoundError, OSError):
+                    pass  # Expected since we're mocking
+
+            # 验证 FFmpeg 被调用
+            mock_run.assert_called_once()
+            cmd = mock_run.call_args[0][0]
+            assert cmd[0] == "ffmpeg"
+            assert "-f" in cmd
+            assert "concat" in cmd
+        finally:
+            for p in paths:
+                os.unlink(p)
+
+    @patch("packages.application.tts_job.audio_merger.subprocess.run")
+    def test_ffmpeg_failure_raises(self, mock_run: MagicMock) -> None:
+        """FFmpeg 失败抛出 AudioMergeError。"""
+        mock_run.return_value = MagicMock(returncode=1, stderr="error details")
+
+        paths = []
+        for i in range(2):
+            with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
+                f.write(b"audio")
+                paths.append(f.name)
+
+        try:
+            merger = AudioMerger()
+            with pytest.raises(AudioMergeError, match="FFmpeg 合并失败"):
+                merger.merge(paths)
+        finally:
+            for p in paths:
+                os.unlink(p)
+
+
+# ── workflow segment methods ─────────────────────────────────
+
+
+def _make_job(**kwargs) -> TTSJob:
+    defaults = {
+        "id": "test_job_seg",
+        "user_id": "user_001",
+        "input_text": "x" * 600,  # > 500 threshold
+        "voice_id": "voice_001",
+        "voice_model": "",
+        "project_id": "",
+        "voice_clone_profile_id": "",
+        "status": TTSJobStatus.PENDING,
+        "output_audio_url": "",
+        "output_audio_key": "",
+        "duration": 0.0,
+        "file_size": 0,
+        "sample_rate": 22050,
+        "format": "mp3",
+        "error_message": "",
+        "retry_count": 0,
+        "max_retries": 3,
+        "metadata": {},
+        "started_at": None,
+        "completed_at": None,
+        "created_at": datetime.now(timezone.utc),
+        "updated_at": datetime.now(timezone.utc),
+    }
+    defaults.update(kwargs)
+    return TTSJob(**defaults)
+
+
+def _make_workflow(
+    cosyvoice_service: MagicMock | None = None,
+    repo: MagicMock | None = None,
+    storage: MagicMock | None = None,
+) -> TTSWorkflowService:
+    if cosyvoice_service is None:
+        cosyvoice_service = MagicMock(spec=CosyVoiceService)
+    if repo is None:
+        repo = MagicMock()
+        repo.get.return_value = _make_job()
+        repo.update.side_effect = lambda j: j
+    if storage is None:
+        storage = MagicMock()
+        storage.upload_file.return_value = "https://oss.example.com/merged.mp3"
+    return TTSWorkflowService(
+        repository=repo,
+        cosyvoice_service=cosyvoice_service,
+        storage_service=storage,
+    )
+
+
+class TestStartSegmentSynthesis:
+    """测试 _start_segment_synthesis 分段合成入口。"""
+
+    def test_short_text_no_segment(self) -> None:
+        """短文本不触发分段。"""
+        service = MagicMock(spec=CosyVoiceService)
+        service.submit_synthesize_task.return_value = {
+            "task_id": "task_1",
+            "audio_url": "",
+        }
+        repo = MagicMock()
+        repo.get.return_value = _make_job(input_text="短文本")
+        repo.update.side_effect = lambda j: j
+
+        workflow = _make_workflow(cosyvoice_service=service, repo=repo)
+        job = workflow.start_synthesis("test_job_seg")
+
+        # 短文本走普通路径，不调用分段
+        assert job.status == TTSJobStatus.PROCESSING
+
+    @patch("packages.application.tts_job.workflow.httpx")
+    def test_long_text_sync_segments(self, mock_httpx: MagicMock) -> None:
+        """长文本同步分段：所有段立即返回 audio_url，直接合并。"""
+        # Mock 分段音频下载
+        mock_resp = MagicMock()
+        mock_resp.content = b"segment audio"
+        mock_resp.raise_for_status.return_value = None
+        mock_httpx.get.return_value = mock_resp
+
+        service = MagicMock(spec=CosyVoiceService)
+        # 每个分段都同步返回 audio_url
+        service.submit_synthesize_task.side_effect = [
+            {"task_id": "", "audio_url": "https://temp.com/seg1.mp3", "duration": 2.0, "file_size": 1000},
+            {"task_id": "", "audio_url": "https://temp.com/seg2.mp3", "duration": 3.0, "file_size": 1500},
+        ]
+
+        storage = MagicMock()
+        storage.upload_file.return_value = "https://oss.example.com/merged.mp3"
+
+        repo = MagicMock()
+        repo.get.return_value = _make_job()
+        repo.update.side_effect = lambda j: j
+
+        workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage)
+
+        # Mock AudioMerger 避免真实 FFmpeg 调用
+        with patch("packages.application.tts_job.workflow.AudioMerger") as MockMerger:
+            mock_merger = MagicMock()
+            mock_merger.merge.return_value = b"merged audio data"
+            MockMerger.return_value = mock_merger
+
+            job = workflow.start_synthesis("test_job_seg")
+
+        assert job.status == TTSJobStatus.COMPLETED
+        assert job.output_audio_url == "https://oss.example.com/merged.mp3"
+        assert job.duration == 5.0  # 2.0 + 3.0
+
+    def test_long_text_async_segments(self) -> None:
+        """长文本异步分段：返回 task_id，存入 metadata。"""
+        service = MagicMock(spec=CosyVoiceService)
+        # 每个分段返回 task_id（异步）
+        service.submit_synthesize_task.side_effect = [
+            {"task_id": "seg_task_1", "audio_url": "", "duration": 0.0, "file_size": 0},
+            {"task_id": "seg_task_2", "audio_url": "", "duration": 0.0, "file_size": 0},
+        ]
+
+        repo = MagicMock()
+        repo.get.return_value = _make_job()
+        repo.update.side_effect = lambda j: j
+
+        workflow = _make_workflow(cosyvoice_service=service, repo=repo)
+        job = workflow.start_synthesis("test_job_seg")
+
+        assert job.status == TTSJobStatus.PROCESSING
+        assert "segment_task_ids" in job.metadata
+        assert job.metadata["segment_task_ids"] == ["seg_task_1", "seg_task_2"]
+
+    def test_segment_submit_failure_marks_failed(self) -> None:
+        """分段提交失败时标记 job 为 failed。"""
+        service = MagicMock(spec=CosyVoiceService)
+        service.submit_synthesize_task.side_effect = CosyVoiceError("API error")
+
+        repo = MagicMock()
+        repo.get.return_value = _make_job()
+        repo.update.side_effect = lambda j: j
+
+        workflow = _make_workflow(cosyvoice_service=service, repo=repo)
+        job = workflow.start_synthesis("test_job_seg")
+
+        assert job.status == TTSJobStatus.FAILED
+
+
+class TestUploadMergedToOSS:
+    """测试 _upload_merged_to_oss 辅助方法。"""
+
+    def test_success(self) -> None:
+        """成功上传返回 URL 和 key。"""
+        storage = MagicMock()
+        storage.upload_file.return_value = "https://oss.example.com/merged.mp3"
+
+        workflow = _make_workflow(storage=storage)
+        url, key = workflow._upload_merged_to_oss(b"audio data", "user_001", "job_001", "mp3")
+
+        assert url == "https://oss.example.com/merged.mp3"
+        assert key == "tts-outputs/user_001/job_001.mp3"
+        storage.upload_file.assert_called_once()
+        call_args = storage.upload_file.call_args
+        assert call_args[1]["content_type"] == "audio/mpeg"
+
+    def test_failure_returns_empty(self) -> None:
+        """上传失败返回空字符串。"""
+        storage = MagicMock()
+        storage.upload_file.side_effect = Exception("OSS error")
+
+        workflow = _make_workflow(storage=storage)
+        url, key = workflow._upload_merged_to_oss(b"audio data", "user_001", "job_001", "mp3")
+
+        assert url == ""
+        assert key == ""
+
+
+class TestHandleSegmentFailure:
+    """测试 _handle_segment_failure。"""
+
+    def test_marks_job_failed(self) -> None:
+        """标记 job 为 failed 并更新。"""
+        repo = MagicMock()
+        job = _make_job(status=TTSJobStatus.PROCESSING)
+        repo.get.return_value = job
+        repo.update.side_effect = lambda j: j
+
+        workflow = _make_workflow(repo=repo)
+        workflow._handle_segment_failure(job, "分段 1 合成失败")
+
+        assert job.status == TTSJobStatus.FAILED
+        assert "分段 1 合成失败" in job.error_message
+        repo.update.assert_called_once()
+
+
+class TestPollSegmentTasks:
+    """测试 _poll_segment_tasks 异步轮询。"""
+
+    @patch("packages.application.tts_job.workflow.time")
+    @patch("packages.application.tts_job.workflow.httpx")
+    def test_all_segments_done(self, mock_httpx: MagicMock, mock_time: MagicMock) -> None:
+        """所有分段完成后合并并标记完成。"""
+        # Mock time.monotonic 让循环只执行一次
+        mock_time.monotonic.side_effect = [0.0, 1.0, 2.0]
+        mock_time.sleep = MagicMock()
+
+        # Mock 下载分段音频
+        mock_resp = MagicMock()
+        mock_resp.content = b"seg audio"
+        mock_resp.raise_for_status.return_value = None
+        mock_httpx.get.return_value = mock_resp
+
+        service = MagicMock(spec=CosyVoiceService)
+        service.poll_synthesize_task.side_effect = [
+            {"audio_url": "https://temp.com/seg1.mp3", "duration": 2.0, "file_size": 100},
+            {"audio_url": "https://temp.com/seg2.mp3", "duration": 3.0, "file_size": 200},
+        ]
+
+        storage = MagicMock()
+        storage.upload_file.return_value = "https://oss.example.com/merged.mp3"
+
+        repo = MagicMock()
+        job = _make_job(
+            status=TTSJobStatus.PROCESSING,
+            metadata={
+                "segment_task_ids": ["task_1", "task_2"],
+                "segment_audio_urls": ["", ""],
+                "segment_format": "mp3",
+            },
+        )
+        repo.get.return_value = job
+        repo.update.side_effect = lambda j: j
+
+        workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage)
+
+        with patch("packages.application.tts_job.workflow.AudioMerger") as MockMerger:
+            mock_merger = MagicMock()
+            mock_merger.merge.return_value = b"merged data"
+            MockMerger.return_value = mock_merger
+
+            result = workflow._poll_segment_tasks(job)
+
+        assert result.status == TTSJobStatus.COMPLETED
+
+    @patch("packages.application.tts_job.workflow.time")
+    def test_segment_poll_failure(self, mock_time: MagicMock) -> None:
+        """分段轮询失败时标记 job failed。"""
+        mock_time.monotonic.side_effect = [0.0, 1.0]
+        mock_time.sleep = MagicMock()
+
+        service = MagicMock(spec=CosyVoiceService)
+        service.poll_synthesize_task.side_effect = CosyVoiceError("Poll failed")
+
+        repo = MagicMock()
+        job = _make_job(
+            status=TTSJobStatus.PROCESSING,
+            metadata={
+                "segment_task_ids": ["task_1"],
+                "segment_audio_urls": [""],
+            },
+        )
+        repo.get.return_value = job
+        repo.update.side_effect = lambda j: j
+
+        workflow = _make_workflow(cosyvoice_service=service, repo=repo)
+        result = workflow._poll_segment_tasks(job)
+
+        assert result.status == TTSJobStatus.FAILED
+
+
+class TestPollAndProcessSynthesisSegmentDetection:
+    """测试 poll_and_process_synthesis 正确识别分段任务。"""
+
+    def test_detects_segment_task(self) -> None:
+        """metadata 中有 segment_task_ids 时走分段轮询路径。"""
+        service = MagicMock(spec=CosyVoiceService)
+
+        repo = MagicMock()
+        job = _make_job(
+            status=TTSJobStatus.PROCESSING,
+            metadata={
+                "segment_task_ids": ["task_1", "task_2"],
+                "segment_audio_urls": ["", ""],
+            },
+        )
+        repo.get.return_value = job
+        repo.update.side_effect = lambda j: j
+
+        workflow = _make_workflow(cosyvoice_service=service, repo=repo)
+
+        with patch.object(workflow, "_poll_segment_tasks") as mock_poll:
+            mock_poll.return_value = job
+            workflow.poll_and_process_synthesis("test_job_seg")
+            mock_poll.assert_called_once()
+
+    def test_normal_task_no_segment(self) -> None:
+        """普通任务不走分段路径。"""
+        service = MagicMock(spec=CosyVoiceService)
+        service.poll_synthesize_task.return_value = {
+            "audio_url": "https://temp.com/audio.mp3",
+            "duration": 5.0,
+            "file_size": 5000,
+        }
+
+        repo = MagicMock()
+        job = _make_job(
+            status=TTSJobStatus.PROCESSING,
+            metadata={"cosyvoice_task_id": "task_normal"},
+        )
+        repo.get.return_value = job
+        repo.update.side_effect = lambda j: j
+
+        storage = MagicMock()
+        storage.upload_file.return_value = "https://oss.example.com/audio.mp3"
+
+        workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage)
+
+        with patch("packages.application.tts_job.workflow.httpx") as mock_httpx:
+            mock_resp = MagicMock()
+            mock_resp.content = b"audio"
+            mock_resp.raise_for_status.return_value = None
+            mock_httpx.get.return_value = mock_resp
+
+            result = workflow.poll_and_process_synthesis("test_job_seg")
+
+        assert result.status == TTSJobStatus.COMPLETED