diff --git a/tests/unit/test_tts_workflow_service.py b/tests/unit/test_tts_workflow_service.py
new file mode 100755
index 000000000..daf1836f2
--- /dev/null
+++ b/tests/unit/test_tts_workflow_service.py
@@ -0,0 +1,549 @@
+"""TTS Workflow service unit tests.
+
+Covers TTSWorkflowService - start_synthesis, poll_and_process_synthesis,
+process_synthesis_result, process_synthesis_failure, segment synthesis.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from packages.application.cosyvoice_service import CosyVoiceAuthError, CosyVoiceError
+from packages.application.tts_job.workflow import (
+    TTSJobNotFoundError,
+    TTSWorkflowError,
+    TTSWorkflowService,
+)
+from packages.domain.tts_job import TTSJob, TTSJobStatus
+
+# ── Helpers ─────────────────────────────────────────────
+
+
+def make_job(**kwargs):
+    defaults = dict(
+        id="job-123",
+        user_id="user-1",
+        input_text="Hello world",
+        voice_id="voice-1",
+        sample_rate=22050,
+        format="mp3",
+    )
+    defaults.update(kwargs)
+    return TTSJob(**defaults)
+
+
+class FakeTTSJobRepository:
+    def __init__(self, job=None):
+        self._job = job
+        self.updated_jobs = []
+        self.get_called = 0
+
+    def get(self, job_id):
+        self.get_called += 1
+        if self._job and self._job.id == job_id:
+            return self._job
+        return None
+
+    def update(self, job):
+        self.updated_jobs.append(job)
+        self._job = job
+        return job
+
+
+class FakeCosyVoiceService:
+    def __init__(self, submit_result=None, submit_error=None, poll_result=None, poll_error=None):
+        self._submit_result = submit_result or {
+            "audio_url": "https://temp.example.com/audio.mp3",
+            "task_id": "",
+            "request_id": "req-1",
+            "duration": 5.0,
+            "file_size": 1024,
+        }
+        self._submit_error = submit_error
+        self._poll_result = poll_result
+        self._poll_error = poll_error
+        self.submit_calls = []
+        self.poll_calls = []
+
+    def submit_synthesize_task(self, **kwargs):
+        self.submit_calls.append(kwargs)
+        if self._submit_error:
+            raise self._submit_error
+        return self._submit_result
+
+    def poll_synthesize_task(self, task_id, timeout=120.0):
+        self.poll_calls.append({"task_id": task_id, "timeout": timeout})
+        if self._poll_error:
+            raise self._poll_error
+        return self._poll_result or {
+            "audio_url": "https://temp.example.com/audio.mp3",
+            "duration": 5.0,
+            "file_size": 1024,
+        }
+
+
+class FakeStorageService:
+    def __init__(self, upload_url="https://oss.example.com/tts-outputs/user-1/job-123.mp3", upload_error=None):
+        self._upload_url = upload_url
+        self._upload_error = upload_error
+        self.uploads = []
+
+    def upload_file(self, file_obj, storage_key, content_type=None):
+        self.uploads.append({"storage_key": storage_key, "content_type": content_type})
+        if self._upload_error:
+            raise self._upload_error
+        return self._upload_url
+
+
+# ── start_synthesis tests ───────────────────────────────
+
+
+class TestStartSynthesis:
+    def test_successful_sync_completion(self):
+        """start_synthesis with sync audio_url → job marked completed."""
+        job = make_job()
+        repo = FakeTTSJobRepository(job=job)
+        cosy = FakeCosyVoiceService(
+            submit_result={
+                "audio_url": "https://temp.example.com/audio.mp3",
+                "request_id": "req-abc",
+                "task_id": "",
+                "duration": 3.5,
+                "file_size": 5000,
+            }
+        )
+        storage = FakeStorageService(upload_url="https://oss.example.com/final.mp3")
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)
+
+        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"fake audio data"):
+            result = svc.start_synthesis("job-123")
+
+        assert result.id == "job-123"
+        assert result.status == TTSJobStatus.COMPLETED.value
+        assert result.output_audio_url == "https://oss.example.com/final.mp3"
+        assert result.output_audio_key == "tts-outputs/user-1/job-123.mp3"
+        assert result.duration == 3.5
+        assert result.file_size == 5000
+        assert result.metadata["cosyvoice_request_id"] == "req-abc"
+
+        # verify cosyvoice was called
+        assert len(cosy.submit_calls) == 1
+        assert cosy.submit_calls[0]["text"] == "Hello world"
+        assert cosy.submit_calls[0]["voice_id"] == "voice-1"
+
+        # verify storage upload
+        assert len(storage.uploads) == 1
+        assert storage.uploads[0]["storage_key"] == "tts-outputs/user-1/job-123.mp3"
+
+    def test_job_not_found_raises(self):
+        repo = FakeTTSJobRepository()  # no job
+        cosy = FakeCosyVoiceService()
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)
+
+        with pytest.raises(TTSJobNotFoundError):
+            svc.start_synthesis("nonexistent")
+
+    def test_job_marked_processing_before_submit(self):
+        job = make_job()
+        assert job.status == TTSJobStatus.PENDING.value
+
+        repo = FakeTTSJobRepository(job=job)
+        cosy = FakeCosyVoiceService(
+            submit_result={
+                "audio_url": "",
+                "task_id": "task-abc",
+                "request_id": "req-1",
+            }
+        )
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=FakeStorageService())
+
+        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"):
+            svc.start_synthesis("job-123")
+
+        # first update should mark processing
+        first_update = repo.updated_jobs[0]
+        assert first_update.status == TTSJobStatus.PROCESSING.value
+
+    def test_cosyvoice_error_marks_failed(self):
+        job = make_job()
+        repo = FakeTTSJobRepository(job=job)
+        cosy = FakeCosyVoiceService(submit_error=CosyVoiceError("API rate limit"))
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)
+
+        result = svc.start_synthesis("job-123")
+
+        assert result.status == TTSJobStatus.FAILED.value
+        assert "API rate limit" in result.error_message
+
+    def test_cosyvoice_auth_error_marks_failed(self):
+        job = make_job()
+        repo = FakeTTSJobRepository(job=job)
+        cosy = FakeCosyVoiceService(submit_error=CosyVoiceAuthError("Invalid key"))
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)
+
+        result = svc.start_synthesis("job-123")
+
+        assert result.status == TTSJobStatus.FAILED.value
+        assert "Invalid key" in result.error_message
+
+    def test_value_error_marks_failed(self):
+        job = make_job()
+        repo = FakeTTSJobRepository(job=job)
+        cosy = FakeCosyVoiceService(submit_error=ValueError("text is empty"))
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)
+
+        result = svc.start_synthesis("job-123")
+
+        assert result.status == TTSJobStatus.FAILED.value
+        assert "text is empty" in result.error_message
+
+    def test_oss_transfer_failure_falls_back_to_temp_url(self):
+        job = make_job()
+        repo = FakeTTSJobRepository(job=job)
+        cosy = FakeCosyVoiceService(
+            submit_result={
+                "audio_url": "https://temp.example.com/original.mp3",
+                "request_id": "req-1",
+                "task_id": "",
+                "duration": 2.0,
+                "file_size": 1000,
+            }
+        )
+        storage = FakeStorageService(upload_error=RuntimeError("OSS down"))
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)
+
+        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"fake"):
+            result = svc.start_synthesis("job-123")
+
+        assert result.status == TTSJobStatus.COMPLETED.value
+        # falls back to temp URL
+        assert result.output_audio_url == "https://temp.example.com/original.mp3"
+        assert result.output_audio_key == ""
+
+    def test_no_audio_url_task_id_saved(self):
+        """Async path: no audio_url, save task_id to metadata."""
+        job = make_job()
+        repo = FakeTTSJobRepository(job=job)
+        cosy = FakeCosyVoiceService(
+            submit_result={
+                "audio_url": "",
+                "task_id": "async-task-123",
+                "request_id": "req-async",
+            }
+        )
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=FakeStorageService())
+
+        result = svc.start_synthesis("job-123")
+
+        assert result.status == TTSJobStatus.PROCESSING.value
+        assert result.metadata["cosyvoice_task_id"] == "async-task-123"
+        assert result.metadata["cosyvoice_request_id"] == "req-async"
+
+
+# ── poll_and_process_synthesis tests ────────────────────
+
+
+class TestPollAndProcessSynthesis:
+    def test_already_completed_returns_immediately(self):
+        job = make_job()
+        job.mark_processing()
+        job.mark_completed(output_audio_url="https://oss.example.com/done.mp3", output_audio_key="key")
+        repo = FakeTTSJobRepository(job=job)
+        cosy = FakeCosyVoiceService()
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)
+
+        result = svc.poll_and_process_synthesis("job-123")
+
+        assert result.status == TTSJobStatus.COMPLETED.value
+        # no new submit calls
+        assert len(cosy.submit_calls) == 0
+        assert len(cosy.poll_calls) == 0
+
+    def test_job_not_found_raises(self):
+        repo = FakeTTSJobRepository()
+        cosy = FakeCosyVoiceService()
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)
+
+        with pytest.raises(TTSJobNotFoundError):
+            svc.poll_and_process_synthesis("nonexistent")
+
+    def test_no_task_id_resynthesizes(self):
+        """No task_id in metadata → re-sync synthesize."""
+        job = make_job()
+        job.mark_processing()
+        job.metadata = {}  # no task_id
+        repo = FakeTTSJobRepository(job=job)
+        cosy = FakeCosyVoiceService(
+            submit_result={
+                "audio_url": "https://temp.example.com/retry.mp3",
+                "request_id": "req-retry",
+                "task_id": "",
+                "duration": 3.0,
+                "file_size": 2048,
+            }
+        )
+        storage = FakeStorageService(upload_url="https://oss.example.com/retry-final.mp3")
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)
+
+        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"):
+            result = svc.poll_and_process_synthesis("job-123")
+
+        assert result.status == TTSJobStatus.COMPLETED.value
+        assert result.output_audio_url == "https://oss.example.com/retry-final.mp3"
+        assert len(cosy.submit_calls) == 1
+
+    def test_old_task_id_poll_fails_resynthesizes(self):
+        """Old task_id poll fails → fallback to re-synthesize."""
+        job = make_job()
+        job.mark_processing()
+        job.metadata = {"cosyvoice_task_id": "old-task-id"}
+        repo = FakeTTSJobRepository(job=job)
+        cosy = FakeCosyVoiceService(
+            poll_error=CosyVoiceError("task not found"),
+            submit_result={
+                "audio_url": "https://temp.example.com/retry2.mp3",
+                "request_id": "req-retry2",
+                "task_id": "",
+                "duration": 3.0,
+                "file_size": 1000,
+            },
+        )
+        storage = FakeStorageService()
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)
+
+        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"):
+            result = svc.poll_and_process_synthesis("job-123")
+
+        assert len(cosy.poll_calls) == 1
+        assert cosy.poll_calls[0]["task_id"] == "old-task-id"
+        assert len(cosy.submit_calls) == 1  # resynthesize
+        assert result.status == TTSJobStatus.COMPLETED.value
+
+
+# ── process_synthesis_result tests ──────────────────────
+
+
+class TestProcessSynthesisResult:
+    def test_success_marks_completed(self):
+        job = make_job()
+        job.mark_processing()
+        repo = FakeTTSJobRepository(job=job)
+        cosy = FakeCosyVoiceService()
+        storage = FakeStorageService(upload_url="https://oss.example.com/final.mp3")
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)
+
+        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"fake audio"):
+            result = svc.process_synthesis_result(
+                "job-123",
+                audio_url="https://temp.example.com/audio.mp3",
+                duration=10.5,
+                file_size=50000,
+            )
+
+        assert result.status == TTSJobStatus.COMPLETED.value
+        assert result.output_audio_url == "https://oss.example.com/final.mp3"
+        assert result.output_audio_key == "tts-outputs/user-1/job-123.mp3"
+        assert result.duration == 10.5
+        assert result.file_size == 50000
+
+    def test_job_not_found_raises(self):
+        repo = FakeTTSJobRepository()
+        cosy = FakeCosyVoiceService()
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)
+
+        with pytest.raises(TTSJobNotFoundError):
+            svc.process_synthesis_result("nonexistent", audio_url="https://x.mp3")
+
+    def test_oss_upload_failure_uses_temp_url(self):
+        job = make_job()
+        job.mark_processing()
+        repo = FakeTTSJobRepository(job=job)
+        storage = FakeStorageService(upload_error=RuntimeError("network error"))
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=FakeCosyVoiceService(), storage_service=storage)
+
+        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"data"):
+            result = svc.process_synthesis_result("job-123", audio_url="https://temp.example.com/x.mp3")
+
+        assert result.status == TTSJobStatus.COMPLETED.value
+        assert result.output_audio_url == "https://temp.example.com/x.mp3"
+        assert result.output_audio_key == ""
+
+
+# ── process_synthesis_failure tests ─────────────────────
+
+
+class TestProcessSynthesisFailure:
+    def test_marks_job_failed(self):
+        job = make_job()
+        job.mark_processing()
+        repo = FakeTTSJobRepository(job=job)
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=FakeCosyVoiceService())
+
+        result = svc.process_synthesis_failure("job-123", "CosyVoice 429 rate limited")
+
+        assert result.status == TTSJobStatus.FAILED.value
+        assert "429 rate limited" in result.error_message
+
+    def test_job_not_found_raises(self):
+        repo = FakeTTSJobRepository()
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=FakeCosyVoiceService())
+
+        with pytest.raises(TTSJobNotFoundError):
+            svc.process_synthesis_failure("nonexistent", "error")
+
+
+# ── Segment synthesis tests ─────────────────────────────
+
+
+class TestSegmentSynthesis:
+    def test_long_text_triggers_segment_synthesis(self):
+        """Text over 500 chars → segment synthesis path."""
+        long_text = "你好" * 300  # 600 chars, over 500 threshold
+        job = make_job(input_text=long_text)
+        repo = FakeTTSJobRepository(job=job)
+
+        # Each segment returns audio_url (sync path)
+        def mock_submit(**kwargs):
+            return {
+                "audio_url": f"https://seg.example.com/{kwargs.get('text', '')[:10]}.mp3",
+                "task_id": "",
+                "request_id": "r",
+                "duration": 1.0,
+                "file_size": 100,
+            }
+
+        cosy = FakeCosyVoiceService()
+        cosy.submit_synthesize_task = MagicMock(side_effect=mock_submit)
+        storage = FakeStorageService()
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)
+
+        with (
+            patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"),
+            patch("packages.application.tts_job.workflow.safe_download_file") as mock_download,
+            patch("packages.application.tts_job.workflow.AudioMerger") as mock_merger_class,
+        ):
+            mock_merger = MagicMock()
+            mock_merger.merge.return_value = b"merged audio data"
+            mock_merger_class.return_value = mock_merger
+
+            result = svc.start_synthesis("job-123")
+
+        assert result.status == TTSJobStatus.COMPLETED.value
+        assert result.file_size == len(b"merged audio data")
+        # Multiple segments submitted
+        assert cosy.submit_synthesize_task.call_count >= 2
+
+    def test_segment_failure_marks_job_failed(self):
+        """One segment fails → whole job fails."""
+        long_text = "A" * 600
+        job = make_job(input_text=long_text)
+        repo = FakeTTSJobRepository(job=job)
+
+        call_count = {"n": 0}
+
+        def mock_submit(**kwargs):
+            call_count["n"] += 1
+            if call_count["n"] == 2:
+                raise CosyVoiceError("segment 2 failed")
+            return {
+                "audio_url": "https://seg.example.com/s.mp3",
+                "task_id": "",
+                "request_id": "r",
+                "duration": 1.0,
+                "file_size": 100,
+            }
+
+        cosy = FakeCosyVoiceService()
+        cosy.submit_synthesize_task = MagicMock(side_effect=mock_submit)
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=FakeStorageService())
+
+        result = svc.start_synthesis("job-123")
+
+        assert result.status == TTSJobStatus.FAILED.value
+        assert "分段" in result.error_message
+
+    def test_segment_task_ids_saved_for_async(self):
+        """Async segment results → task_ids saved to metadata."""
+        long_text = "B" * 600
+        job = make_job(input_text=long_text)
+        repo = FakeTTSJobRepository(job=job)
+
+        def mock_submit(**kwargs):
+            return {"audio_url": "", "task_id": f"task-{kwargs.get('text', '')[:5]}", "request_id": "r"}
+
+        cosy = FakeCosyVoiceService()
+        cosy.submit_synthesize_task = MagicMock(side_effect=mock_submit)
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=FakeStorageService())
+
+        result = svc.start_synthesis("job-123")
+
+        assert result.status == TTSJobStatus.PROCESSING.value
+        assert "segment_task_ids" in result.metadata
+        assert len(result.metadata["segment_task_ids"]) >= 2
+        assert result.metadata["segment_count"] >= 2
+
+
+# ── _resynthesize_and_complete tests ────────────────────
+
+
+class TestResynthesizeAndComplete:
+    def test_resynthesize_success(self):
+        job = make_job(input_text="Retry me")
+        job.mark_processing()
+        repo = FakeTTSJobRepository(job=job)
+        cosy = FakeCosyVoiceService(
+            submit_result={
+                "audio_url": "https://temp.example.com/retry.mp3",
+                "request_id": "req-r",
+                "task_id": "",
+                "duration": 2.5,
+                "file_size": 1500,
+            }
+        )
+        storage = FakeStorageService(upload_url="https://oss.example.com/retry.mp3")
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)
+
+        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"):
+            # call via poll_and_process_synthesis which uses _resynthesize_and_complete
+            result = svc.poll_and_process_synthesis("job-123")
+
+        assert result.status == TTSJobStatus.COMPLETED.value
+        assert len(cosy.submit_calls) == 1
+
+    def test_resynthesize_failure_marks_failed(self):
+        job = make_job()
+        job.mark_processing()
+        repo = FakeTTSJobRepository(job=job)
+        cosy = FakeCosyVoiceService(submit_error=CosyVoiceError("permanent failure"))
+        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)
+
+        result = svc.poll_and_process_synthesis("job-123")
+
+        assert result.status == TTSJobStatus.FAILED.value
+        assert "permanent failure" in result.error_message
+
+
+# ── Error classes tests ─────────────────────────────────
+
+
+class TestErrorClasses:
+    def test_workflow_error_inherits_from_exception(self):
+        err = TTSWorkflowError("test error")
+        assert isinstance(err, Exception)
+        assert str(err) == "test error"
+
+    def test_not_found_error_inherits_from_exception(self):
+        err = TTSJobNotFoundError("not found")
+        assert isinstance(err, Exception)
+        assert "not found" in str(err)
+
+    def test_storage_property_lazy_init(self):
+        """_storage property lazily initializes storage service."""
+        from packages.shared.storage import SharedStorageService
+
+        # With provided storage_service
+        storage = FakeStorageService()
+        svc = TTSWorkflowService(repository=MagicMock(), cosyvoice_service=MagicMock(), storage_service=storage)
+        assert svc._storage is storage