diff --git a/alembic/versions/058_uq_asset_lib_project_kind.py b/alembic/versions/058_uq_asset_lib_project_kind.py
new file mode 100644
index 000000000..2519970e6
--- /dev/null
+++ b/alembic/versions/058_uq_asset_lib_project_kind.py
@@ -0,0 +1,49 @@
+"""Add unique index on asset_libraries(project_id, kind)
+
+Revision ID: 058_uq_asset_lib_project_kind
+Revises: 057_title_config
+Create Date: 2026-08-30
+
+同一项目下同 kind 的素材库业务上唯一（前端 getOrCreate 语义、TTS 保存自动建库）。
+加唯一索引兜底并发创建竞态，避免重复素材库。
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "058_uq_asset_lib_project_kind"
+down_revision = "057_title_config"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # 建唯一索引前清洗历史重复：同 (project_id, kind) 只保留 created_at 最新的一条。
+    # project_id 为 NULL 的系统级行不参与去重（NULL 在唯一索引中互不冲突）。
+    op.execute("""
+        DELETE FROM asset_libraries
+        WHERE id IN (
+            SELECT id FROM (
+                SELECT id,
+                       ROW_NUMBER() OVER (
+                           PARTITION BY project_id, kind
+                           ORDER BY created_at DESC, id DESC
+                       ) AS rn
+                FROM asset_libraries
+                WHERE project_id IS NOT NULL
+            ) t
+            WHERE t.rn > 1
+        )
+        """)
+    # 与 model 的 UniqueConstraint 定义保持一致（pg_constraint + pg_index 同时注册），
+    # 避免 Alembic autogenerate 检测到 schema drift
+    op.create_unique_constraint(
+        "uq_asset_libraries_project_kind",
+        "asset_libraries",
+        ["project_id", "kind"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint("uq_asset_libraries_project_kind", "asset_libraries", type_="unique")
diff --git a/apps/api/app/api/routes/tts.py b/apps/api/app/api/routes/tts.py
index eb9581e07..f1e06dc0f 100755
--- a/apps/api/app/api/routes/tts.py
+++ b/apps/api/app/api/routes/tts.py
@@ -3,17 +3,21 @@
 from __future__ import annotations
 
 import logging
-from typing import Optional
+import tempfile
+from pathlib import Path
+from typing import Any, 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,
@@ -27,12 +31,12 @@ from app.schemas.tts import (
     TTSSynthesizeResponse,
 )
 from fastapi import APIRouter, Depends, HTTPException, Query, Response, WebSocket, WebSocketDisconnect, status
+from sqlalchemy.exc import IntegrityError
 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 +48,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 +137,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="音色克隆尚未完成，请稍后再试",
+            )
+        # 命中克隆音色：无论 voice_id 直接传 profile UUID 还是显式传 voice_clone_profile_id，
+        # job.voice_id 统一存解析后的 CosyVoice 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 +307,62 @@ def delete_tts_job(
     return
 
 
+def _find_or_create_voice_library(
+    *,
+    user_id: str,
+    project_repository: ProjectRepository,
+    asset_library_repository: Any,  # port Protocol 声明为 async，SQLAlchemy 实现为同步，与 upload/asset_libraries 路由惯例一致用 Any
+) -> 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 库 → 在第一个可访问项目中自动创建默认配音素材库。
+    # asset_libraries 有 (project_id, kind) 唯一索引兜底并发：若两个请求同时创建，
+    # 落败方捕获 IntegrityError 回滚后重新查询，返回抢先创建成功的库。
+    project = projects[0]
+    library = AssetLibrary.create(
+        project_id=project.id,
+        name="配音素材库",
+        kind=AssetLibraryKind.VOICE,
+    )
+    try:
+        return asset_library_repository.create(library)
+    except IntegrityError:
+        # 并发下另一个请求已抢先创建：回滚当前事务（立即 commit 模式下 session 已
+        # 自动回滚，rollback 为幂等 no-op；UoW/flush 模式下必须显式回滚才能继续查询），
+        # 再重查返回抢先创建成功的库。
+        session = getattr(asset_library_repository, "session", None)
+        if session is not None:
+            try:
+                session.rollback()
+            except Exception:
+                logger.warning("IntegrityError 后回滚 session 失败（可能已关闭）", exc_info=True)
+        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
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="配音素材库创建失败，请重试",
+        ) from None  # IntegrityError 已处理，不保留异常链
+
+
 @router.post(
     "/jobs/{job_id}/save-to-library",
     response_model=SaveToLibraryResponse,
@@ -294,13 +373,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 +401,117 @@ 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 任务不影响配音库素材）
+    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_asset 自动识别输入：http(s):// 开头走 HTTP 下载，否则按 OSS key 走 SDK。
+        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:
+        asset = asset_repository.create(asset)
+    except Exception as e:
+        # DB 写入失败：清理已上传到 OSS 的素材文件，避免产生无法索引的孤儿文件
+        logger.error("素材记录创建失败，清理 OSS 文件: %s, error=%s", storage_key, e, exc_info=True)
+        try:
+            storage_service.delete_file(storage_key)
+        except Exception:
+            logger.warning("清理孤儿 OSS 文件失败: %s", storage_key, exc_info=True)
         raise HTTPException(
-            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
-            detail=f"配音库配额已满（{exc.used}/{exc.limit}），请升级套餐",
-        ) from exc
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail="素材保存失败，请重试",
+        ) from e
 
     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,
diff --git a/apps/api/app/api/routes/voice_clones.py b/apps/api/app/api/routes/voice_clones.py
index cc513b615..39875379e 100755
--- a/apps/api/app/api/routes/voice_clones.py
+++ b/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,68 @@ 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="无权使用该素材",
+            )
+        # 类型校验：仅支持音频素材
+        if asset.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 异步轮询
diff --git a/apps/api/app/schemas/voice_clone.py b/apps/api/app/schemas/voice_clone.py
index b13517366..261fbfaf2 100755
--- a/apps/api/app/schemas/voice_clone.py
+++ b/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="性别")
diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py
index 8e0cdbd55..df5cf79f3 100755
--- a/packages/adapters/sqlalchemy_impl/models.py
+++ b/packages/adapters/sqlalchemy_impl/models.py
@@ -55,6 +55,7 @@ class ProjectModel(Base):
 
 class AssetLibraryModel(Base):
     __tablename__ = "asset_libraries"
+    __table_args__ = (UniqueConstraint("project_id", "kind", name="uq_asset_libraries_project_kind"),)
 
     id = Column(String(36), primary_key=True)
     project_id = Column(String(36), nullable=True, index=True)
diff --git a/tests/integration/test_tts_api.py b/tests/integration/test_tts_api.py
index 63e065b0c..7c0acb081 100644
--- a/tests/integration/test_tts_api.py
+++ b/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
@@ -453,10 +577,14 @@ class TestCreateTTSJob:
         )
         assert resp.status_code == 201
 
-    def test_create_with_voice_clone_profile_id(self, client, voice_clone_repo):
-        """使用音色克隆档案创建 TTS。"""
-        # 准备一个克隆档案
-        profile = _make_voice_clone_profile()
+    def test_create_with_voice_clone_profile_id(self, client, tts_repo, voice_clone_repo):
+        """显式传 voice_clone_profile_id 创建 TTS：
+
+        - 仅传 profile id（voice_id 为空）→ job.voice_id 解析为克隆 CosyVoice id
+        - voice_id 传预置音色 + 显式 profile id（两者不一致）→ 仍以克隆 profile 为准，
+          job.voice_id 必须是克隆 CosyVoice id，不能保留预置音色 id
+        """
+        profile = _make_voice_clone_profile()  # ready，voice_id="clone-voice-001"
         voice_clone_repo.create(profile)
 
         resp = client.post(
@@ -467,6 +595,23 @@ class TestCreateTTSJob:
             },
         )
         assert resp.status_code == 201
+        job = tts_repo.get(resp.json()["job_id"])
+        assert job.voice_id == "clone-voice-001"
+        assert job.voice_clone_profile_id == profile.id
+
+        # voice_id 传预置音色 + 显式克隆 profile：合成音色必须以克隆 profile 为准
+        resp2 = client.post(
+            "/tts/synthesize",
+            json={
+                "text": "预置voice_id加克隆profile",
+                "voice_id": "longxiaochun_v2",
+                "voice_clone_profile_id": profile.id,
+            },
+        )
+        assert resp2.status_code == 201, resp2.text
+        job2 = tts_repo.get(resp2.json()["job_id"])
+        assert job2.voice_id == "clone-voice-001"
+        assert job2.voice_clone_profile_id == profile.id
 
     def test_create_with_nonexistent_clone_profile_returns_404(self, client):
         """使用不存在的克隆档案返回 404。"""
@@ -493,6 +638,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 +1009,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
 
 
 # ---------------------------------------------------------------------------
diff --git a/tests/integration/test_voice_clone_api.py b/tests/integration/test_voice_clone_api.py
index a63d402c1..def454b87 100755
--- a/tests/integration/test_voice_clone_api.py
+++ b/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
diff --git a/tests/unit/test_tts_save_to_library_assets.py b/tests/unit/test_tts_save_to_library_assets.py
new file mode 100644
index 000000000..92544dc31
--- /dev/null
+++ b/tests/unit/test_tts_save_to_library_assets.py
@@ -0,0 +1,313 @@
+"""save_tts_job_to_library 改写为 assets 素材体系的单元测试（路由级）。
+
+覆盖：TTS 音频转存素材 OSS → 创建 audio/ready asset → 返回结构；
+voice 素材库查找/自动创建/并发竞态兜底；未完成/无音频/下载失败等分支。
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from unittest.mock import MagicMock
+
+from packages.domain import AssetLibrary, AssetLibraryKind
+from packages.domain.tts_job import TTSJob, TTSJobStatus
+
+
+def _completed_job(**kwargs) -> TTSJob:
+    defaults = {
+        "id": "job-save-001",
+        "user_id": "user-1",
+        "input_text": "测试",
+        "voice_id": "longxiaochun_v2",
+        "voice_model": "",
+        "project_id": "",
+        "voice_clone_profile_id": "",
+        "status": TTSJobStatus.COMPLETED,
+        "output_audio_url": "https://oss.example.com/tts-outputs/u/job.mp3",
+        "output_audio_key": "tts-outputs/user-1/job-save-001.mp3",
+        "duration": 6.0,
+        "file_size": 12345,
+        "sample_rate": 22050,
+        "format": "mp3",
+        "error_message": "",
+        "retry_count": 0,
+        "max_retries": 3,
+        "metadata": {"speed": 1.0},
+        "started_at": None,
+        "completed_at": None,
+        "created_at": datetime.now(timezone.utc),
+        "updated_at": datetime.now(timezone.utc),
+    }
+    defaults.update(kwargs)
+    return TTSJob(**defaults)
+
+
+class FakeTTSRepo:
+    def __init__(self, job):
+        self._job = job
+
+    def get(self, job_id, user_id=None):
+        if self._job and self._job.id == job_id and (user_id is None or self._job.user_id == user_id):
+            return self._job
+        return None
+
+
+class FakeAssetRepo:
+    def __init__(self):
+        self.created = []
+
+    def create(self, asset):
+        self.created.append(asset)
+        return asset
+
+
+class FakeProject:
+    def __init__(self, pid="proj-1", owner="user-1"):
+        self.id = pid
+        self.owner_user_id = owner
+
+    def can_access(self, user_id):
+        return user_id == self.owner_user_id
+
+
+class FakeProjectRepo:
+    def __init__(self, projects):
+        self._projects = projects
+
+    def find_accessible_projects(self, user_id):
+        return [p for p in self._projects if p.can_access(user_id)]
+
+    def find_by_id(self, pid):
+        for p in self._projects:
+            if p.id == pid:
+                return p
+        return None
+
+
+class FakeAssetLibraryRepo:
+    def __init__(self, libs=None, fail_integrity=False, rollback_raises=False):
+        self._libs = list(libs or [])
+        self.fail_integrity = fail_integrity
+        self.session = MagicMock()
+        if rollback_raises:
+            self.session.rollback.side_effect = RuntimeError("session already closed")
+
+    def find_by_project(self, project_id):
+        return [lib for lib in self._libs if lib.project_id == project_id]
+
+    def create(self, library):
+        if self.fail_integrity and not any(
+            lib.project_id == library.project_id and lib.kind == library.kind for lib in self._libs
+        ):
+            from sqlalchemy.exc import IntegrityError
+
+            # 模拟并发：另一个请求抢先创建了同名库
+            existing = AssetLibrary.create(
+                project_id=library.project_id, name="配音素材库", kind=AssetLibraryKind.VOICE
+            )
+            self._libs.append(existing)
+            raise IntegrityError("INSERT", {}, Exception("duplicate key"))
+        self._libs.append(library)
+        return library
+
+
+class FakeStorage:
+    def __init__(self, download_ok=True):
+        self.download_ok = download_ok
+        self.uploads = []
+
+    def download_asset(self, source, local_path):
+        if not self.download_ok:
+            return False
+        import os
+
+        with open(local_path, "wb") as f:
+            f.write(b"FAKEAUDIO" * 100)
+        return os.path.exists(local_path) and os.path.getsize(local_path) > 0
+
+    def upload_file(self, local_path, storage_key, content_type=None):
+        self.uploads.append((storage_key, content_type))
+        return storage_key
+
+    def delete_file(self, storage_key):
+        self.deleted = getattr(self, "deleted", [])
+        self.deleted.append(storage_key)
+
+    def get_download_url(self, key, expires_seconds=3600):
+        return f"https://oss.example.com/signed/{key}?sig=xxx"
+
+
+def _build_app(*, job, libs=None, projects=None, storage=None, lib_fail=False, rollback_raises=False):
+    from fastapi import FastAPI
+
+    app = FastAPI()
+    from app.api.routes.tts import _get_repository, router
+    from app.auth import get_current_user
+    from app.core.storage import get_storage_service
+    from app.dependencies import (
+        get_asset_library_repository,
+        get_asset_repository,
+        get_audio_url_signer,
+        get_project_repository,
+    )
+
+    app.include_router(router, prefix="/tts")
+
+    fake_user = MagicMock()
+    fake_user.user.id = "user-1"
+    app.dependency_overrides[get_current_user] = lambda: fake_user
+    app.dependency_overrides[_get_repository] = lambda: FakeTTSRepo(job)
+
+    asset_repo = FakeAssetRepo()
+    app.dependency_overrides[get_asset_repository] = lambda: asset_repo
+    app.dependency_overrides[get_asset_library_repository] = lambda: FakeAssetLibraryRepo(
+        libs, lib_fail, rollback_raises=rollback_raises
+    )
+    app.dependency_overrides[get_project_repository] = lambda: FakeProjectRepo(
+        projects if projects is not None else [FakeProject()]
+    )
+    storage = storage or FakeStorage()
+    app.dependency_overrides[get_storage_service] = lambda: storage
+    app.dependency_overrides[get_audio_url_signer] = lambda: (lambda key: f"https://signed/{key}")
+    return app, asset_repo, storage
+
+
+class TestSaveToLibraryAssets:
+    def test_save_creates_audio_asset_in_voice_library(self):
+        """保存成功：转存音频 + 创建 ready/audio asset，返回结构完整。"""
+        from fastapi.testclient import TestClient
+
+        voice_lib = AssetLibrary.create(project_id="proj-1", name="配音素材库", kind=AssetLibraryKind.VOICE)
+        app, asset_repo, storage = _build_app(job=_completed_job(), libs=[voice_lib])
+        client = TestClient(app)
+
+        resp = client.post("/tts/jobs/job-save-001/save-to-library", json={"name": "我的配音"})
+        assert resp.status_code == 201, resp.text
+        data = resp.json()
+        assert data["id"]
+        assert data["name"] == "我的配音"
+        assert data["duration"] == 6.0
+        assert data["voice_id"] == "longxiaochun_v2"
+        assert data["status"] == "completed"
+        assert data["audio_url"].startswith("https://signed/")
+
+        assert len(asset_repo.created) == 1
+        asset = asset_repo.created[0]
+        assert asset.file_type == "audio"
+        assert asset.status.value == "ready"
+        assert asset.duration == 6.0
+        assert asset.storage_key.startswith("uploads/voice/tts/")
+        assert asset.metadata["source"] == "tts_job"
+        assert asset.metadata["tts_job_id"] == "job-save-001"
+        assert asset.uploaded_by_user_id == "user-1"
+        # 音频确实转存到了素材 OSS 路径，且 content_type 正确
+        assert storage.uploads[0][0] == "uploads/voice/tts/job-save-001.mp3"
+        assert storage.uploads[0][1] == "audio/mpeg"
+
+    def test_save_auto_creates_voice_library_when_missing(self):
+        """用户有项目但没有 voice 素材库 → 自动创建后挂 asset。"""
+        from fastapi.testclient import TestClient
+
+        app, asset_repo, _ = _build_app(job=_completed_job(), libs=[], projects=[FakeProject()])
+        client = TestClient(app)
+
+        resp = client.post("/tts/jobs/job-save-001/save-to-library", json={})
+        assert resp.status_code == 201, resp.text
+        asset = asset_repo.created[0]
+        assert asset.project_id == "proj-1"
+
+    def test_save_concurrent_library_creation_integrity_error(self):
+        """并发建库竞态：create 抛 IntegrityError → 回滚重查返回抢先创建的库。"""
+        from fastapi.testclient import TestClient
+
+        app, asset_repo, _ = _build_app(job=_completed_job(), libs=[], projects=[FakeProject()], lib_fail=True)
+        client = TestClient(app)
+
+        resp = client.post("/tts/jobs/job-save-001/save-to-library", json={})
+        assert resp.status_code == 201, resp.text
+        assert len(asset_repo.created) == 1
+
+    def test_save_rejects_uncompleted_job(self):
+        """未完成的 job → 400。"""
+        from fastapi.testclient import TestClient
+
+        app, _, _ = _build_app(job=_completed_job(status=TTSJobStatus.PROCESSING))
+        client = TestClient(app)
+        resp = client.post("/tts/jobs/job-save-001/save-to-library", json={})
+        assert resp.status_code == 400
+
+    def test_save_rejects_job_without_audio(self):
+        """已完成但无 output url/key → 400。"""
+        from fastapi.testclient import TestClient
+
+        app, _, _ = _build_app(
+            job=_completed_job(output_audio_url="", output_audio_key=""),
+            libs=[AssetLibrary.create(project_id="proj-1", name="配音库", kind=AssetLibraryKind.VOICE)],
+        )
+        client = TestClient(app)
+        resp = client.post("/tts/jobs/job-save-001/save-to-library", json={})
+        assert resp.status_code == 400
+
+    def test_save_download_failure_returns_502(self):
+        """TTS 音频下载失败 → 502，不创建 asset。"""
+        from fastapi.testclient import TestClient
+
+        voice_lib = AssetLibrary.create(project_id="proj-1", name="配音素材库", kind=AssetLibraryKind.VOICE)
+        app, asset_repo, _ = _build_app(job=_completed_job(), libs=[voice_lib], storage=FakeStorage(download_ok=False))
+        client = TestClient(app)
+        resp = client.post("/tts/jobs/job-save-001/save-to-library", json={})
+        assert resp.status_code == 502
+        assert asset_repo.created == []
+
+    def test_save_concurrent_race_tolerates_closed_session(self):
+        """IntegrityError 后 session.rollback() 抛异常（session 已关闭）→ 容错继续重查成功。"""
+        from fastapi.testclient import TestClient
+
+        app, asset_repo, _ = _build_app(
+            job=_completed_job(), libs=[], projects=[FakeProject()], lib_fail=True, rollback_raises=True
+        )
+        client = TestClient(app)
+        resp = client.post("/tts/jobs/job-save-001/save-to-library", json={})
+        assert resp.status_code == 201, resp.text
+        assert len(asset_repo.created) == 1
+
+    def test_save_db_failure_cleans_orphan_oss_file(self):
+        """asset_repository.create 抛异常 → 已上传的 OSS 文件被删除，返回 502。"""
+        from fastapi.testclient import TestClient
+
+        class FailAssetRepo(FakeAssetRepo):
+            def create(self, asset):
+                raise RuntimeError("DB connection lost")
+
+        voice_lib = AssetLibrary.create(project_id="proj-1", name="配音素材库", kind=AssetLibraryKind.VOICE)
+        storage = FakeStorage()
+        app, _, _ = _build_app(job=_completed_job(), libs=[voice_lib], storage=storage)
+        # 替换 asset repo 为会失败的实现
+        from app.dependencies import get_asset_repository
+
+        app.dependency_overrides[get_asset_repository] = lambda: FailAssetRepo()
+        client = TestClient(app)
+
+        resp = client.post("/tts/jobs/job-save-001/save-to-library", json={})
+        assert resp.status_code == 502, resp.text
+        # OSS 文件已上传后又被清理
+        assert storage.uploads, "音频应已上传"
+        assert hasattr(storage, "deleted") and storage.deleted, "失败后应删除孤儿 OSS 文件"
+
+    def test_save_no_project_returns_400(self):
+        """用户没有任何可访问项目 → 400。"""
+        from fastapi.testclient import TestClient
+
+        app, _, _ = _build_app(job=_completed_job(), projects=[])
+        client = TestClient(app)
+        resp = client.post("/tts/jobs/job-save-001/save-to-library", json={})
+        assert resp.status_code == 400
+
+    def test_save_other_users_job_returns_404(self):
+        """保存他人 job → 404。"""
+        from fastapi.testclient import TestClient
+
+        app, _, _ = _build_app(job=_completed_job(user_id="other-user"))
+        client = TestClient(app)
+        resp = client.post("/tts/jobs/job-save-001/save-to-library", json={})
+        assert resp.status_code == 404
diff --git a/tests/unit/test_tts_synthesize_clone_voice.py b/tests/unit/test_tts_synthesize_clone_voice.py
new file mode 100644
index 000000000..821df2c55
--- /dev/null
+++ b/tests/unit/test_tts_synthesize_clone_voice.py
@@ -0,0 +1,168 @@
+"""POST /tts/synthesize 克隆音色解析的单元测试（路由级）。
+
+覆盖：voice_id 传克隆 profile UUID / 显式 voice_clone_profile_id /
+预置音色+profile 同传时以克隆为准 / 403 / 400 未完成 / 404 / 预置透传。
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from packages.domain.tts_job import TTSJob, TTSJobStatus
+
+
+class FakeCloneProfile:
+    def __init__(self, pid, user_id="user-1", voice_id="clone-voice-001"):
+        self.id = pid
+        self.user_id = user_id
+        self.voice_id = voice_id
+
+
+class FakeCloneRepo:
+    def __init__(self, profiles):
+        self._profiles = {p.id: p for p in profiles}
+
+    def get(self, pid):
+        return self._profiles.get(pid)
+
+
+def _make_job(voice_id, voice_clone_profile_id=""):
+    return TTSJob(
+        id="job-new",
+        user_id="user-1",
+        input_text="x",
+        voice_id=voice_id,
+        voice_model="",
+        project_id="",
+        voice_clone_profile_id=voice_clone_profile_id,
+        status=TTSJobStatus.PROCESSING,
+        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),
+    )
+
+
+def _build_app(clone_repo):
+    from fastapi import FastAPI
+
+    app = FastAPI()
+    from app.api.routes.tts import _get_repository, router
+    from app.auth import get_current_user
+    from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repository
+
+    app.include_router(router, prefix="/tts")
+
+    fake_user = MagicMock()
+    fake_user.user.id = "user-1"
+    app.dependency_overrides[get_current_user] = lambda: fake_user
+    app.dependency_overrides[get_voice_clone_profile_repository] = lambda: clone_repo
+    app.dependency_overrides[get_cosyvoice_service] = lambda: MagicMock()
+    app.dependency_overrides[_get_repository] = lambda: MagicMock()
+    return app
+
+
+class TestSynthesizeCloneVoiceResolution:
+    def _post(self, client, payload):
+        captured = {}
+
+        def fake_execute(*args, **kwargs):
+            captured.update(kwargs)
+            return _make_job(kwargs["voice_id"], kwargs.get("voice_clone_profile_id", ""))
+
+        def fake_start(*args, **kwargs):
+            return _make_job(captured.get("voice_id", ""), captured.get("voice_clone_profile_id", ""))
+
+        with (
+            patch(
+                "packages.application.tts_job.use_cases.CreateTTSJobUseCase.execute",
+                side_effect=fake_execute,
+            ),
+            patch(
+                "packages.application.tts_job.workflow.TTSWorkflowService.start_synthesis",
+                side_effect=fake_start,
+            ),
+            patch("app.api.routes.tts.celery_app.send_task"),
+        ):
+            resp = client.post("/tts/synthesize", json=payload)
+        return resp, captured
+
+    def test_voice_id_as_profile_uuid_resolves_to_clone_voice(self):
+        from fastapi.testclient import TestClient
+
+        profile = FakeCloneProfile("profile-uuid-1")
+        client = TestClient(_build_app(FakeCloneRepo([profile])))
+        resp, captured = self._post(client, {"text": "克隆音色", "voice_id": "profile-uuid-1"})
+        assert resp.status_code == 201, resp.text
+        assert captured["voice_id"] == "clone-voice-001"
+        assert captured["voice_clone_profile_id"] == "profile-uuid-1"
+
+    def test_explicit_profile_id_resolves(self):
+        from fastapi.testclient import TestClient
+
+        profile = FakeCloneProfile("profile-uuid-2")
+        client = TestClient(_build_app(FakeCloneRepo([profile])))
+        resp, captured = self._post(client, {"text": "显式", "voice_clone_profile_id": "profile-uuid-2"})
+        assert resp.status_code == 201, resp.text
+        assert captured["voice_id"] == "clone-voice-001"
+
+    def test_preset_voice_id_with_explicit_profile_uses_clone(self):
+        """voice_id 传预置音色 + 显式克隆 profile（不一致）→ 必须以克隆 profile 为准。"""
+        from fastapi.testclient import TestClient
+
+        profile = FakeCloneProfile("profile-uuid-3")
+        client = TestClient(_build_app(FakeCloneRepo([profile])))
+        resp, captured = self._post(
+            client,
+            {"text": "混合", "voice_id": "longxiaochun_v2", "voice_clone_profile_id": "profile-uuid-3"},
+        )
+        assert resp.status_code == 201, resp.text
+        assert captured["voice_id"] == "clone-voice-001"
+        assert captured["voice_clone_profile_id"] == "profile-uuid-3"
+
+    def test_other_user_profile_returns_403(self):
+        from fastapi.testclient import TestClient
+
+        profile = FakeCloneProfile("profile-other", user_id="other-user")
+        client = TestClient(_build_app(FakeCloneRepo([profile])))
+        resp, _ = self._post(client, {"text": "越权", "voice_id": "profile-other"})
+        assert resp.status_code == 403
+
+    def test_unfinished_clone_returns_400(self):
+        from fastapi.testclient import TestClient
+
+        profile = FakeCloneProfile("profile-pending", voice_id="")
+        client = TestClient(_build_app(FakeCloneRepo([profile])))
+        resp, _ = self._post(client, {"text": "未完成", "voice_id": "profile-pending"})
+        assert resp.status_code == 400
+        assert "克隆尚未完成" in resp.json()["detail"]
+
+    def test_nonexistent_profile_returns_404(self):
+        from fastapi.testclient import TestClient
+
+        client = TestClient(_build_app(FakeCloneRepo([])))
+        resp, _ = self._post(client, {"text": "不存在", "voice_clone_profile_id": "no-such-profile"})
+        assert resp.status_code == 404
+
+    def test_preset_voice_passes_through(self):
+        """预置音色（不命中任何 profile）→ voice_id 原样透传。"""
+        from fastapi.testclient import TestClient
+
+        client = TestClient(_build_app(FakeCloneRepo([])))
+        resp, captured = self._post(client, {"text": "预置", "voice_id": "longxiaochun_v2"})
+        assert resp.status_code == 201, resp.text
+        assert captured["voice_id"] == "longxiaochun_v2"
+        assert captured["voice_clone_profile_id"] == ""
diff --git a/tests/unit/test_voice_clone_from_asset.py b/tests/unit/test_voice_clone_from_asset.py
new file mode 100644
index 000000000..cf8c43220
--- /dev/null
+++ b/tests/unit/test_voice_clone_from_asset.py
@@ -0,0 +1,184 @@
+"""POST /voice-clones 支持 asset_id 的单元测试（路由级）。
+
+覆盖：asset_id 成功（签名 URL 传给 workflow）/ 与 source_audio_url 同传 400 /
+素材不存在 404 / 他人素材 403 / 非音频 400 / 无 storage_key 400 /
+原 source_audio_url 方式不受影响。
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+
+
+class FakeAsset:
+    def __init__(self, aid, project_id="proj-1", file_type="audio", storage_key="uploads/voice/a.mp3"):
+        self.id = aid
+        self.project_id = project_id
+        self.file_type = file_type
+        self.storage_key = storage_key
+
+
+class FakeAssetRepo:
+    def __init__(self, assets):
+        self._assets = {a.id: a for a in assets}
+
+    def find_by_id(self, aid):
+        return self._assets.get(aid)
+
+
+class FakeProject:
+    def __init__(self, pid="proj-1", owner="user-1"):
+        self.id = pid
+        self.owner_user_id = owner
+
+    def can_access(self, user_id):
+        return user_id == self.owner_user_id
+
+
+class FakeProjectRepo:
+    def __init__(self, projects):
+        self._projects = {p.id: p for p in projects}
+
+    def find_by_id(self, pid):
+        return self._projects.get(pid)
+
+
+class FakeStorage:
+    def get_download_url(self, key, expires_seconds=3600):
+        return f"https://oss.example.com/signed/{key}?expires={expires_seconds}"
+
+
+def _build_app(*, assets, projects, current_user="user-1"):
+    from fastapi import FastAPI
+
+    app = FastAPI()
+    from app.api.routes.voice_clones import _get_workflow_service, router
+    from app.auth import get_current_user
+    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,
+    )
+
+    app.include_router(router, prefix="/voice-clones")
+
+    fake_user = MagicMock()
+    fake_user.user.id = current_user
+    app.dependency_overrides[get_current_user] = lambda: fake_user
+
+    asset_repo = FakeAssetRepo(assets)
+    project_repo = FakeProjectRepo(projects)
+    storage = FakeStorage()
+    app.dependency_overrides[get_asset_repository] = lambda: asset_repo
+    app.dependency_overrides[get_project_repository] = lambda: project_repo
+    app.dependency_overrides[get_storage_service] = lambda: storage
+    app.dependency_overrides[get_voice_clone_profile_repository] = lambda: MagicMock()
+    app.dependency_overrides[get_cosyvoice_service] = lambda: MagicMock()
+
+    from packages.domain.voice_clone_profile import VoiceCloneProfile
+
+    workflow = MagicMock()
+    profile = VoiceCloneProfile.create(
+        user_id=current_user,
+        name="我的克隆",
+        voice_model="cosyvoice-v2",
+    )  # pending 状态，不触发 celery send_task
+    workflow.start_clone.return_value = profile
+    app.dependency_overrides[_get_workflow_service] = lambda: workflow
+    return app, workflow
+
+
+class TestCreateVoiceCloneFromAsset:
+    def test_success_with_asset_id(self):
+        from fastapi.testclient import TestClient
+
+        asset = FakeAsset("asset-1")
+        app, workflow = _build_app(assets=[asset], projects=[FakeProject()])
+        client = TestClient(app)
+
+        resp = client.post("/voice-clones", json={"name": "我的克隆", "asset_id": "asset-1"})
+        assert resp.status_code in (200, 201), resp.text
+
+        kwargs = workflow.start_clone.call_args.kwargs
+        # worker 拿到的是服务端用 storage_key 生成的签名 URL，不依赖前端
+        assert kwargs["source_audio_url"].startswith("https://oss.example.com/signed/uploads/voice/a.mp3")
+        assert "expires=604800" in kwargs["source_audio_url"]
+        assert kwargs["metadata"]["source_asset_id"] == "asset-1"
+
+    def test_asset_id_and_url_together_returns_400(self):
+        from fastapi.testclient import TestClient
+
+        app, workflow = _build_app(assets=[FakeAsset("asset-1")], projects=[FakeProject()])
+        client = TestClient(app)
+        resp = client.post(
+            "/voice-clones",
+            json={
+                "name": "冲突",
+                "asset_id": "asset-1",
+                "source_audio_url": "https://example.com/a.mp3",
+            },
+        )
+        assert resp.status_code == 400
+        workflow.start_clone.assert_not_called()
+
+    def test_nonexistent_asset_returns_404(self):
+        from fastapi.testclient import TestClient
+
+        app, workflow = _build_app(assets=[], projects=[FakeProject()])
+        client = TestClient(app)
+        resp = client.post("/voice-clones", json={"name": "x", "asset_id": "no-such-asset"})
+        assert resp.status_code == 404
+        workflow.start_clone.assert_not_called()
+
+    def test_other_users_asset_returns_403(self):
+        from fastapi.testclient import TestClient
+
+        asset = FakeAsset("asset-2", project_id="proj-other")
+        app, workflow = _build_app(
+            assets=[asset],
+            projects=[FakeProject(pid="proj-other", owner="other-user")],
+        )
+        client = TestClient(app)
+        resp = client.post("/voice-clones", json={"name": "越权", "asset_id": "asset-2"})
+        assert resp.status_code == 403
+        workflow.start_clone.assert_not_called()
+
+    def test_non_audio_asset_returns_400(self):
+        from fastapi.testclient import TestClient
+
+        asset = FakeAsset("asset-3", file_type="video")
+        app, workflow = _build_app(assets=[asset], projects=[FakeProject()])
+        client = TestClient(app)
+        resp = client.post("/voice-clones", json={"name": "视频素材", "asset_id": "asset-3"})
+        assert resp.status_code == 400
+        assert "音频" in resp.json()["detail"]
+        workflow.start_clone.assert_not_called()
+
+    def test_asset_without_storage_key_returns_400(self):
+        from fastapi.testclient import TestClient
+
+        asset = FakeAsset("asset-4", storage_key="")
+        app, workflow = _build_app(assets=[asset], projects=[FakeProject()])
+        client = TestClient(app)
+        resp = client.post("/voice-clones", json={"name": "空key", "asset_id": "asset-4"})
+        assert resp.status_code == 400
+        workflow.start_clone.assert_not_called()
+
+    def test_legacy_source_audio_url_still_works(self):
+        """不传 asset_id、只传 source_audio_url 的旧流程保持兼容。"""
+        from fastapi.testclient import TestClient
+
+        app, workflow = _build_app(assets=[], projects=[])
+        client = TestClient(app)
+        resp = client.post(
+            "/voice-clones",
+            json={"name": "旧流程", "source_audio_url": "https://example.com/voice.mp3"},
+        )
+        assert resp.status_code in (200, 201), resp.text
+        kwargs = workflow.start_clone.call_args.kwargs
+        assert kwargs["source_audio_url"] == "https://example.com/voice.mp3"
+        assert "source_asset_id" not in kwargs["metadata"]