fix(tts): AI Review 第二轮——migration 去重清洗 + session 回滚容错

1. migration 058：create unique index 前先按 (project_id, kind) 分区
   ROW_NUMBER 去重（保留 created_at 最新），防止存量重复导致建索引失败阻塞部署。
2. _find_or_create_voice_library 竞态处理：IntegrityError 后显式 rollback
   （立即 commit 模式下为幂等 no-op，UoW/flush 模式下必须回滚才能继续查询），
   rollback 本身抛异常（session 已关闭）时容错继续重查。补对应用例。
3. voice_clones：移除 file_type 的 hasattr 防御判断（Asset 领域对象确定有该字段）。

alembic/versions/058_uq_asset_lib_project_kind.py

@@ -8,6 +8,8 @@ Create Date: 2026-08-30
 加唯一索引兜底并发创建竞态，避免重复素材库。
 """
 
+import sqlalchemy as sa
+
 from alembic import op
 
 revision = "058_uq_asset_lib_project_kind"
@@ -17,6 +19,23 @@ 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
+        )
+        """)
     op.create_index(
         "uq_asset_libraries_project_kind",
         "asset_libraries",

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

@@ -342,8 +342,15 @@ def _find_or_create_voice_library(
     try:
         return asset_library_repository.create(library)
     except IntegrityError:
-        # 并发下另一个请求已抢先创建：SQLAlchemy commit 失败后 session 会自动回滚，
-        # 直接重查返回已存在的库即可（不依赖 repository 的内部 session 实现）。
+        # 并发下另一个请求已抢先创建：回滚当前事务（立即 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:

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

@@ -130,8 +130,7 @@ def create_voice_clone(
                 detail="无权使用该素材",
             )
         # 类型校验：仅支持音频素材
-        file_type = asset.file_type if hasattr(asset, "file_type") else ""
-        if file_type != "audio":
+        if asset.file_type != "audio":
             raise HTTPException(
                 status_code=status.HTTP_400_BAD_REQUEST,
                 detail="仅支持音频素材进行音色克隆",

tests/unit/test_tts_save_to_library_assets.py

@@ -85,9 +85,12 @@ class FakeProjectRepo:
 
 
 class FakeAssetLibraryRepo:
-    def __init__(self, libs=None, fail_integrity=False):
+    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]
@@ -130,7 +133,7 @@ class FakeStorage:
         return f"https://oss.example.com/signed/{key}?sig=xxx"
 
 
-def _build_app(*, job, libs=None, projects=None, storage=None, lib_fail=False):
+def _build_app(*, job, libs=None, projects=None, storage=None, lib_fail=False, rollback_raises=False):
     from fastapi import FastAPI
 
     app = FastAPI()
@@ -153,7 +156,9 @@ def _build_app(*, job, libs=None, projects=None, storage=None, lib_fail=False):
 
     asset_repo = FakeAssetRepo()
     app.dependency_overrides[get_asset_repository] = lambda: asset_repo
-    app.dependency_overrides[get_asset_library_repository] = lambda: FakeAssetLibraryRepo(libs, lib_fail)
+    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()]
     )
@@ -250,6 +255,18 @@ class TestSaveToLibraryAssets:
         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_no_project_returns_400(self):
         """用户没有任何可访问项目 → 400。"""
         from fastapi.testclient import TestClient