fix(segments): AI Review 阻塞项修复 — 记录写入行级锁 + 克隆失败不降级共用

- record_used_segments 读取 AssetModel 加 with_for_update() 行级锁，
  序列化并发事务对 classification_result 的读-改-写，杜绝区间记录丢失更新
  （SQLite 下 with_for_update 为 no-op）
- 批量生成：变体 plan 克隆前置到任务创建之前，全部成功才建任务（无脏数据）；
  克隆含 1 次重试抗 DB 抖动，仍失败则 500 中断，不再静默退回共用源 plan
- 预览生成：克隆失败标记任务 failed 并返回 500，不再退回原 plan
  （两条修复共同保证批量/多预览视频内容不重复的业务约束）
- 测试：FakeSession 支持 with_for_update；预览相关测试 mock EditPlanService
  或显式置空 source_edit_plan_id；zip 加 strict（B905）

apps/api/app/api/routes/generation_preview.py

@@ -379,12 +379,19 @@ def create_preview_generation_task(
                 _preview_plan.id,
             )
         except Exception as clone_err:
-            logger.warning(
-                "[预览生成] 克隆预览变体 plan 失败，退回原 plan: task_id=%s error=%s",
+            # 不退回共用原 plan（否则多条预览内容相同，违反去重诉求）：
+            # 标记任务失败并中断，前端可重新发起预览
+            logger.error(
+                "[预览生成] 克隆预览变体 plan 失败，任务标记失败: task_id=%s error=%s",
                 task.id,
                 clone_err,
                 exc_info=True,
             )
+            _mark_task_failed(generation_task_repository, task, "预览变体计划创建失败")
+            raise HTTPException(
+                status_code=500,
+                detail="创建预览任务失败：无法生成独立剪辑计划，请重试",
+            ) from clone_err
 
     # 入队执行；若入队失败则标记任务为 failed 避免僵尸数据
     try:

apps/api/app/api/routes/generation_tasks.py

@@ -419,34 +419,52 @@ def create_generation_task(
         effective_strategy_id = "one_take"
 
     # 批量生成时每个任务关联独立克隆 plan（片段起点重算），
-    # 禁止 N 条任务共用同一 source_edit_plan_id 导致片段一模一样
+    # 禁止 N 条任务共用同一 source_edit_plan_id 导致片段一模一样。
+    # 在创建任何任务【之前】预克隆全部变体：克隆失败直接中断（此时无脏数据），
+    # 绝不静默退回共用源 plan（否则批量视频内容重复，违反去重诉求）。
     variant_plan_ids: list[str] = []
+    if count > 1 and request.source_edit_plan_id:
+        from app.services.edit_plan_service import EditPlanService
 
-    try:
-        for task_index in range(count):
-            # 第 1 条复用源 plan（保留用户编辑结果）；其余克隆独立变体 plan
-            effective_plan_id = request.source_edit_plan_id
-            if task_index > 0 and request.source_edit_plan_id:
+        _plan_svc = EditPlanService(db)
+        for task_index in range(1, count):
+            variant = None
+            last_err: Exception | None = None
+            for _attempt in range(2):  # 1 次重试，抗 DB 瞬时抖动
                 try:
-                    from app.services.edit_plan_service import EditPlanService
-
-                    _plan_svc = EditPlanService(db)
-                    _variant_plan = _plan_svc.clone_plan_for_variant(
+                    variant = _plan_svc.clone_plan_for_variant(
                         request.source_edit_plan_id,
                         created_by_user_id=user_id,
                         name_suffix=f"批量{task_index + 1}",
                     )
-                    effective_plan_id = _variant_plan.id
-                    variant_plan_ids.append(_variant_plan.id)
-                except Exception as clone_err:
-                    # 克隆失败不阻断：退回源 plan（保证任务可创建），记录日志便于排查
+                    break
+                except Exception as clone_err:  # noqa: PERF203
+                    last_err = clone_err
                     logger.warning(
-                        "[生成任务] 克隆变体 plan 失败，退回源 plan: source=%s error=%s",
+                        "[生成任务] 克隆变体 plan 失败(尝试%d/2): source=%s error=%s",
+                        _attempt + 1,
                         request.source_edit_plan_id,
                         clone_err,
                         exc_info=True,
                     )
-                    effective_plan_id = request.source_edit_plan_id
+            if variant is None:
+                logger.error(
+                    "[生成任务] 克隆变体 plan 重试仍失败，中断批量创建: source=%s",
+                    request.source_edit_plan_id,
+                    exc_info=last_err,
+                )
+                raise HTTPException(
+                    status_code=500,
+                    detail="创建批量任务失败：无法生成独立剪辑计划，请重试",
+                ) from last_err
+            variant_plan_ids.append(variant.id)
+
+    try:
+        for task_index in range(count):
+            # 第 1 条复用源 plan（保留用户编辑结果）；其余使用预克隆的独立变体 plan
+            effective_plan_id = (
+                variant_plan_ids[task_index - 1] if task_index > 0 else request.source_edit_plan_id
+            )
 
             task = use_case.execute(
                 CreateGenerationTaskCommand(

apps/api/app/services/asset_segment_tracker.py

@@ -71,8 +71,14 @@ def _read_meta(model: AssetModel) -> dict:
         return {}
 
 
-def _get_model(db: Session, asset_id: str) -> AssetModel | None:
-    return db.query(AssetModel).filter(AssetModel.id == asset_id).first()
+def _get_model(db: Session, asset_id: str, for_update: bool = False) -> AssetModel | None:
+    query = db.query(AssetModel).filter(AssetModel.id == asset_id)
+    if for_update:
+        # 行级锁（PostgreSQL SELECT ... FOR UPDATE）：序列化同一素材的
+        # classification_result 读-改-写，避免并发事务丢失使用记录。
+        # SQLite 不支持时 SQLAlchemy 会忽略该子句（no-op）。
+        query = query.with_for_update()
+    return query.first()
 
 
 def get_used_segments(db: Session, asset_ids: list[str]) -> dict[str, list[tuple[float, float]]]:
@@ -113,7 +119,8 @@ def record_used_segments(
     MediaKit 挪到历史区间），则累加该记录的 ``use_count`` 并刷新 ``last_used_at``，
     不新增记录；否则追加一条新记录（use_count=1）。
     """
-    model = _get_model(db, asset_id)
+    # 行级锁读取：与并发生成任务互斥，保证区间记录读-改-写一致
+    model = _get_model(db, asset_id, for_update=True)
     if model is None:
         logger.warning("[片段追踪] 素材不存在，跳过记录: asset_id=%s", asset_id)
         return

tests/unit/test_asset_segment_tracker.py

@@ -66,6 +66,10 @@ class _EqExpr:
         self._target_id = target_id
         self._models = models
 
+    def with_for_update(self):
+        # 模拟 SQLAlchemy Query.with_for_update() 链式返回自身
+        return self
+
     def first(self):
         return self._models.get(self._target_id)
 

tests/unit/test_clone_plan_variant.py

@@ -124,7 +124,7 @@ class TestClonePlanForVariant:
         assert source.id not in plan_ids
 
         # 每次克隆的起点各自不同
-        for clips_data, starts in zip(all_clips, start_sets):
+        for clips_data, starts in zip(all_clips, start_sets, strict=True):
             assert [c["start_time"] for c in clips_data] == starts
 
         # 三次克隆的起点集合互不相同

tests/unit/test_generation_preview.py

@@ -612,6 +612,8 @@ def _make_task(
     task.extra_meta = extra_meta or {}
     task.asset_ids = asset_ids or []
     task.created_by_user_id = "test_user_001"
+    # 默认无关联编辑计划：涉及克隆变体的测试自行设置并 mock EditPlanService
+    task.source_edit_plan_id = None
     return task
 
 

tests/unit/test_preview_edit_plan_association.py

@@ -14,6 +14,7 @@ import os
 import sys
 from dataclasses import dataclass, field
 from typing import Any, Optional
+from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
@@ -180,17 +181,23 @@ class TestPreviewEditPlanAutoAssociation:
             "packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository",
             return_value=fake_plan_repo,
         ):
-            resp = client.post(
-                "/api/v1/generation/preview",
-                json=_make_request_body(source_edit_plan_id=""),
-            )
+            with patch(
+                "app.services.edit_plan_service.EditPlanService"
+            ) as MockPlanSvc:
+                MockPlanSvc.return_value.clone_plan_for_variant.return_value = SimpleNamespace(
+                    id="plan-clone-001"
+                )
+                resp = client.post(
+                    "/api/v1/generation/preview",
+                    json=_make_request_body(source_edit_plan_id=""),
+                )
 
         assert resp.status_code == 201
-        # 找到 store 中的 task 并验证 source_edit_plan_id 被设置
+        # 找到 store 中的 task 并验证 source_edit_plan_id 被设置（自动关联后再克隆为独立 plan）
         tasks = list(gen_task_repo._store.values())
         assert len(tasks) == 1
         task = tasks[0]
-        assert task.source_edit_plan_id == "plan-auto-001"
+        assert task.source_edit_plan_id == "plan-clone-001"
 
     @patch(
         "app.api.routes.generation_preview._resolve_strategy_id_from_template",
@@ -209,16 +216,20 @@ class TestPreviewEditPlanAutoAssociation:
         client: TestClient,
         gen_task_repo: StubGenerationTaskRepository,
     ):
-        """前端已传 source_edit_plan_id 时，不应触发自动关联"""
-        resp = client.post(
-            "/api/v1/generation/preview",
-            json=_make_request_body(source_edit_plan_id="plan-explicit-001"),
-        )
+        """前端已传 source_edit_plan_id 时，不触发自动关联，但仍克隆独立变体 plan"""
+        with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc:
+            MockPlanSvc.return_value.clone_plan_for_variant.return_value = SimpleNamespace(
+                id="plan-clone-explicit"
+            )
+            resp = client.post(
+                "/api/v1/generation/preview",
+                json=_make_request_body(source_edit_plan_id="plan-explicit-001"),
+            )
 
         assert resp.status_code == 201
         tasks = list(gen_task_repo._store.values())
         assert len(tasks) == 1
-        assert tasks[0].source_edit_plan_id == "plan-explicit-001"
+        assert tasks[0].source_edit_plan_id == "plan-clone-explicit"
 
     @patch(
         "app.api.routes.generation_preview._resolve_strategy_id_from_template",