feat(tts): 正式合成支持克隆音色、保存到配音库改写 assets 素材体系、克隆接口支持 asset_id

1. TTS synthesize() 复用 preview 的克隆音色解析逻辑：voice_id 传克隆 profile UUID 时
   校验归属(403)/克隆完成状态(400 音色克隆尚未完成)，job.voice_id 存解析后的
   CosyVoice voice_id，voice_clone_profile_id 记录 profile id。
2. save_tts_job_to_library 改写为新素材体系：TTS 音频持久化独立副本到
   uploads/voice/tts/，创建 file_type=audio/status=ready 的 asset 挂用户 voice
   素材库（无则自动创建），duration/metadata(source=tts_job) 完整记录；
   返回结构保持 id/name/audio_url/duration 兼容前端；移除旧 voice_library 表写入。
3. voice clone 创建接口支持 asset_id 字段：校验素材归属与音频类型，用 asset
   storage_key 生成 7 天有效签名 URL 供 worker 下载，不依赖前端签名；
   source_audio_url 方式保留兼容。

补集成测试：synthesize 克隆音色 4 用例、save-to-library asset 断言、
clone from asset 5 用例（成功/404/403/非音频400/两参同传400）。

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

@@ -7,13 +7,15 @@ from typing import Optional
 
 from app.auth import AuthenticatedUser, get_current_user
 from app.core.celery_app import celery_app
+from app.core.storage import get_storage_service
 from app.dependencies import (
+    get_asset_library_repository,
+    get_asset_repository,
     get_audio_url_signer,
     get_cosyvoice_service,
     get_db_session,
-    get_user_repository,
+    get_project_repository,
     get_voice_clone_profile_repository,
-    get_voice_library_repository,
 )
 from app.schemas.tts import (
     ListTTSJobResponse,
@@ -32,7 +34,6 @@ from sqlalchemy.orm import Session
 from packages.adapters.sqlalchemy_impl.tts_job_repository import (
     SQLAlchemyTTSJobRepository,
 )
-from packages.adapters.sqlalchemy_impl.voice_library_repository import SQLAlchemyVoiceLibraryRepository
 from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService
 from packages.application.tts_job.streaming_service import TTSStreamingService
 from packages.application.tts_job.use_cases import (
@@ -44,13 +45,12 @@ from packages.application.tts_job.use_cases import (
     TTSJobNotFoundError,
 )
 from packages.application.tts_job.workflow import TTSWorkflowService
-from packages.application.voice_library.commands import CreateVoiceLibraryCommand
-from packages.application.voice_library.use_cases import (
-    CreateVoiceLibraryUseCase,
-    QuotaExceededError,
-)
+from packages.domain import Asset, AssetLibrary, AssetLibraryKind, AssetStatus, ClassificationStatus
 from packages.domain.voice_presets import list_voices
-from packages.ports.user_repository import UserRepository
+from packages.ports.asset_library_repository import AssetLibraryRepository
+from packages.ports.asset_repository import AssetRepository
+from packages.ports.project_repository import ProjectRepository
+from packages.shared.storage import SharedStorageService
 
 logger = logging.getLogger(__name__)
 
@@ -134,27 +134,47 @@ def synthesize(
     """
     user_id = authenticated_user.user.id
 
-    # 校验 voice_clone_profile_id 归属（防止越权使用他人克隆音色）
-    if request.voice_clone_profile_id:
-        profile = voice_clone_repo.get(request.voice_clone_profile_id)
-        if profile is None:
+    # 解析 voice_id：前端可能传克隆音色 profile UUID（而非 CosyVoice voice_id），
+    # 与 /tts/preview 保持一致：命中 profile → 校验归属 → 取 CosyVoice voice_id
+    actual_voice_id = request.voice_id
+    voice_clone_profile_id = request.voice_clone_profile_id
+    resolved_profile = None
+    if actual_voice_id:
+        resolved_profile = voice_clone_repo.get(actual_voice_id)
+        if resolved_profile is not None:
+            voice_clone_profile_id = actual_voice_id
+
+    # 显式传了 voice_clone_profile_id（且与 voice_id 不同）时再查一次归属
+    if voice_clone_profile_id and (resolved_profile is None or resolved_profile.id != voice_clone_profile_id):
+        resolved_profile = voice_clone_repo.get(voice_clone_profile_id)
+        if resolved_profile is None:
             raise HTTPException(
                 status_code=status.HTTP_404_NOT_FOUND,
                 detail="Voice clone profile not found",
             )
-        if profile.user_id != user_id:
+
+    if resolved_profile is not None:
+        if resolved_profile.user_id != user_id:
             raise HTTPException(
                 status_code=status.HTTP_403_FORBIDDEN,
-                detail="Access denied to voice clone profile",
+                detail="无权访问该音色",
             )
+        if not resolved_profile.voice_id:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="音色克隆尚未完成，请稍后再试",
+            )
+        # 命中克隆音色：job 存解析后的 CosyVoice voice_id
+        if voice_clone_profile_id == actual_voice_id:
+            actual_voice_id = resolved_profile.voice_id
 
     use_case = CreateTTSJobUseCase(repository)
     job = use_case.execute(
         user_id=user_id,
         input_text=request.text,
-        voice_id=request.voice_id,
+        voice_id=actual_voice_id,
         voice_model=request.voice_model,
-        voice_clone_profile_id=request.voice_clone_profile_id,
+        voice_clone_profile_id=voice_clone_profile_id,
         metadata=request.metadata_,
     )
 
@@ -284,6 +304,41 @@ def delete_tts_job(
     return
 
 
+def _find_or_create_voice_library(
+    *,
+    user_id: str,
+    project_repository: ProjectRepository,
+    asset_library_repository: AssetLibraryRepository,
+) -> AssetLibrary:
+    """在用户可访问的项目中找到（或自动创建）voice 素材库。
+
+    与前端配音素材页逻辑一致：素材库挂在项目下，配音素材读取
+    getAssetsByKind("voice") → 用户所有可访问项目中的 voice 库。
+    优先使用已有 voice 库；没有则在第一个可访问项目中自动创建。
+    """
+    projects = project_repository.find_accessible_projects(user_id)
+    if not projects:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="没有可用的项目，请先创建项目后再保存配音素材",
+        )
+
+    for project in projects:
+        for lib in asset_library_repository.find_by_project(project.id):
+            kind = lib.kind.value if hasattr(lib.kind, "value") else lib.kind
+            if kind == AssetLibraryKind.VOICE.value:
+                return lib
+
+    # 所有项目都没有 voice 库 → 在第一个可访问项目中自动创建默认配音素材库
+    project = projects[0]
+    library = AssetLibrary.create(
+        project_id=project.id,
+        name="配音素材库",
+        kind=AssetLibraryKind.VOICE,
+    )
+    return asset_library_repository.create(library)
+
+
 @router.post(
     "/jobs/{job_id}/save-to-library",
     response_model=SaveToLibraryResponse,
@@ -294,13 +349,17 @@ def save_tts_job_to_library(
     request: SaveToLibraryRequest = SaveToLibraryRequest(),
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     tts_repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
-    voice_library_repository: SQLAlchemyVoiceLibraryRepository = Depends(get_voice_library_repository),
-    user_repository: UserRepository = Depends(get_user_repository),
+    asset_repository: AssetRepository = Depends(get_asset_repository),
+    asset_library_repository: AssetLibraryRepository = Depends(get_asset_library_repository),
+    project_repository: ProjectRepository = Depends(get_project_repository),
+    storage_service: SharedStorageService = Depends(get_storage_service),
     sign_url=Depends(get_audio_url_signer),
 ) -> SaveToLibraryResponse:
-    """将已完成的 TTS 合成结果保存到配音库。
+    """将已完成的 TTS 合成结果保存到配音素材库（assets 表新素材体系）。
 
-    自动携带音色名、时长、语速等元信息。
+    流程：把 TTS 输出音频转存到用户素材 OSS 路径 → 创建 file_type=audio、
+    status=ready 的 asset（挂用户 voice 素材库）→ 返回前端可用结构。
+    配额策略与素材上传一致（上传/ingest 链路无额外配额拦截）。
     """
     user_id = authenticated_user.user.id
 
@@ -318,64 +377,107 @@ def save_tts_job_to_library(
             detail="TTS job is not completed yet",
         )
 
-    # 构建配音素材名称
+    if not job.output_audio_url and not job.output_audio_key:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="TTS job 缺少输出音频，无法保存",
+        )
+
+    # 素材名称
     name = request.name or f"TTS-{job.id[:8]}"
 
-    # 构建元信息
-    metadata_ = {
+    # 找到（或自动创建）用户 voice 素材库
+    library = _find_or_create_voice_library(
+        user_id=user_id,
+        project_repository=project_repository,
+        asset_library_repository=asset_library_repository,
+    )
+
+    # 转存音频到素材 OSS 路径（tts-outputs/ 下的产物归 TTS 任务所有，
+    # 素材独立持有副本，删除 TTS 任务不影响配音库素材）
+    import tempfile
+    from pathlib import Path
+
+    audio_format = (job.format or "mp3").strip() or "mp3"
+    content_type_map = {
+        "mp3": "audio/mpeg",
+        "wav": "audio/wav",
+        "pcm": "audio/pcm",
+        "opus": "audio/opus",
+    }
+    content_type = content_type_map.get(audio_format, "audio/mpeg")
+    storage_key = f"uploads/voice/tts/{job.id}.{audio_format}"
+
+    tmp_path: Path | None = None
+    try:
+        with tempfile.NamedTemporaryFile(suffix=f".{audio_format}", delete=False) as tmp:
+            tmp_path = Path(tmp.name)
+        # 优先用 OSS storage_key（走 oss2 SDK，私有 bucket 也可下载）；
+        # 兜底用 output_audio_url（旧任务可能没有 key）
+        download_source = job.output_audio_key or job.output_audio_url
+        downloaded = storage_service.download_asset(download_source, tmp_path)
+        if not downloaded or not tmp_path.exists() or tmp_path.stat().st_size == 0:
+            raise HTTPException(
+                status_code=status.HTTP_502_BAD_GATEWAY,
+                detail="TTS 音频下载失败，无法保存到配音库",
+            )
+        file_size = tmp_path.stat().st_size
+        storage_service.upload_file(tmp_path, storage_key, content_type=content_type)
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error("TTS 音频转存素材失败: job_id=%s, error=%s", job.id, e, exc_info=True)
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail="TTS 音频转存失败，无法保存到配音库",
+        ) from e
+    finally:
+        if tmp_path and tmp_path.exists():
+            try:
+                tmp_path.unlink()
+            except OSError:
+                pass
+
+    # 构建素材元信息
+    metadata_: dict[str, object] = {
         "source": "tts_job",
         "tts_job_id": job.id,
         "format": job.format,
         "sample_rate": job.sample_rate,
+        "voice_id": job.voice_id,
+        "voice_name": job.voice_model or "",
     }
     if job.metadata:
-        # 保留原始 job 的有用元信息
         for key in ("speed", "language"):
             if key in job.metadata:
                 metadata_[key] = job.metadata[key]
 
-    # 获取用户套餐（用于配额检查）
-    user = user_repository.find_by_id(user_id)
-    plan_name = getattr(user, "subscription_plan", "free") if user else "free"
-
-    # 构建命令并执行
-    command = CreateVoiceLibraryCommand(
-        user_id=user_id,
+    asset = Asset.create(
+        project_id=library.project_id,
+        library_id=library.id,
         name=name,
-        text=job.input_text,
-        voice_provider="cosyvoice",
-        voice_id=job.voice_id,
-        voice_name=job.voice_model or "",
-        audio_url=job.output_audio_url,
-        duration=job.duration,
-        file_size=job.file_size,
-        status="completed",
-        project_id=job.project_id or "",
-        tags=[],
-        metadata_=metadata_,
+        storage_key=storage_key,
+        mime_type=content_type,
+        metadata=metadata_,
+        file_size=file_size,
+        duration=job.duration or None,
+        status=AssetStatus.READY,
+        classification_status=ClassificationStatus.PENDING,  # 音频不参与内容分类，保持 pending 与 ingest 链路一致
+        uploaded_by_user_id=user_id,
     )
-
-    use_case = CreateVoiceLibraryUseCase(voice_library_repository)
-    try:
-        item = use_case.execute(command, plan_name=plan_name or "free")
-    except QuotaExceededError as exc:
-        raise HTTPException(
-            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
-            detail=f"配音库配额已满（{exc.used}/{exc.limit}），请升级套餐",
-        ) from exc
+    asset = asset_repository.create(asset)
 
     return SaveToLibraryResponse(
-        id=item.id,
-        name=item.name,
-        audio_url=sign_url(item.audio_url) if item.audio_url else "",
-        duration=item.duration,
-        voice_id=item.voice_id,
-        voice_name=item.voice_name,
-        status=item.status,
+        id=asset.id,
+        name=asset.name,
+        audio_url=sign_url(storage_key),
+        duration=asset.duration or 0.0,
+        voice_id=job.voice_id,
+        voice_name=job.voice_model or "",
+        status="completed",
     )
 
 
-
 @router.post("/preview", response_model=TTSPreviewResponse)
 def preview_tts(
     request: TTSPreviewRequest,

apps/api/app/api/routes/voice_clones.py

@@ -7,7 +7,13 @@ from typing import Optional
 
 from app.auth import AuthenticatedUser, get_current_user
 from app.core.celery_app import celery_app
-from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repository
+from app.core.storage import get_storage_service
+from app.dependencies import (
+    get_asset_repository,
+    get_cosyvoice_service,
+    get_project_repository,
+    get_voice_clone_profile_repository,
+)
 from app.schemas.voice_clone import (
     CreateVoiceCloneRequest,
     ListVoiceCloneResponse,
@@ -32,6 +38,9 @@ from packages.application.voice_clone.use_cases import (
 from packages.application.voice_clone.workflow import (
     VoiceCloneWorkflowService,
 )
+from packages.ports.asset_repository import AssetRepository
+from packages.ports.project_repository import ProjectRepository
+from packages.shared.storage import SharedStorageService
 
 logger = logging.getLogger(__name__)
 
@@ -83,23 +92,69 @@ def create_voice_clone(
     request: CreateVoiceCloneRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     workflow: VoiceCloneWorkflowService = Depends(_get_workflow_service),
+    asset_repository: AssetRepository = Depends(get_asset_repository),
+    project_repository: ProjectRepository = Depends(get_project_repository),
+    storage_service: SharedStorageService = Depends(get_storage_service),
 ) -> VoiceCloneProfileResponse:
     """创建音色克隆任务。
 
     创建 VoiceCloneProfile → 提交 CosyVoice 克隆任务 → 触发 Celery 异步轮询。
-    如果有 source_audio_url，状态会变为 processing；否则保持 pending。
+    参考音频两种来源（二选一）：
+    - source_audio_url：前端直传后的音频 URL（兼容旧流程）
+    - asset_id：配音素材库中的音频素材，服务端用其 OSS storage_key 生成
+      预签名下载 URL（不依赖前端签名，避免签名过期导致克隆失败）
+    如果有参考音频，状态会变为 processing；否则保持 pending。
     """
     user_id = authenticated_user.user.id
+
+    source_audio_url = request.source_audio_url
+    clone_metadata = dict(request.metadata_ or {})
+
+    if request.asset_id:
+        if source_audio_url:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="asset_id 与 source_audio_url 只能传一个",
+            )
+        asset = asset_repository.find_by_id(request.asset_id)
+        if asset is None:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="素材不存在",
+            )
+        # 归属校验：素材挂在项目素材库下，用户必须能访问该项目
+        project = project_repository.find_by_id(asset.project_id)
+        if project is None or not project.can_access(user_id):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="无权使用该素材",
+            )
+        # 类型校验：仅支持音频素材
+        file_type = asset.file_type if hasattr(asset, "file_type") else ""
+        if file_type != "audio":
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="仅支持音频素材进行音色克隆",
+            )
+        if not asset.storage_key:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="该素材缺少音频文件，无法用于克隆",
+            )
+        # 用 OSS storage_key 生成服务端预签名 URL（7 天有效，覆盖克隆重试周期）
+        source_audio_url = storage_service.get_download_url(asset.storage_key, expires_seconds=7 * 24 * 3600)
+        clone_metadata["source_asset_id"] = asset.id
+
     profile = workflow.start_clone(
         user_id=user_id,
         name=request.name,
         description=request.description,
-        source_audio_url=request.source_audio_url,
+        source_audio_url=source_audio_url,
         voice_model=request.voice_model,
         language=request.language,
         gender=request.gender,
         max_retries=request.max_retries,
-        metadata=request.metadata_,
+        metadata=clone_metadata,
     )
 
     # 如果 profile 处于 processing 且有 task_id，触发 Celery 异步轮询

apps/api/app/schemas/voice_clone.py

@@ -13,7 +13,8 @@ class CreateVoiceCloneRequest(BaseModel):
 
     name: str = Field(..., min_length=1, max_length=100, description="音色名称")
     description: str = Field("", description="音色描述")
-    source_audio_url: str = Field("", description="参考音频 URL")
+    source_audio_url: str = Field("", description="参考音频 URL（与 asset_id 二选一）")
+    asset_id: str = Field("", description="参考音频素材 ID（配音素材库中的音频 asset，与 source_audio_url 二选一）")
     voice_model: str = Field("", description="语音模型名称")
     language: str = Field("zh-CN", description="语言")
     gender: str = Field("unknown", description="性别")

tests/integration/test_tts_api.py

@@ -31,8 +31,12 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "
 
 from app.api.routes.tts import router
 from app.auth import AuthenticatedUser, get_current_user
+from app.core.storage import get_storage_service
 from app.dependencies import (
+    get_asset_library_repository,
+    get_asset_repository,
     get_cosyvoice_service,
+    get_project_repository,
     get_user_repository,
     get_voice_clone_profile_repository,
     get_voice_library_repository,
@@ -322,6 +326,92 @@ def _make_voice_clone_profile(
     return profile
 
 
+# ---------------------------------------------------------------------------
+# 3b. 素材体系内存 Repository（save-to-library 新链路）
+# ---------------------------------------------------------------------------
+
+
+class InMemoryAssetRepository:
+    """内存 Asset 仓储（save-to-library 只用到 create/find_by_id）。"""
+
+    def __init__(self):
+        self._items: dict = {}
+
+    def create(self, asset):
+        self._items[asset.id] = asset
+        return asset
+
+    def find_by_id(self, asset_id: str):
+        return self._items.get(asset_id)
+
+
+class InMemoryAssetLibraryRepository:
+    """内存 AssetLibrary 仓储。"""
+
+    def __init__(self):
+        self._items: dict = {}
+
+    def create(self, library):
+        self._items[library.id] = library
+        return library
+
+    def get(self, library_id: str):
+        return self._items.get(library_id)
+
+    def find_by_id(self, library_id: str):
+        return self._items.get(library_id)
+
+    def find_by_project(self, project_id: str):
+        return [lib for lib in self._items.values() if lib.project_id == project_id]
+
+
+class InMemoryProjectRepository2:
+    """内存 Project 仓储（save-to-library 需要）。"""
+
+    def __init__(self):
+        self._items: dict = {}
+
+    def find_by_id(self, project_id: str):
+        return self._items.get(project_id)
+
+    def find_accessible_projects(self, user_id: str):
+        from packages.domain.entities import Project
+
+        projects = [p for p in self._items.values() if p.can_access(user_id)]
+        if projects:
+            return projects
+        # 没有任何项目时自动给一个默认项目（与前端 getOrCreateDefaultProject 行为对齐）
+        project = Project.create(owner_user_id=user_id, name="默认项目")
+        self._items[project.id] = project
+        return [project]
+
+
+class MockStorageService:
+    """Mock 存储：下载写出小文件，上传不做真实 OSS 操作。"""
+
+    def __init__(self):
+        self.uploaded_keys: list[str] = []
+        self.download_calls: list[str] = []
+
+    def download_asset(self, storage_key_or_url: str, local_path) -> bool:
+        self.download_calls.append(storage_key_or_url)
+        from pathlib import Path
+
+        path = Path(local_path)
+        path.write_bytes(b"FAKE-AUDIO-BYTES" * 100)
+        return True
+
+    def upload_file(self, file_or_path, storage_key: str, content_type: str = "application/octet-stream") -> str:
+        self.uploaded_keys.append(storage_key)
+        return f"https://cdn.example.com/{storage_key}"
+
+    def get_download_url(self, storage_key: str, expires_seconds: int = 3600) -> str:
+        return f"https://cdn.example.com/{storage_key}?signed=1"
+
+    def get_url(self, storage_key: str) -> str:
+        return f"https://cdn.example.com/{storage_key}"
+
+
 # ---------------------------------------------------------------------------
 # 4. Fixtures
 # ---------------------------------------------------------------------------
@@ -342,6 +432,26 @@ def voice_library_repo():
     return InMemoryVoiceLibraryRepository()
 
 
+@pytest.fixture
+def asset_repo():
+    return InMemoryAssetRepository()
+
+
+@pytest.fixture
+def asset_library_repo():
+    return InMemoryAssetLibraryRepository()
+
+
+@pytest.fixture
+def project_repo2():
+    return InMemoryProjectRepository2()
+
+
+@pytest.fixture
+def storage_service():
+    return MockStorageService()
+
+
 @pytest.fixture
 def user_repo():
     repo = InMemoryUserRepository()
@@ -355,7 +465,17 @@ def cosyvoice_service():
 
 
 @pytest.fixture
-def client(tts_repo, voice_clone_repo, voice_library_repo, user_repo, cosyvoice_service):
+def client(
+    tts_repo,
+    voice_clone_repo,
+    voice_library_repo,
+    user_repo,
+    cosyvoice_service,
+    asset_repo,
+    asset_library_repo,
+    project_repo2,
+    storage_service,
+):
     """创建带有依赖覆盖的 TestClient。"""
     test_app = FastAPI()
     test_app.include_router(router, prefix="/tts")
@@ -371,6 +491,10 @@ def client(tts_repo, voice_clone_repo, voice_library_repo, user_repo, cosyvoice_
     test_app.dependency_overrides[get_voice_clone_profile_repository] = lambda: voice_clone_repo
     test_app.dependency_overrides[get_voice_library_repository] = lambda: voice_library_repo
     test_app.dependency_overrides[get_user_repository] = lambda: user_repo
+    test_app.dependency_overrides[get_asset_repository] = lambda: asset_repo
+    test_app.dependency_overrides[get_asset_library_repository] = lambda: asset_library_repo
+    test_app.dependency_overrides[get_project_repository] = lambda: project_repo2
+    test_app.dependency_overrides[get_storage_service] = lambda: storage_service
 
     # 使用 FastAPI dependency_overrides 覆盖 TTS repository
     from app.api.routes import tts as tts_module
@@ -493,6 +617,65 @@ class TestCreateTTSJob:
         )
         assert resp.status_code == 403
 
+    def test_synthesize_with_clone_profile_in_voice_id(self, client, tts_repo, voice_clone_repo):
+        """voice_id 直接传克隆 profile UUID（新前端流程）：
+
+        命中 profile → 校验归属 → job.voice_id 存解析后的 CosyVoice voice_id，
+        voice_clone_profile_id 记录该 profile。
+        """
+        profile = _make_voice_clone_profile()  # ready，voice_id="clone-voice-001"
+        voice_clone_repo.create(profile)
+
+        resp = client.post(
+            "/tts/synthesize",
+            json={"text": "克隆音色合成", "voice_id": profile.id},
+        )
+        assert resp.status_code == 201, resp.text
+        job = tts_repo.get(resp.json()["job_id"])
+        assert job.voice_id == "clone-voice-001"
+        assert job.voice_clone_profile_id == profile.id
+
+    def test_synthesize_with_other_user_clone_voice_id_returns_403(self, client, voice_clone_repo):
+        """voice_id 传他人克隆 profile UUID → 403。"""
+        profile = _make_voice_clone_profile(user_id="other-user")
+        voice_clone_repo.create(profile)
+
+        resp = client.post(
+            "/tts/synthesize",
+            json={"text": "越权克隆音色", "voice_id": profile.id},
+        )
+        assert resp.status_code == 403, resp.text
+
+    def test_synthesize_with_unfinished_clone_voice_id_returns_400(self, client, voice_clone_repo):
+        """voice_id 传克隆未完成（voice_id 为空）的 profile → 400。"""
+        from packages.domain.voice_clone_profile import VoiceCloneProfile
+
+        profile = VoiceCloneProfile.create(
+            user_id="user-test-001",
+            name="未完成克隆",
+            voice_model="cosyvoice-v2",
+        )
+        profile.mark_processing()  # processing 状态，尚未 mark_ready，voice_id 为空
+        voice_clone_repo.create(profile)
+
+        resp = client.post(
+            "/tts/synthesize",
+            json={"text": "未完成克隆", "voice_id": profile.id},
+        )
+        assert resp.status_code == 400, resp.text
+        assert "克隆尚未完成" in resp.json()["detail"]
+
+    def test_synthesize_with_preset_voice_id_unaffected(self, client, tts_repo):
+        """预置音色 voice_id 不匹配任何 profile 时走原流程，不受影响。"""
+        resp = client.post(
+            "/tts/synthesize",
+            json={"text": "预置音色", "voice_id": "longxiaochun_v2"},
+        )
+        assert resp.status_code == 201, resp.text
+        job = tts_repo.get(resp.json()["job_id"])
+        assert job.voice_id == "longxiaochun_v2"
+        assert job.voice_clone_profile_id == ""
+
 
 # ---------------------------------------------------------------------------
 # 6. GET /jobs — 列出 TTS 任务
@@ -805,18 +988,30 @@ class TestSaveToLibrary:
         # 自动生成的名称应该以 TTS- 开头
         assert data["name"].startswith("TTS-")
 
-    def test_save_creates_library_item(self, client, tts_repo, voice_library_repo):
-        """保存后配音库中新增一条记录。"""
-        before_count = voice_library_repo.count_by_user("user-test-001")
-
-        job = _make_tts_job("入库测试", status=TTSJobStatus.COMPLETED)
+    def test_save_creates_library_item(self, client, tts_repo, asset_repo, asset_library_repo):
+        """保存后在素材体系（assets 表）中新增一条 ready 音频素材，并自动创建 voice 库。"""
+        job = _make_tts_job("入库测试", status=TTSJobStatus.COMPLETED, duration=6.0)
         tts_repo.create(job)
 
         resp = client.post(f"/tts/jobs/{job.id}/save-to-library", json={"name": "入库"})
-        assert resp.status_code == 201
+        assert resp.status_code == 201, resp.text
+        data = resp.json()
 
-        after_count = voice_library_repo.count_by_user("user-test-001")
-        assert after_count == before_count + 1
+        # asset 已创建
+        asset = asset_repo.find_by_id(data["id"])
+        assert asset is not None
+        assert asset.status.value == "ready"
+        assert asset.file_type == "audio"
+        assert asset.duration == 6.0
+        assert (asset.metadata or {}).get("source") == "tts_job"
+        assert (asset.metadata or {}).get("tts_job_id") == job.id
+        assert asset.uploaded_by_user_id == "user-test-001"
+        assert asset.storage_key.startswith("uploads/voice/tts/")
+
+        # voice 素材库自动创建，asset 挂到该库
+        assert asset.library_id in asset_library_repo._items
+        voice_lib = asset_library_repo.get(asset.library_id)
+        assert voice_lib is not None
 
 
 # ---------------------------------------------------------------------------

tests/integration/test_voice_clone_api.py

@@ -29,15 +29,20 @@ from fastapi.testclient import TestClient
 
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
 
+from unittest.mock import MagicMock
+
 from app.api.routes.voice_clones import router
 from app.auth import AuthenticatedUser, get_current_user
+from app.core.storage import get_storage_service
 from app.dependencies import (
+    get_asset_repository,
     get_audio_url_signer,
     get_cosyvoice_service,
+    get_project_repository,
     get_voice_clone_profile_repository,
 )
 
-from packages.domain.entities import User
+from packages.domain.entities import Asset, AssetStatus, Project, User
 from packages.domain.voice_clone_profile import (
     VoiceCloneProfile,
     VoiceCloneStatus,
@@ -232,8 +237,78 @@ def cosyvoice_service():
     return MockCosyVoiceService(async_mode=True)  # 异步模式，匹配真实 CosyVoice API 行为
 
 
+def _make_voice_asset(
+    *,
+    asset_id: str = "asset-voice-1",
+    project_id: str = "proj-1",
+    user_id: str = "user-test-001",
+    file_type: str = "audio",
+    storage_key: str = "uploads/voice/sample.m4a",
+) -> Asset:
+    """构造一个音频素材（可改 file_type 模拟非音频）。"""
+    mime = {"audio": "audio/mp4", "video": "video/mp4", "image": "image/jpeg"}[file_type]
+    return Asset(
+        id=asset_id,
+        project_id=project_id,
+        library_id="lib-1",
+        name="配音素材.m4a",
+        storage_key=storage_key,
+        mime_type=mime,
+        file_size=12345,
+        duration=10.0,
+        status=AssetStatus.READY,
+        uploaded_by_user_id=user_id,
+    )
+
+
+class InMemoryAssetRepository:
+    """最小内存 Asset 仓储（find_by_id 即可）。"""
+
+    def __init__(self, assets: list[Asset] | None = None):
+        self._items = {a.id: a for a in (assets or [])}
+
+    def find_by_id(self, asset_id: str):
+        return self._items.get(asset_id)
+
+
+class InMemoryProjectRepository:
+    """最小内存 Project 仓储（find_by_id 即可）。"""
+
+    def __init__(self, projects: list[Project] | None = None):
+        self._items = {p.id: p for p in (projects or [])}
+
+    def find_by_id(self, project_id: str):
+        return self._items.get(project_id)
+
+
+class MockStorageService:
+    """Mock 存储服务：get_download_url 返回固定签名 URL。"""
+
+    def __init__(self):
+        self.signed_keys: list[str] = []
+
+    def get_download_url(self, storage_key: str, expires_seconds: int = 3600) -> str:
+        self.signed_keys.append(storage_key)
+        return f"https://oss.example.com/{storage_key}?signed=1&exp={expires_seconds}"
+
+
 @pytest.fixture
-def client(clone_repo, cosyvoice_service):
+def asset_repo():
+    return InMemoryAssetRepository()
+
+
+@pytest.fixture
+def project_repo():
+    return InMemoryProjectRepository()
+
+
+@pytest.fixture
+def storage_service():
+    return MockStorageService()
+
+
+@pytest.fixture
+def client(clone_repo, cosyvoice_service, asset_repo, project_repo, storage_service):
     """创建带有依赖覆盖的 TestClient。"""
     test_app = FastAPI()
     test_app.include_router(router, prefix="/voice-clones")
@@ -245,6 +320,9 @@ def client(clone_repo, cosyvoice_service):
     test_app.dependency_overrides[get_voice_clone_profile_repository] = lambda: clone_repo
     test_app.dependency_overrides[get_cosyvoice_service] = lambda: cosyvoice_service
     test_app.dependency_overrides[get_audio_url_signer] = lambda: (lambda url: url)
+    test_app.dependency_overrides[get_asset_repository] = lambda: asset_repo
+    test_app.dependency_overrides[get_project_repository] = lambda: project_repo
+    test_app.dependency_overrides[get_storage_service] = lambda: storage_service
 
     yield TestClient(test_app)
 
@@ -714,3 +792,77 @@ class TestVoiceCloneLifecycle:
 
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])
+
+
+# ---------------------------------------------------------------------------
+# 6. POST /?asset_id=xxx — 从配音素材创建克隆
+# ---------------------------------------------------------------------------
+
+
+class TestCreateVoiceCloneFromAsset:
+    """asset_id 参数测试：用配音素材库音频发起克隆。"""
+
+    def test_create_from_asset_success(self, client, clone_repo, asset_repo, project_repo, storage_service):
+        """正常路径：asset 归属本人且为音频 → 201，source_audio_url 为签名 URL。"""
+        project = Project.create(owner_user_id="user-test-001", name="默认项目")
+        project_repo._items[project.id] = project
+        asset = _make_voice_asset(asset_id="asset-ok", project_id=project.id)
+        asset_repo._items[asset.id] = asset
+
+        resp = client.post(
+            "/voice-clones",
+            json={"name": "素材克隆", "asset_id": asset.id},
+        )
+        assert resp.status_code == 201, resp.text
+        data = resp.json()
+        assert data["name"] == "素材克隆"
+        # 用 OSS storage_key 生成了服务端签名 URL
+        assert any("uploads/voice/sample.m4a" in k for k in storage_service.signed_keys)
+        assert data["source_audio_url"].startswith("https://oss.example.com/uploads/voice/sample.m4a")
+
+    def test_create_from_asset_not_found_returns_404(self, client, asset_repo, project_repo):
+        """asset 不存在 → 404。"""
+        resp = client.post("/voice-clones", json={"name": "克隆", "asset_id": "no-such-asset"})
+        assert resp.status_code == 404, resp.text
+
+    def test_create_from_asset_other_user_returns_403(self, client, asset_repo, project_repo):
+        """asset 属于他人项目 → 403。"""
+        project = Project.create(owner_user_id="other-user", name="他人项目")
+        project_repo._items[project.id] = project
+        asset = _make_voice_asset(asset_id="asset-other", project_id=project.id, user_id="other-user")
+        asset_repo._items[asset.id] = asset
+
+        resp = client.post("/voice-clones", json={"name": "克隆", "asset_id": asset.id})
+        assert resp.status_code == 403, resp.text
+
+    def test_create_from_non_audio_asset_returns_400(self, client, asset_repo, project_repo):
+        """asset 是视频 → 400。"""
+        project = Project.create(owner_user_id="user-test-001", name="默认项目")
+        project_repo._items[project.id] = project
+        asset = _make_voice_asset(
+            asset_id="asset-video",
+            project_id=project.id,
+            file_type="video",
+            storage_key="uploads/v/test.mp4",
+        )
+        asset_repo._items[asset.id] = asset
+
+        resp = client.post("/voice-clones", json={"name": "克隆", "asset_id": asset.id})
+        assert resp.status_code == 400, resp.text
+
+    def test_create_with_both_asset_id_and_url_returns_400(self, client, asset_repo, project_repo):
+        """asset_id 与 source_audio_url 同时传 → 400。"""
+        project = Project.create(owner_user_id="user-test-001", name="默认项目")
+        project_repo._items[project.id] = project
+        asset = _make_voice_asset(asset_id="asset-both", project_id=project.id)
+        asset_repo._items[asset.id] = asset
+
+        resp = client.post(
+            "/voice-clones",
+            json={
+                "name": "克隆",
+                "asset_id": asset.id,
+                "source_audio_url": "https://example.com/a.wav",
+            },
+        )
+        assert resp.status_code == 400, resp.text