apps/api/app/api/routes/assets.py

@@ -26,6 +26,7 @@ from app.schemas.asset import (
     UpdateAssetReviewRequest,
 )
 from app.schemas.tag import TagAssetsRequest
+from app.services.asset_segment_tracker import compute_asset_availability
 from fastapi import APIRouter, Depends, HTTPException, Query, Response
 
 from packages.domain.smart_match import smart_select_assets
@@ -35,6 +36,23 @@ logger = logging.getLogger(__name__)
 router = APIRouter()
 
 
+def _asset_availability_fields(item) -> dict:
+    """视频素材返回余量四字段；非视频/无时长/异常时返回 None + usable=True（零影响）。"""
+    try:
+        info = compute_asset_availability(item)
+    except Exception:
+        logger.warning("计算素材余量失败，按可用处理: asset_id=%s", getattr(item, "id", "?"), exc_info=True)
+        info = None
+    if info is None:
+        return {
+            "used_duration": None,
+            "available_duration": None,
+            "used_ratio": None,
+            "usable": True,
+        }
+    return info
+
+
 def _to_asset_response(item, storage_service=None) -> AssetResponse:
     # 生成签名文件 URL（用于视频播放 / 文件下载）
     file_url = None
@@ -79,6 +97,7 @@ def _to_asset_response(item, storage_service=None) -> AssetResponse:
         created_at=format_utc_datetime(item.created_at),
         uploaded_by_user_id=item.uploaded_by_user_id,
         tag_ids=getattr(item, "tag_ids", []),
+        **_asset_availability_fields(item),
     )
 
 
@@ -569,13 +588,35 @@ def smart_match_assets(
         kind=None,
     )
 
+    # 结果层过滤：usable=false（零重复可切区间耗尽且历史区间均达复用上限）的素材
+    # 不返回给前端；不动 smart_select_assets 评分逻辑本身
+    filtered_results = []
+    for r in results:
+        try:
+            avail = compute_asset_availability(r.asset)
+        except Exception:
+            logger.warning(
+                "smart-match 余量计算失败，按可用处理: asset_id=%s",
+                getattr(r.asset, "id", "?"),
+                exc_info=True,
+            )
+            avail = None
+        if avail is not None and not avail["usable"]:
+            logger.info(
+                "smart-match 排除已用尽素材: asset_id=%s name=%s",
+                getattr(r.asset, "id", "?"),
+                getattr(r.asset, "name", ""),
+            )
+            continue
+        filtered_results.append(r)
+
     items = [
         SmartMatchItem(
             asset=_to_asset_response(r.asset),
             score=r.score,
             breakdown=r.breakdown,
         )
-        for r in results
+        for r in filtered_results
     ]
 
     return SmartMatchResponse(items=items, total_candidates=total_candidates)

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

@@ -358,6 +358,41 @@ def create_preview_generation_task(
                 exc_info=True,
             )
 
+    # 每条预览都关联独立克隆 plan：多预览前端为 N 次并发调用，若共用同一 plan
+    # 则 N 条预览片段完全相同；克隆时片段起点按持久化历史区间重算（含受控复用），
+    # 保证各预览版本内容不同
+    if task.source_edit_plan_id:
+        try:
+            from app.services.edit_plan_service import EditPlanService
+
+            _plan_svc = EditPlanService(db)
+            _preview_plan = _plan_svc.clone_plan_for_variant(
+                task.source_edit_plan_id,
+                created_by_user_id=user_id,
+                name_suffix="预览变体",
+            )
+            task.source_edit_plan_id = _preview_plan.id
+            generation_task_repository.update(task)
+            logger.info(
+                "[预览生成] 预览关联独立克隆 plan: task_id=%s clone_plan_id=%s",
+                task.id,
+                _preview_plan.id,
+            )
+        except Exception as clone_err:
+            # 不退回共用原 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:
         if not safe_enqueue_generation_task(

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

@@ -418,8 +418,56 @@ def create_generation_task(
         logger.info("画中画已下线，strategy_id %s → one_take", effective_strategy_id)
         effective_strategy_id = "one_take"
 
+    # 批量生成时每个任务关联独立克隆 plan（片段起点重算），
+    # 禁止 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
+
+        _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:
+                    variant = _plan_svc.clone_plan_for_variant(
+                        request.source_edit_plan_id,
+                        created_by_user_id=user_id,
+                        name_suffix=f"批量{task_index + 1}",
+                    )
+                    break
+                except Exception as clone_err:  # noqa: PERF203
+                    last_err = clone_err
+                    logger.warning(
+                        "[生成任务] 克隆变体 plan 失败(尝试%d/2): source=%s error=%s",
+                        _attempt + 1,
+                        request.source_edit_plan_id,
+                        clone_err,
+                        exc_info=True,
+                    )
+            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 _ in range(count):
+        for task_index in range(count):
+            # 第 1 条复用源 plan（保留用户编辑结果）；其余使用预克隆的独立变体 plan。
+            # 无源 plan（source_edit_plan_id 为空）时无可克隆对象，variant_plan_ids
+            # 为空列表：各任务走自身随机选片流程，不做索引访问（防 IndexError）
+            effective_plan_id = request.source_edit_plan_id
+            if task_index > 0 and variant_plan_ids:
+                effective_plan_id = variant_plan_ids[task_index - 1]
+
             task = use_case.execute(
                 CreateGenerationTaskCommand(
                     project_id=project_id,
@@ -431,7 +479,7 @@ def create_generation_task(
                     title_ids=request.title_ids,
                     voice_ids=request.voice_ids,
                     created_by_user_id=user_id,
-                    source_edit_plan_id=request.source_edit_plan_id,
+                    source_edit_plan_id=effective_plan_id,
                     asset_select_mode=request.asset_select_mode,
                     batch_id=batch_id,
                     video_title=request.video_title,

apps/api/app/api/routes/templates_editor/clips.py

@@ -24,8 +24,10 @@ from app.auth import AuthenticatedUser, get_current_user
 from app.core.storage import get_storage_service
 from app.dependencies import get_asset_repository, get_db_session
 from app.services.asset_segment_tracker import (
+    REUSE_RATIO_LIMIT,
+    SEGMENT_EDGE_GAP,
     get_used_segments,
-    make_reset_callback,
+    make_reuse_callback,
     record_used_segments,
     remove_used_segment,
 )
@@ -436,11 +438,16 @@ def _recommended_time_conflicts(
     start: float,
     duration: float,
     used: list[tuple[float, float]],
+    edge_gap: float = SEGMENT_EDGE_GAP,
 ) -> bool:
-    """检查推荐起始时间是否与已使用时间段冲突."""
+    """检查推荐起始时间是否与已使用时间段冲突.
+
+    冲突检测统一加 ``edge_gap`` 秒边缘间隙：已用区间按 [s-gap, e+gap] 扩边后判定，
+    避免推荐片段与已用片段首尾紧贴导致画面观感重复。
+    """
     end = start + duration
     for used_start, used_end in used:
-        if start < used_end and end > used_start:
+        if start < used_end + edge_gap and end > used_start - edge_gap:
             return True
     return False
 
@@ -611,55 +618,98 @@ def create_clips_from_assets_editor(
     used_segments: dict[str, list[tuple[float, float]]] = get_used_segments(
         db, unique_asset_ids
     )
-    reset_cb = make_reset_callback(db, used_segments)
+    # 受控复用回调：可用区间耗尽时复用最久未用且未达复用上限（3次）的历史区间，
+    # 复用片段时长累加到 reused_durations 供 15% 占比控制
+    reused_durations: dict[str, float] = {}
+    # 本条成片中每个素材被分配的片段总时长（复用占比分母）
+    asset_assigned_durations: dict[str, float] = {}
+    # 受控复用回调：区间耗尽时复用最久未用且 use_count<3 的历史区间；
+    # 回调内部预判复用后占比是否超 15%，超限拒绝复用（返回 None）
+    reuse_cb = make_reuse_callback(
+        db,
+        asset_durations,
+        reused_durations,
+        assigned_tracker=asset_assigned_durations,
+    )
     clips_data: list[dict] = []
 
+    def _reuse_ratio_exceeded(aid: str, extra: float = 0.0) -> bool:
+        """该素材在本条成片中「已复用片段时长 / 已分配片段总时长」是否已超 15%。
+
+        在为下一片段选素材时调用：本片段尚未分配，复用状态只在分配后的回调里
+        更新，因此直接检查当前占比——一旦已超 15%，该素材不再参与后续分配。
+        assigned=0（首个片段）放行；reused=0（尚未发生复用）时不误拦正常分配。
+        """
+        assigned = asset_assigned_durations.get(aid, 0.0)
+        if assigned <= 0:
+            return False
+        return reused_durations.get(aid, 0.0) / assigned > REUSE_RATIO_LIMIT
+
     for i, (_seg_order, dur_min, dur_max) in enumerate(segments):
-        # 轮询分配素材
-        asset_id = body.asset_ids[i % len(body.asset_ids)]
-        asset_total = asset_durations.get(asset_id, 0.0)
-
-        # 素材时长为 0 或缺失时无法创建有效片段
-        if asset_total <= 0:
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail=f"素材 {asset_id} 时长信息缺失或为0，无法创建片段",
-            )
-
         # 在 segment 的 duration_min ~ duration_max 之间随机取值（保留一位小数）
         raw_duration = random.uniform(dur_min, dur_max)
-        clip_duration = round(raw_duration, 1)
 
-        # 素材时长不足时缩短 clip duration
-        clip_duration = min(clip_duration, asset_total)
-
-        if clip_duration <= 0:
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail=f"素材 {asset_id} 时长不足，无法创建有效片段",
+        # 轮询分配素材：跳过时长缺失、复用占比已超 15% 阈值的素材；
+        # 选中后计算起点，若该素材可用区间耗尽且复用被闸门拒绝（calc 返回 None），
+        # 继续轮询下一个素材
+        asset_id = ""
+        clip_duration = 0.0
+        start_time: float | None = None
+        n_assets = len(body.asset_ids)
+        for offset in range(n_assets):
+            candidate = body.asset_ids[(i + offset) % n_assets]
+            candidate_total = asset_durations.get(candidate, 0.0)
+            if candidate_total <= 0:
+                continue
+            candidate_duration = min(round(raw_duration, 1), candidate_total)
+            if candidate_duration <= 0:
+                continue
+            if _reuse_ratio_exceeded(candidate, candidate_duration):
+                logger.info(
+                    "from-assets 素材复用占比超 %.0f%% 阈值，跳过分配: asset_id=%s",
+                    REUSE_RATIO_LIMIT * 100,
+                    candidate,
+                )
+                continue
+            # 随机起始时间（不调用 MediaKit，保证接口快速返回）；100 次避不开
+            # 历史区间时走受控复用回调（复用片段累加 reused_durations，回调内部
+            # 预判复用后占比超 15% 则拒绝并返回 None）
+            candidate_start = _calc_random_start_time(
+                candidate,
+                candidate_duration,
+                asset_durations,
+                used_segments,
+                on_exhausted=reuse_cb,
             )
+            if candidate_start is None:
+                # 该素材可用区间耗尽且复用被闸门/use_count 上限拒绝 → 尝试下一素材
+                logger.info(
+                    "from-assets 素材无可用可切区间（复用被拒），轮询下一素材: asset_id=%s",
+                    candidate,
+                )
+                continue
+            asset_id = candidate
+            clip_duration = candidate_duration
+            start_time = candidate_start
+            break
 
-        # 使用随机起始时间（不调用MediaKit，保证接口快速返回）
-        start_time = _calc_random_start_time(
-            asset_id,
-            clip_duration,
-            asset_durations,
-            used_segments,
-            on_exhausted=reset_cb,
-        )
-
-        if start_time is None:
+        if not asset_id or start_time is None:
+            # 所有素材时长缺失、复用占比超阈值，或区间耗尽且复用被拒 → 素材可切区间不足
             raise HTTPException(
                 status_code=status.HTTP_400_BAD_REQUEST,
-                detail=f"素材 {asset_id} 时长信息缺失，无法计算起始时间",
+                detail="素材可切区间不足，请补充新素材",
             )
 
         # 记录已使用时间段（内存，供本次后续片段避开）
         used_segments.setdefault(asset_id, []).append(
             (start_time, start_time + clip_duration)
         )
+        asset_assigned_durations[asset_id] = (
+            asset_assigned_durations.get(asset_id, 0.0) + clip_duration
+        )
         # 同步写入素材 metadata（不 commit，与下方 replace_all_clips_transactional
-        # 处于同一事务，任一步失败整体回滚，不留脏数据）
+        # 处于同一事务，任一步失败整体回滚，不留脏数据）；
+        # 复用区间与历史记录高度重叠时 record 内部自动累加 use_count
         record_used_segments(
             db, asset_id, start_time, start_time + clip_duration, plan_id
         )
@@ -766,6 +816,10 @@ def _update_mediakit_recommendations_async(  # pragma: no cover
                     (clip.id, clip.start_time, clip.start_time + clip.duration)
                 )
 
+        # 读取素材全部历史已用区间（跨任务/跨 plan 持久化记录）：
+        # MediaKit 挪点必须与随机选片一样避让历史区间，否则会把片段挪回已用过的画面
+        historical_segments = get_used_segments(db, unique_asset_ids)
+
         # 已更新的片段ID（用于排除已移动的旧时间段）
         updated_clip_ids: set[str] = set()
         # 已更新的时间段
@@ -807,11 +861,23 @@ def _update_mediakit_recommendations_async(  # pragma: no cover
                 if cid != clip.id and cid not in updated_clip_ids
             ]
             other_segments.extend(updated_segments.get(asset_id, []))
+            # 并入该素材全部历史已用区间（含其他 plan/其他任务），set 去重：
+            # 本 plan 片段创建时已写入历史记录
+            # 并入该素材全部历史已用区间（含其他 plan/其他任务）。
+            # set 去重前先归一化精度（round 3 位），避免浮点尾差导致逻辑相同的
+            # 区间（如 1.0 与 1.0000000001）被误判为不同区间
+            def _norm(segs):
+                return {(round(float(a), 3), round(float(b), 3)) for a, b in segs}
 
-            # 检查是否与同素材其他片段时间段冲突
+            other_segments = list(
+                _norm(other_segments) | _norm(historical_segments.get(asset_id, []))
+            )
+
+            # 检查推荐时间是否与同 plan 片段或历史已用区间冲突（含 0.3s 边缘间隙）：
+            # 冲突时放弃该推荐、保留原随机起点（不硬挪到已用过的画面）
             if _recommended_time_conflicts(recommended_start, clip_duration, other_segments):
                 logger.info(
-                    "后台任务: 推荐时间冲突，跳过: asset_id=%s recommended=%.2f",
+                    "后台任务: 推荐时间与同片/历史区间冲突，保留原起点: asset_id=%s recommended=%.2f",
                     asset_id,
                     recommended_start,
                 )

apps/api/app/schemas/asset.py

@@ -53,6 +53,14 @@ class AssetResponse(BaseModel):
     created_at: str
     uploaded_by_user_id: str
     tag_ids: list[str] = Field(default_factory=list)
+    # 片段级余量信息（仅视频素材返回，非视频/无时长记录为 None，前端按可用处理）
+    used_duration: float | None = Field(default=None, description="已使用片段时长（秒，历史区间合并去重后）")
+    available_duration: float | None = Field(default=None, description="剩余可用时长（秒）= 素材总时长 - 已用时长")
+    used_ratio: float | None = Field(default=None, description="已用时长占比（0~1）")
+    usable: bool = Field(
+        default=True,
+        description="是否仍可用于新片段：零重复可切区间耗尽且所有历史区间复用次数" "（use_count）均达上限时为 false",
+    )
 
 
 MAX_BATCH_SIZE = 200

apps/api/app/services/asset_segment_tracker.py

@@ -1,17 +1,27 @@
-"""素材片段级使用记录追踪.
+"""素材片段级使用记录追踪与受控复用.
 
 在素材 metadata（assets.classification_result JSON）中持久化已使用的片段时间区间，
-供 from-assets 创建片段时避开历史区间，实现跨任务/跨调用的片段去重。
+供 from-assets 创建片段时避开历史区间，实现跨任务/跨调用的片段去重；
+素材可用区间耗尽后进入受控复用：允许有限次数（MAX_RANGE_USE_COUNT）复用最久未用
+的历史区间，配合调用方的成片复用占比控制（MAX_REUSE_RATIO = 15%），把任意两条
+成片的画面重复率控制在阈值内。
 
-metadata 中新增字段 ``used_time_ranges``::
+metadata 中的记录字段 ``used_time_ranges``::
 
     "used_time_ranges": [
-        {"start": 12.5, "end": 20.3, "plan_id": "plan-xxx", "created_at": "2026-08-29T12:00:00+00:00"},
+        {
+            "start": 12.5, "end": 20.3,
+            "plan_id": "plan-xxx",
+            "created_at": "2026-08-29T12:00:00+00:00",
+            "use_count": 1,              # 该区间累计被使用次数（复用一次 +1）
+            "last_used_at": "2026-08-29T12:00:00+00:00"  # 最近一次使用时间
+        },
         ...
     ]
 
 注意：本模块所有函数都不自行 commit，由调用方控制事务边界
 （from-assets 与 replace_all_clips_transactional 同事务；异步任务各自 commit）。
+历史记录永不自动清空（自动轮回重置已下线，reset_used_segments 仅保留给运维/测试）。
 """
 
 from __future__ import annotations
@@ -29,24 +39,51 @@ logger = logging.getLogger(__name__)
 
 USED_RANGES_KEY = "used_time_ranges"
 
+# ── 受控复用配置常量 ─────────────────────────────────────────────────────────
+MAX_RANGE_USE_COUNT = 3
+"""单条历史区间最多被使用次数（含首次），达到后不再参与复用。"""
 
-def _read_ranges(model: AssetModel) -> list[dict]:
+REUSE_RATIO_LIMIT = 0.15
+"""单条成片中，单个素材的复用片段累计时长 / 该素材在成片中的总时长上限（15%）。
+超过则该素材不再分配新片段（调用方在轮询分配时跳过）。"""
+
+SEGMENT_EDGE_GAP = 0.3
+"""冲突判定边缘间隙（秒）：历史区间按 [start-gap, end+gap] 扩边后参与冲突检测，
+避免两条片段首尾紧贴导致画面观感重复；记录仍存实际值。"""
+
+# 判定"新片段与历史区间为同一次使用（复用）"的重叠率阈值：
+# 重叠时长 / 新区间时长超过该比例视为复用该历史区间（累加 use_count）而非新增记录。
+_REUSE_OVERLAP_RATIO = 0.6
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _read_meta(model: AssetModel) -> dict:
     """从 AssetModel 读取 metadata dict（classification_result 列承载的 JSON）."""
     if not model.classification_result:
         return {}
     try:
-        return json.loads(model.classification_result)
+        data = json.loads(model.classification_result)
+        return data if isinstance(data, dict) else {}
     except Exception:
         return {}
 
 
+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]]]:
     """聚合多个素材的历史已用片段区间。
 
-    Args:
-        db: SQLAlchemy session
-        asset_ids: 素材 ID 列表
-
     Returns:
         ``{asset_id: [(start, end), ...]}`` 格式，与 ``_calc_random_start_time`` 的
         ``used_segments`` 参数格式一致，可直接传入。
@@ -56,7 +93,7 @@ def get_used_segments(db: Session, asset_ids: list[str]) -> dict[str, list[tuple
     result: dict[str, list[tuple[float, float]]] = {}
     models = db.query(AssetModel).filter(AssetModel.id.in_(list(set(asset_ids)))).all()
     for model in models:
-        meta = _read_ranges(model)
+        meta = _read_meta(model)
         ranges = meta.get(USED_RANGES_KEY) or []
         segments: list[tuple[float, float]] = []
         for r in ranges:
@@ -76,19 +113,49 @@ def record_used_segments(
     end: float,
     plan_id: str,
 ) -> None:
-    """向素材 metadata 追加一条片段使用记录（不 commit）."""
-    model = db.query(AssetModel).filter(AssetModel.id == asset_id).first()
+    """记录一次片段使用（不 commit）.
+
+    若新区间与某条历史区间高度重叠（复用场景，如受控复用回调返回的区间、
+    MediaKit 挪到历史区间），则累加该记录的 ``use_count`` 并刷新 ``last_used_at``，
+    不新增记录；否则追加一条新记录（use_count=1）。
+    """
+    # 行级锁读取：与并发生成任务互斥，保证区间记录读-改-写一致
+    model = _get_model(db, asset_id, for_update=True)
     if model is None:
         logger.warning("[片段追踪] 素材不存在，跳过记录: asset_id=%s", asset_id)
         return
-    meta = _read_ranges(model)
+    meta = _read_meta(model)
     ranges = list(meta.get(USED_RANGES_KEY) or [])
+
+    new_start = round(float(start), 3)
+    new_end = round(float(end), 3)
+    new_dur = max(new_end - new_start, 1e-6)
+    now = _now_iso()
+
+    for r in ranges:
+        try:
+            rs, re_ = float(r["start"]), float(r["end"])
+        except (KeyError, TypeError, ValueError):
+            continue
+        overlap = max(0.0, min(new_end, re_) - max(new_start, rs))
+        if overlap / new_dur >= _REUSE_OVERLAP_RATIO:
+            # 复用同一条历史区间：累加次数、刷新时间
+            r["use_count"] = int(r.get("use_count", 1)) + 1
+            r["last_used_at"] = now
+            r["plan_id"] = plan_id
+            meta[USED_RANGES_KEY] = ranges
+            model.classification_result = json.dumps(meta, ensure_ascii=False)
+            model.updated_at = datetime.now(timezone.utc)
+            return
+
     ranges.append(
         {
-            "start": round(float(start), 3),
-            "end": round(float(end), 3),
+            "start": new_start,
+            "end": new_end,
             "plan_id": plan_id,
-            "created_at": datetime.now(timezone.utc).isoformat(),
+            "created_at": now,
+            "use_count": 1,
+            "last_used_at": now,
         }
     )
     meta[USED_RANGES_KEY] = ranges
@@ -111,10 +178,10 @@ def remove_used_segment(
     Returns:
         是否找到并删除了记录。
     """
-    model = db.query(AssetModel).filter(AssetModel.id == asset_id).first()
+    model = _get_model(db, asset_id)
     if model is None:
         return False
-    meta = _read_ranges(model)
+    meta = _read_meta(model)
     ranges = list(meta.get(USED_RANGES_KEY) or [])
     remaining: list[dict] = []
     removed = False
@@ -127,8 +194,7 @@ def remove_used_segment(
             remaining.append(r)
             continue
         # plan_id 校验：传入 plan_id 时，记录有 plan_id 则必须相等；
-        # 记录本身缺 plan_id（本功能上线前的旧数据）时退化为按时间匹配，
-        # 避免旧区间永远删不掉导致素材容量泄漏
+        # 记录本身缺 plan_id（旧数据）时退化为按时间匹配，避免旧区间永远删不掉
         if plan_id is not None and r.get("plan_id") is not None and r.get("plan_id") != plan_id:
             match = False
         if match and not removed:
@@ -145,30 +211,219 @@ def remove_used_segment(
 def reset_used_segments(db: Session, asset_id: str) -> None:
     """清空单个素材的历史片段使用记录（不 commit）.
 
-    单个素材的可用区间被全部占用（轮回一圈）后调用，使后续片段可重新使用整段素材。
+    仅供运维/测试使用；正常生成流程中历史记录永不自动清空（受控复用取代自动轮回）。
     """
-    model = db.query(AssetModel).filter(AssetModel.id == asset_id).first()
+    model = _get_model(db, asset_id)
     if model is None:
         return
-    meta = _read_ranges(model)
+    meta = _read_meta(model)
     if meta.get(USED_RANGES_KEY):
         meta[USED_RANGES_KEY] = []
         model.classification_result = json.dumps(meta, ensure_ascii=False)
         model.updated_at = datetime.now(timezone.utc)
-        logger.info("[片段追踪] 素材区间轮回重置: asset_id=%s", asset_id)
+        logger.info("[片段追踪] 素材区间记录手动清空: asset_id=%s", asset_id)
 
 
-def make_reset_callback(db: Session, used_segments: dict) -> Callable[[str], None]:
-    """构造给 _calc_random_start_time 用的 reset 回调.
+# ── 素材余量/可用性计算（Task H：素材库角标 + smart-match 过滤）──────────────
 
-    回调同时清空持久化 metadata 和内存中的 used_segments，使重试随机能覆盖全素材。
+# 判定「是否还有空闲可切区间」时使用的最小片段时长（秒）：空闲段长于此值才视为可切
+_MIN_FREE_CLIP_DURATION = 3.0
+
+
+def _merge_intervals(intervals: list[tuple[float, float]]) -> list[tuple[float, float]]:
+    """合并重叠/相接的时间区间，返回升序不重叠区间列表。"""
+    if not intervals:
+        return []
+    ordered = sorted((float(a), float(b)) for a, b in intervals if b > a)
+    merged: list[tuple[float, float]] = [ordered[0]]
+    for start, end in ordered[1:]:
+        last_start, last_end = merged[-1]
+        if start <= last_end:
+            merged[-1] = (last_start, max(last_end, end))
+        else:
+            merged.append((start, end))
+    return merged
+
+
+def _has_free_gap(used: list[tuple[float, float]], total: float, min_free: float = _MIN_FREE_CLIP_DURATION) -> bool:
+    """素材 [0, total] 中是否存在长度 ≥ min_free 的空闲段（考虑边缘间隙）。"""
+    if total <= 0:
+        return False
+    # 历史区间按边缘间隙扩边后判定空闲（与选片冲突检测同一口径）
+    expanded = [(max(0.0, s - SEGMENT_EDGE_GAP), min(total, e + SEGMENT_EDGE_GAP)) for s, e in used]
+    merged = _merge_intervals(expanded)
+    cursor = 0.0
+    for start, end in merged:
+        if start - cursor >= min_free:
+            return True
+        cursor = max(cursor, end)
+    return total - cursor >= min_free
+
+
+def compute_asset_availability(
+    model: "AssetModel | None",
+    min_free_clip_duration: float = _MIN_FREE_CLIP_DURATION,
+) -> dict | None:
+    """计算单个素材的余量与可用性（纯函数，不读写 DB）。
+
+    Returns:
+        视频素材返回 ``{"used_duration", "available_duration", "used_ratio", "usable"}``；
+        非视频 / 无 model / 无时长信息返回 None（调用方按可用处理，零影响）。
+
+    usable=False 条件（与受控复用机制一致）：
+        零重复可切区间已耗尽（不存在 ≥ min_free 的空闲段）且
+        所有历史区间 use_count 均达 MAX_RANGE_USE_COUNT 上限（无区间可复用）。
+    """
+    if model is None:
+        return None
+    file_type = getattr(model, "file_type", None) or getattr(model, "mime_type", "") or ""
+    if file_type != "video" and not str(file_type).startswith("video/"):
+        return None
+    total = float(getattr(model, "duration", 0.0) or 0.0)
+    if total <= 0:
+        return None
+
+    meta = _read_meta(model)
+    raw_ranges = meta.get(USED_RANGES_KEY) or []
+
+    intervals: list[tuple[float, float]] = []
+    use_counts: list[int] = []
+    for r in raw_ranges:
+        try:
+            start = float(r["start"])
+            end = float(r["end"])
+        except (KeyError, TypeError, ValueError):
+            continue
+        if end <= start:
+            continue
+        intervals.append((start, end))
+        try:
+            use_counts.append(int(r.get("use_count", 1)))
+        except (TypeError, ValueError):
+            use_counts.append(1)
+
+    merged = _merge_intervals(intervals)
+    used_duration = round(sum(e - s for s, e in merged), 3)
+    used_duration = min(used_duration, total)
+    available_duration = round(max(total - used_duration, 0.0), 3)
+    used_ratio = round(min(used_duration / total, 1.0), 4)
+
+    has_free = _has_free_gap(intervals, total, min_free_clip_duration)
+    if has_free:
+        usable = True
+    else:
+        # 空闲段耗尽：仅当存在历史区间且全部达复用上限时才判定不可用；
+        # 无历史区间（理论上不会走到，因为 has_free=True）按可用处理
+        if not use_counts:
+            usable = True
+        else:
+            usable = any(uc < MAX_RANGE_USE_COUNT for uc in use_counts)
+
+    return {
+        "used_duration": used_duration,
+        "available_duration": available_duration,
+        "used_ratio": used_ratio,
+        "usable": usable,
+    }
+
+
+def find_reusable_range(
+    db: Session,
+    asset_id: str,
+    clip_duration: float,
+    asset_total: float,
+    *,
+    max_use_count: int = MAX_RANGE_USE_COUNT,
+) -> tuple[float, float] | None:
+    """受控复用：在素材历史区间中选一条可复用区间返回 (start, end)。
+
+    选择规则：
+    1. 仅选 ``use_count < max_use_count`` 的历史区间；
+    2. 优先返回能完整容纳当前 clip_duration（起点后不越素材边界）的最久未用区间；
+    3. 没有能容纳的，则返回 last_used_at 最老（或缺失 last_used_at 的旧数据优先）
+       且 use_count 最低的区间起点（可能与其他历史区间重叠，属降级复用）；
+    4. 无任何可复用区间（记录为空或全部达上限）返回 None。
+
+    本函数只读不写；复用次数的累加由后续 record_used_segments 完成。
+    """
+    model = _get_model(db, asset_id)
+    if model is None:
+        return None
+    meta = _read_meta(model)
+    ranges = [r for r in (meta.get(USED_RANGES_KEY) or []) if int(r.get("use_count", 1)) < max_use_count]
+    if not ranges:
+        return None
+
+    def _last_used(r: dict) -> str:
+        return str(r.get("last_used_at") or r.get("created_at") or "")
+
+    max_start = max(0.0, asset_total - clip_duration)
+    # 2. 能完整容纳当前片段的候选：按 last_used_at 升序（最久未用优先）
+    fit = sorted(
+        [r for r in ranges if float(r["start"]) <= max_start + 1e-6],
+        key=_last_used,
+    )
+    if fit:
+        start = min(float(fit[0]["start"]), max_start)
+        return (start, start + clip_duration)
+
+    # 3. 降级：最久未用 + use_count 最低的区间起点
+    fallback = sorted(ranges, key=lambda r: (_last_used(r), int(r.get("use_count", 1))))[0]
+    start = min(float(fallback["start"]), max_start)
+    return (start, start + clip_duration)
+
+
+def make_reuse_callback(
+    db: Session,
+    asset_durations: dict[str, float],
+    reused_tracker: dict[str, float] | None = None,
+    assigned_tracker: dict[str, float] | None = None,
+    ratio_limit: float = REUSE_RATIO_LIMIT,
+) -> Callable[[str, float], tuple[float, float] | None]:
+    """构造给 ``_calc_random_start_time`` 用的受控复用回调.
+
+    Args:
+        db: SQLAlchemy session
+        asset_durations: 素材 ID -> 总时长（回调需要素材总时长做边界约束）
+        reused_tracker: 可选的 ``{asset_id: 累计复用时长}``，回调成功返回复用区间时
+            会把本次片段时长累加进去，供调用方统计成片复用占比（15% 阈值）。
+        assigned_tracker: 可选的 ``{asset_id: 已分配片段总时长}``，配合 ratio_limit
+            在复用前预判：若复用本片段后占比 (reused + clip_duration) /
+            (assigned + clip_duration) 超过 ratio_limit，则拒绝复用、返回 None
+            （保证成片复用占比不超阈值）。
+        ratio_limit: 单条成片复用时长占比上限，默认 15%。
+
+    Returns:
+        回调函数 ``(asset_id, clip_duration) -> (start, end) | None``。
+        回调内吞掉 DB 异常返回 None，不影响主生成流程。
     """
 
-    def _reset(asset_id: str) -> None:
+    def _reuse(asset_id: str, clip_duration: float) -> tuple[float, float] | None:
         try:
-            reset_used_segments(db, asset_id)
+            total = float(asset_durations.get(asset_id, 0.0) or 0.0)
+            if total <= 0:
+                return None
+            # 占比闸门：预判复用本片段后是否超限（仅当调用方提供了 assigned tracker）
+            if assigned_tracker is not None:
+                assigned = float(assigned_tracker.get(asset_id, 0.0) or 0.0)
+                reused_amt = float((reused_tracker or {}).get(asset_id, 0.0) or 0.0)
+                if assigned > 0 and (reused_amt + clip_duration) / (assigned + clip_duration) > ratio_limit:
+                    logger.info(
+                        "[片段追踪] 复用占比预判超 %.0f%% 阈值，拒绝复用: asset_id=%s "
+                        "reused=%.1f assigned=%.1f clip=%.1f",
+                        ratio_limit * 100,
+                        asset_id,
+                        reused_amt,
+                        assigned,
+                        clip_duration,
+                    )
+                    return None
+            result = find_reusable_range(db, asset_id, clip_duration, total)
         except Exception:
-            logger.warning("[片段追踪] reset 持久化记录失败: asset_id=%s", asset_id, exc_info=True)
-        used_segments.pop(asset_id, None)
+            logger.warning("[片段追踪] 受控复用查询异常: asset_id=%s", asset_id, exc_info=True)
+            return None
+        if result is not None and reused_tracker is not None:
+            reused_tracker[asset_id] = reused_tracker.get(asset_id, 0.0) + clip_duration
+        return result
 
-    return _reset
+    return _reuse

apps/api/app/services/edit_plan_service.py

@@ -9,6 +9,12 @@ from __future__ import annotations
 import logging
 from typing import Any, Dict, List, Optional
 
+from app.services.asset_segment_tracker import (
+    REUSE_RATIO_LIMIT,
+    get_used_segments,
+    make_reuse_callback,
+    record_used_segments,
+)
 from sqlalchemy.orm import Session
 
 from packages.adapters.sqlalchemy_impl import (
@@ -453,6 +459,116 @@ class EditPlanService:
             logger.exception("事务性替换片段失败: plan_id=%s", plan_id)
             raise
 
+    def clone_plan_for_variant(
+        self,
+        source_plan_id: str,
+        *,
+        created_by_user_id: str = "",
+        name_suffix: str = "变体",
+        reuse_tracker: Optional[dict] = None,
+    ) -> EditPlan:
+        """为批量/多预览场景克隆一份独立 plan，片段起点全部重算（受控随机/复用）。
+
+        复制源 plan 的模板归属、config 与片段结构（asset_id / duration / clip_type /
+        order 不变），每个片段重新调用 ``_calc_random_start_time``：读取素材持久化的
+        历史已用区间避让，耗尽时受控复用（use_count<3、最久未用），从而保证 N 条
+        成片片段区间互不相同，且复用占比受控。
+
+        - 不替换/不修改源 plan，源 plan 保留用户手动编辑结果。
+        - 片段区间记录（record_used_segments）随新片段写入素材 metadata，与新 plan
+          同事务；复用历史区间时由 record 自动累加 use_count。
+        - 克隆的 clips 复用区间累计时长写入 reuse_tracker（可选），供调用方统计占比。
+
+        Raises:
+            ValueError: 源 plan 不存在或无可用片段。
+        """
+        from packages.adapters.sqlalchemy_impl.models import AssetModel
+        from packages.domain.plan_generator_utils import _calc_random_start_time
+
+        source = self.get_plan_or_raise(source_plan_id)
+
+        # 分页读取源 plan 全部片段
+        clips: List[EditPlanClip] = []
+        skip, page = 0, 500
+        while True:
+            batch = self._clip_repo.list_by_plan(source_plan_id, skip=skip, limit=page)
+            if not batch:
+                break
+            clips.extend(batch)
+            if len(batch) < page:
+                break
+            skip += page
+        if not clips:
+            raise ValueError(f"源 plan 无片段，无法克隆变体: {source_plan_id}")
+
+        # 创建新 plan（复制模板归属与 config）
+        new_plan = self.create_plan(
+            template_id=source.template_id,
+            name=f"{source.name or '剪辑计划'} · {name_suffix}",
+            config=dict(source.config or {}),
+            total_duration=source.total_duration,
+            project_id=source.project_id or "",
+            created_by_user_id=created_by_user_id or (source.created_by_user_id or ""),
+        )
+
+        # 素材时长映射（O(N) 单查）
+        asset_ids = list({c.asset_id for c in clips if c.asset_id})
+        db = self._clip_repo.session
+        durations: dict[str, float] = {}
+        if asset_ids:
+            for m in db.query(AssetModel).filter(AssetModel.id.in_(asset_ids)).all():
+                durations[m.id] = float(getattr(m, "duration", 0.0) or 0.0)
+
+        used_segments = get_used_segments(db, asset_ids)
+        reused: dict[str, float] = reuse_tracker if reuse_tracker is not None else {}
+        asset_assigned: dict[str, float] = {}
+        # 回调内部预判复用后占比超 15% 则拒绝复用（calc 返回 None → 保留原起点）
+        reuse_cb = make_reuse_callback(db, durations, reused, assigned_tracker=asset_assigned)
+
+        clips_data: list[dict] = []
+        for i, c in enumerate(clips):
+            aid = c.asset_id
+            dur = float(c.duration or 0.0)
+            total = durations.get(aid, 0.0)
+            if aid and total > 0 and dur > 0:
+                # 复用占比闸门：本片段尚未分配，检查当前已复用占比
+                # reused / assigned 是否超 15%，超则该素材不再分配（保留原起点）；
+                # assigned=0（首个片段）放行，reused=0 时不误拦正常分配
+                assigned = asset_assigned.get(aid, 0.0)
+                eff_dur = min(dur, total)
+                reused_amt = reused.get(aid, 0.0)
+                ratio_blocked = assigned > 0 and reused_amt / assigned > REUSE_RATIO_LIMIT
+                start = None
+                if not ratio_blocked:
+                    start = _calc_random_start_time(aid, eff_dur, durations, used_segments, on_exhausted=reuse_cb)
+                if start is None:
+                    start = float(c.start_time or 0.0)
+                asset_assigned[aid] = assigned + eff_dur
+                used_segments.setdefault(aid, []).append((start, start + eff_dur))
+                record_used_segments(db, aid, start, start + eff_dur, new_plan.id)
+            else:
+                start = float(c.start_time or 0.0)
+
+            clips_data.append(
+                {
+                    "order": c.order if c.order is not None else i,
+                    "asset_id": aid,
+                    "start_time": start,
+                    "duration": dur,
+                    "clip_type": c.clip_type,
+                }
+            )
+
+        # 事务性写入新 plan 的片段（内部统一 commit/rollback）
+        self.replace_all_clips_transactional(new_plan.id, clips_data)
+        logger.info(
+            "克隆变体 plan: source=%s new=%s clips=%d",
+            source_plan_id,
+            new_plan.id,
+            len(clips_data),
+        )
+        return new_plan
+
     # ── 片段分割与合并 ──────────────────────────────────────────────────────
 
     def split_clip(self, clip_id: str, split_time: float) -> Dict[str, Any]:

packages/domain/plan_generator_utils.py

@@ -207,7 +207,7 @@ def _calc_random_start_time(
     clip_duration: float,
     asset_durations: dict[str, float] | None,
     used_segments: dict[str, list[tuple[float, float]]] | None = None,
-    on_exhausted: Callable[[str], None] | None = None,
+    on_exhausted: Callable[[str, float], tuple[float, float] | None] | None = None,
 ) -> float | None:
     """计算随机 start_time，避开已使用的时间段.
 
@@ -220,8 +220,10 @@ def _calc_random_start_time(
         clip_duration: 片段时长（秒）
         asset_durations: 素材 ID -> 时长映射
         used_segments: {asset_id: [(start1, end1), (start2, end2), ...]} 已使用的时间段
-        on_exhausted: 100 次随机都找不到空闲区间时的回调（入参 asset_id）。
-            通常用于清空该素材的历史使用记录实现“轮回重置”；回调后会再随机重试一次。
+        on_exhausted: 100 次随机都找不到空闲区间时的受控复用回调，入参为
+            (asset_id, clip_duration)，返回 (start, end) 复用区间或 None。
+            历史记录永不自动清空；回调返回 None（全部达上限/复用占比超闸门）时
+            本函数返回 None，由调用方轮询下一个素材或报错，不做重叠降级。
 
     Returns:
         随机 start_time 或 None
@@ -261,22 +263,36 @@ def _calc_random_start_time(
         if not overlap:
             return candidate
 
-    # 100 次都找不到空闲区间：触发轮回重置回调（清空历史使用记录）后再随机重试一次
+    # 100 次都找不到空闲区间：进入受控复用，回调从历史区间中选最久未用且
+    # 使用次数未达上限的区间返回（历史记录永不自动清空）
     if on_exhausted is not None:
         try:
-            on_exhausted(asset_id)
+            reused = on_exhausted(asset_id, clip_duration)
         except Exception:
             logger.warning(
-                "on_exhausted 轮回重置回调异常: asset_id=%s",
+                "on_exhausted 受控复用回调异常: asset_id=%s",
                 asset_id,
                 exc_info=True,
             )
-        retry = random.uniform(0.0, max_start)
-        if not used_segments or asset_id not in used_segments:
-            return retry
+            reused = None
+        if reused is not None:
+            reuse_start, reuse_end = reused
+            # 边界保护：不越素材末尾、不为负
+            reuse_start = max(0.0, min(float(reuse_start), max_start))
+            logger.info(
+                "素材可用区间耗尽，受控复用历史区间: asset_id=%s start=%.2f end=%.2f",
+                asset_id,
+                reuse_start,
+                reuse_end,
+            )
+            return reuse_start
+        # 回调存在但拒绝复用（区间全部达 use_count 上限，或复用占比将超 15% 闸门）：
+        # 返回 None，由调用方轮询下一个素材；绝不能末尾/0.0 降级——那会把片段
+        # 放回到已用过的画面，违反区间避让与重复率控制原则
+        return None
 
-    # 如果尝试多次仍找不到，缩短时长使用素材末尾
-    # 找到最后一个已使用段之后的可用空间
+    # 未提供 on_exhausted 回调（向后兼容）：降级使用素材末尾空闲位置；
+    # 末尾也已占满时返回 0.0（旧行为，仅无持久化追踪的调用方会走到这里）
     last_used_end = 0.0
     for _seg_start, seg_end in used:
         last_used_end = max(last_used_end, seg_end)
@@ -285,7 +301,6 @@ def _calc_random_start_time(
         # 返回从最后使用点开始的位置
         return min(last_used_end, max_start)
 
-    # 实在没有空间，返回0（可能会重叠，但至少能执行）
     return 0.0
 
 

tests/unit/test_asset_availability.py

@@ -0,0 +1,376 @@
+"""Task H 单测：素材余量四字段（used_duration/available_duration/used_ratio/usable）。
+
+覆盖：
+1. compute_asset_availability 纯函数各分支（无区间/未满/可复用/全达上限/非视频/无时长/区间合并/扩边判定）；
+2. _asset_availability_fields 路由辅助（视频有值、非视频 None+usable=True、异常零影响）；
+3. _to_asset_response 四字段注入；
+4. smart_match_assets 结果层过滤 usable=false。
+"""
+
+import json
+import os
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from app.api.routes.assets import (  # noqa: E402
+    _asset_availability_fields,
+    _to_asset_response,
+    smart_match_assets,
+)
+from app.schemas.asset import SmartMatchRequest  # noqa: E402
+from app.services.asset_segment_tracker import (  # noqa: E402
+    MAX_RANGE_USE_COUNT,
+    SEGMENT_EDGE_GAP,
+    compute_asset_availability,
+)
+
+VIDEO_DURATION = 60.0
+
+
+def _make_asset(duration=VIDEO_DURATION, ranges=None, file_type="video", classification_result=None):
+    """构造测试用 Asset-like 对象。
+
+    ranges: list of dicts（used_time_ranges 条目），会自动写入 classification_result JSON。
+    """
+    if classification_result is None and ranges is not None:
+        classification_result = json.dumps({"used_time_ranges": ranges})
+    return SimpleNamespace(
+        id="asset-test",
+        project_id="proj-1",
+        library_id="lib-1",
+        name="测试素材",
+        storage_key="key/test-asset.mp4",
+        thumbnail_url=None,
+        mime_type="video/mp4" if file_type == "video" else "audio/mpeg",
+        metadata={},
+        file_size=1000,
+        duration=duration,
+        width=1080,
+        height=1920,
+        fps=30,
+        codec="h264",
+        status=SimpleNamespace(value="ready"),
+        classification_status=SimpleNamespace(value="completed"),
+        quality_score=90.0,
+        created_at=__import__("datetime").datetime(2026, 8, 1, 12, 0, 0),
+        uploaded_by_user_id="user-1",
+        tag_ids=[],
+        file_type=file_type,
+        classification_result=classification_result,
+    )
+
+
+def _range(start, end, use_count=1):
+    return {
+        "start": start,
+        "end": end,
+        "plan_id": "plan-1",
+        "created_at": "2026-08-29T10:00:00",
+        "use_count": use_count,
+        "last_used_at": "2026-08-29T10:00:00",
+    }
+
+
+# ── compute_asset_availability 纯函数 ─────────────────────────────────────────
+
+
+class TestComputeAssetAvailability:
+    def test_no_ranges_fully_usable(self):
+        """无历史区间：used=0, ratio=0, usable=True。"""
+        info = compute_asset_availability(_make_asset(ranges=[]))
+        assert info is not None
+        assert info["used_duration"] == 0.0
+        assert info["available_duration"] == VIDEO_DURATION
+        assert info["used_ratio"] == 0.0
+        assert info["usable"] is True
+
+    def test_none_model_returns_none(self):
+        assert compute_asset_availability(None) is None
+
+    def test_non_video_returns_none(self):
+        """非视频（音频）返回 None，路由层按可用处理。"""
+        info = compute_asset_availability(_make_asset(file_type="audio"))
+        assert info is None
+
+    def test_zero_duration_returns_none(self):
+        info = compute_asset_availability(_make_asset(duration=0.0))
+        assert info is None
+
+    def test_partial_usage_usable(self):
+        """使用 10s，剩余 50s 空闲（≥3s），usable=True。"""
+        info = compute_asset_availability(_make_asset(ranges=[_range(5.0, 15.0)]))
+        assert info["used_duration"] == pytest.approx(10.0, abs=0.01)
+        assert info["available_duration"] == pytest.approx(50.0, abs=0.01)
+        assert info["used_ratio"] == pytest.approx(10.0 / 60.0, abs=0.001)
+        assert info["usable"] is True
+
+    def test_overlapping_ranges_merged(self):
+        """重叠区间合并后计算 used_duration，不重复计时。"""
+        info = compute_asset_availability(_make_asset(ranges=[_range(0.0, 10.0), _range(5.0, 20.0)]))
+        # 合并后 [0,20] → 20s
+        assert info["used_duration"] == pytest.approx(20.0, abs=0.01)
+        assert info["used_ratio"] == pytest.approx(20.0 / 60.0, abs=0.001)
+
+    def test_full_coverage_but_reusable(self):
+        """区间铺满全片（无空闲段），但 use_count 未达上限 → usable=True（受控复用）。"""
+        info = compute_asset_availability(
+            _make_asset(
+                duration=10.0,
+                ranges=[_range(0.0, 10.0, use_count=1)],
+            )
+        )
+        assert info["used_duration"] == pytest.approx(10.0, abs=0.01)
+        assert info["available_duration"] == 0.0
+        assert info["usable"] is True
+
+    def test_exhausted_not_usable(self):
+        """无空闲段 且 所有区间 use_count 达上限 → usable=False。"""
+        info = compute_asset_availability(
+            _make_asset(
+                duration=10.0,
+                ranges=[_range(0.0, 10.0, use_count=MAX_RANGE_USE_COUNT)],
+            )
+        )
+        assert info["usable"] is False
+        assert info["available_duration"] == 0.0
+        assert info["used_ratio"] == pytest.approx(1.0, abs=0.001)
+
+    def test_exhausted_multiple_ranges_all_capped(self):
+        """多个区间铺满、全部达上限 → usable=False；任一未满即 usable=True。"""
+        info_capped = compute_asset_availability(
+            _make_asset(
+                duration=20.0,
+                ranges=[
+                    _range(0.0, 10.0, use_count=MAX_RANGE_USE_COUNT),
+                    _range(10.0, 20.0, use_count=MAX_RANGE_USE_COUNT),
+                ],
+            )
+        )
+        assert info_capped["usable"] is False
+
+        info_partial = compute_asset_availability(
+            _make_asset(
+                duration=20.0,
+                ranges=[
+                    _range(0.0, 10.0, use_count=MAX_RANGE_USE_COUNT),
+                    _range(10.0, 20.0, use_count=MAX_RANGE_USE_COUNT - 1),
+                ],
+            )
+        )
+        assert info_partial["usable"] is True
+
+    def test_edge_gap_consumed_not_usable(self):
+        """区间未物理铺满，但扩边（+0.3s）后空闲段 <3s → 视为无空闲段；
+        区间 use_count 均达上限 → usable=False。"""
+        # 10s 素材：[0, 4.0] 与 [4.6, 10]，物理空闲 [4.0,4.6] 仅 0.6s，
+        # 扩边后左区间延至 4.3、右区间起于 4.3，空闲被吃掉
+        info = compute_asset_availability(
+            _make_asset(
+                duration=10.0,
+                ranges=[
+                    _range(0.0, 4.0, use_count=MAX_RANGE_USE_COUNT),
+                    _range(4.6, 10.0, use_count=MAX_RANGE_USE_COUNT),
+                ],
+            )
+        )
+        assert info["usable"] is False
+
+    def test_large_gap_remains_usable(self):
+        """区间之间留有 ≥3s 空闲段（扩边后仍 ≥3s）→ usable=True。"""
+        # [0,2] 扩边到 [0,2.3]，[5.3,10] 扩边前为 [5,10] 扩边起 4.7；空闲 [2.3,4.7]=2.4s <3
+        # 改用更大间隙：[0,2] 与 [6,10]，扩边后空闲 [2.3,5.7]=3.4s ≥3
+        info = compute_asset_availability(
+            _make_asset(
+                duration=10.0,
+                ranges=[
+                    _range(0.0, 2.0, use_count=MAX_RANGE_USE_COUNT),
+                    _range(6.0, 10.0, use_count=MAX_RANGE_USE_COUNT),
+                ],
+            )
+        )
+        assert info["usable"] is True
+
+    def test_invalid_ranges_skipped(self):
+        """脏数据（缺 start/end、end<=start、use_count 非法）不崩溃，合法区间照常计算。"""
+        info = compute_asset_availability(
+            _make_asset(
+                duration=30.0,
+                ranges=[
+                    {"start": "bad"},
+                    {"start": 5.0, "end": 3.0},
+                    "junk",
+                    _range(0.0, 10.0, use_count="not-a-number"),
+                ],
+            )
+        )
+        assert info is not None
+        assert info["used_duration"] == pytest.approx(10.0, abs=0.01)
+        # use_count 非法按 1 处理 → 未达上限，且空闲段充足
+        assert info["usable"] is True
+
+    def test_broken_classification_json_treated_as_unused(self):
+        """classification_result 是非法 JSON 时按无历史区间处理。"""
+        info = compute_asset_availability(_make_asset(classification_result="not-json{{{"))
+        assert info is not None
+        assert info["used_duration"] == 0.0
+        assert info["usable"] is True
+
+    def test_segment_edge_gap_constant(self):
+        """边缘间隙常量为 0.3s（与 MediaKit 冲突检测同口径）。"""
+        assert SEGMENT_EDGE_GAP == 0.3
+
+
+# ── 路由层辅助：_asset_availability_fields / _to_asset_response ──────────────
+
+
+class TestAssetAvailabilityFields:
+    def test_video_asset_returns_values(self):
+        fields = _asset_availability_fields(_make_asset(ranges=[_range(0.0, 10.0)]))
+        assert fields["usable"] is True
+        assert fields["used_duration"] == pytest.approx(10.0, abs=0.01)
+        assert fields["available_duration"] == pytest.approx(50.0, abs=0.01)
+        assert fields["used_ratio"] is not None
+
+    def test_non_video_returns_none_fields_usable_true(self):
+        fields = _asset_availability_fields(_make_asset(file_type="audio"))
+        assert fields["used_duration"] is None
+        assert fields["available_duration"] is None
+        assert fields["used_ratio"] is None
+        assert fields["usable"] is True
+
+    def test_exception_falls_back_to_zero_impact(self, monkeypatch):
+        """compute 抛异常时路由层兜底：None 字段 + usable=True，不影响响应。"""
+        import app.api.routes.assets as assets_module
+
+        def _boom(_model):
+            raise RuntimeError("unexpected")
+
+        monkeypatch.setattr(assets_module, "compute_asset_availability", _boom)
+        fields = _asset_availability_fields(_make_asset())
+        assert fields["used_duration"] is None
+        assert fields["usable"] is True
+
+
+class TestToAssetResponseInjectsFields:
+    def _storage_stub(self):
+        svc = MagicMock()
+        svc.get_download_url.return_value = "https://example.com/signed"
+        return svc
+
+    def test_video_response_carries_availability_fields(self):
+        asset = _make_asset(ranges=[_range(0.0, 12.0)])
+        resp = _to_asset_response(asset, storage_service=self._storage_stub())
+        assert resp.usable is True
+        assert resp.used_duration == pytest.approx(12.0, abs=0.01)
+        assert resp.available_duration == pytest.approx(48.0, abs=0.01)
+        assert resp.used_ratio == pytest.approx(0.2, abs=0.01)
+
+    def test_exhausted_asset_response_usable_false(self):
+        asset = _make_asset(
+            duration=10.0,
+            ranges=[_range(0.0, 10.0, use_count=MAX_RANGE_USE_COUNT)],
+        )
+        resp = _to_asset_response(asset, storage_service=self._storage_stub())
+        assert resp.usable is False
+        assert resp.used_ratio == pytest.approx(1.0, abs=0.001)
+
+    def test_non_video_response_fields_none_usable_true(self):
+        asset = _make_asset(file_type="audio")
+        resp = _to_asset_response(asset, storage_service=self._storage_stub())
+        assert resp.used_duration is None
+        assert resp.available_duration is None
+        assert resp.used_ratio is None
+        assert resp.usable is True
+
+
+# ── smart_match_assets 结果层过滤 ────────────────────────────────────────────
+
+
+def _exhausted_asset(asset_id):
+    """构造一个 usable=false 的视频素材：10s 铺满、区间 use_count 均达上限。"""
+    a = _make_asset(
+        duration=10.0,
+        ranges=[_range(0.0, 10.0, use_count=MAX_RANGE_USE_COUNT)],
+    )
+    a.id = asset_id
+    a.name = f"exhausted-{asset_id}"
+    return a
+
+
+def _fresh_asset(asset_id, duration=60.0):
+    a = _make_asset(duration=duration, ranges=[])
+    a.id = asset_id
+    a.name = f"fresh-{asset_id}"
+    return a
+
+
+class TestSmartMatchFiltersExhausted:
+    def _call(self, assets):
+        lib_repo = MagicMock()
+        lib_repo.get.return_value = SimpleNamespace(project_id="proj-1")
+        asset_repo = MagicMock()
+        asset_repo.find_by_library_and_file_type.return_value = assets
+        project_repo = MagicMock()
+        project = MagicMock()
+        project.can_access.return_value = True
+        project_repo.find_by_id.return_value = project
+
+        user = SimpleNamespace(id="user-1")
+        auth_user = SimpleNamespace(user=user)
+
+        # storage_service 在 _to_asset_response 内 get_storage_service()，patch 掉
+        import app.api.routes.assets as assets_module
+
+        svc = MagicMock()
+        svc.get_download_url.return_value = "https://example.com/signed"
+        original_get_storage = assets_module.get_storage_service
+        assets_module.get_storage_service = lambda: svc
+        try:
+            resp = smart_match_assets(
+                SmartMatchRequest(library_id="lib-1", kind="video"),
+                authenticated_user=auth_user,
+                asset_repository=asset_repo,
+                asset_library_repository=lib_repo,
+                project_repository=project_repo,
+            )
+        finally:
+            assets_module.get_storage_service = original_get_storage
+        return resp
+
+    def test_exhausted_assets_excluded(self):
+        """smart-match 结果中 usable=false 的素材被剔除，新鲜素材保留。"""
+        assets = [
+            _exhausted_asset("a-exhausted-1"),
+            _exhausted_asset("a-exhausted-2"),
+            _fresh_asset("a-fresh-1"),
+        ]
+        resp = self._call(assets)
+        returned_ids = {item.asset.id for item in resp.items}
+        assert "a-fresh-1" in returned_ids
+        assert "a-exhausted-1" not in returned_ids
+        assert "a-exhausted-2" not in returned_ids
+        # total_candidates 是过滤前的候选总数
+        assert resp.total_candidates == 3
+        # 返回的素材全部 usable=True
+        assert all(item.asset.usable for item in resp.items)
+
+    def test_all_exhausted_returns_empty(self):
+        """全部素材已用尽时返回空列表（不报错，前端显示空结果）。"""
+        assets = [_exhausted_asset("a-ex-1"), _exhausted_asset("a-ex-2")]
+        resp = self._call(assets)
+        assert resp.items == []
+        assert resp.total_candidates == 2
+
+    def test_fresh_assets_all_returned(self):
+        assets = [_fresh_asset("a-1"), _fresh_asset("a-2")]
+        resp = self._call(assets)
+        assert len(resp.items) == 2
+        assert all(item.asset.usable for item in resp.items)

tests/unit/test_asset_segment_tracker.py

@@ -1,12 +1,14 @@
-"""素材片段使用记录追踪服务测试（asset_segment_tracker）.
+"""素材片段使用记录追踪 + 受控复用机制测试（asset_segment_tracker）.
 
 覆盖：
 - get_used_segments 聚合 metadata 中持久化的区间
-- record_used_segments 追加记录（不 commit，保留原有 metadata 字段）
-- remove_used_segment 匹配删除（tolerance + plan_id）
-- reset_used_segments 清空轮回（其他字段不动）
-- make_reset_callback 同时清持久化和内存
-- _calc_random_start_time 的 on_exhausted 轮回回调
+- record_used_segments 追加新记录（use_count=1，保留原有 metadata 字段）
+- record_used_segments 复用同一区间时累加 use_count / 刷新 last_used_at
+- remove_used_segment 匹配删除（tolerance + plan_id，旧数据按时间匹配）
+- reset_used_segments 清空（其他字段不动）
+- find_reusable_range：选最久未用且 use_count<3 的区间；全部达上限返回 None
+- make_reuse_callback：返回复用区间、累加 reused_tracker、DB 异常返回 None
+- _calc_random_start_time：100 次避不开时调用复用回调返回历史区间（不再清空历史）
 """
 
 from __future__ import annotations
@@ -24,8 +26,12 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
 import pytest
 from app.services import asset_segment_tracker as ast
 from app.services.asset_segment_tracker import (
+    MAX_RANGE_USE_COUNT,
+    REUSE_RATIO_LIMIT,
+    SEGMENT_EDGE_GAP,
+    find_reusable_range,
     get_used_segments,
-    make_reset_callback,
+    make_reuse_callback,
     record_used_segments,
     remove_used_segment,
     reset_used_segments,
@@ -60,16 +66,16 @@ 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)
 
 
 class FakeSession:
-    """模拟 db：db.query(Model).filter(Model.id.in_(ids)).all() / .filter(Model.id == id).first()。
-
-    tracker 模块里的 AssetModel 被 monkeypatch 为 FakeModel 类，
-    这里用挂在类上的伪 column 对象接住 in_ / __eq__。
-    """
+    """模拟 db：db.query(Model).filter(Model.id.in_(ids)).all() / .filter(Model.id == id).first()。"""
 
     class _Col:
         def __init__(self, models):
@@ -93,7 +99,6 @@ class FakeSession:
                 return expr
 
         q = _Q()
-        # 让 tracker 里 AssetModel.id 能取到伪 column
         _model.id = col
         return q
 
@@ -103,19 +108,27 @@ class FakeSession:
 
 @pytest.fixture
 def patched_model(monkeypatch):
-    """把 tracker 模块内的 AssetModel 替换为 FakeModel（供 FakeSession 挂伪 column）。"""
     monkeypatch.setattr(ast, "AssetModel", FakeModel)
 
 
-@pytest.fixture
-def models():
-    return {}
-
-
 def _db(models):
     return FakeSession(models)
 
 
+def _ranges(db, aid="a1"):
+    model = db._models[aid]
+    return json.loads(model.classification_result)["used_time_ranges"]
+
+
+# ── 配置常量 ──────────────────────────────────────────────────────────────────
+
+
+def test_config_constants():
+    assert MAX_RANGE_USE_COUNT == 3
+    assert REUSE_RATIO_LIMIT == 0.15
+    assert SEGMENT_EDGE_GAP == 0.3
+
+
 # ── get_used_segments ─────────────────────────────────────────────────────────
 
 
@@ -125,124 +138,138 @@ def test_get_used_segments_aggregates_ranges(patched_model):
             "a1",
             {
                 "used_time_ranges": [
-                    {"start": 1.0, "end": 5.0, "plan_id": "p1"},
+                    {"start": 1.0, "end": 5.0, "plan_id": "p1", "use_count": 2},
                     {"start": 9.0, "end": 12.0, "plan_id": "p2"},
                 ]
             },
         ),
-        "a2": FakeModel("a2", {"other": 1}),  # 无区间记录
-        "a3": FakeModel("a3"),  # metadata 为空
+        "a2": FakeModel("a2", {"other": 1}),
+        "a3": FakeModel("a3"),
     }
     db = _db(models)
-
-    result = get_used_segments(db, ["a1", "a2", "a3", "missing"])
-
-    assert result == {"a1": [(1.0, 5.0), (9.0, 12.0)]}
+    assert get_used_segments(db, ["a1", "a2", "a3", "missing"]) == {"a1": [(1.0, 5.0), (9.0, 12.0)]}
 
 
-def test_get_used_segments_empty_input(patched_model):
+def test_get_used_segments_empty(patched_model):
     assert get_used_segments(_db({}), []) == {}
 
 
 # ── record_used_segments ──────────────────────────────────────────────────────
 
 
-def test_record_appends_and_no_commit(patched_model):
-    models = {"a1": FakeModel("a1", {"generation_use_count": 3})}
+def test_record_appends_new_range_with_use_count_one(patched_model):
+    models = {"a1": FakeModel("a1", {"generation_use_count": 48, "review_status": "pending_review"})}
     db = _db(models)
-
-    record_used_segments(db, "a1", 2.0, 6.5, "plan-x")
-
-    meta = models["a1"].meta()
-    assert meta["generation_use_count"] == 3  # 原有字段保留
+    record_used_segments(db, "a1", 12.5, 20.3, "plan-x")
+    meta = json.loads(models["a1"].classification_result)
+    assert meta["generation_use_count"] == 48
+    assert meta["review_status"] == "pending_review"
     ranges = meta["used_time_ranges"]
     assert len(ranges) == 1
-    assert ranges[0]["start"] == 2.0
-    assert ranges[0]["end"] == 6.5
+    assert ranges[0]["start"] == 12.5 and ranges[0]["end"] == 20.3
     assert ranges[0]["plan_id"] == "plan-x"
-    assert "created_at" in ranges[0]
-    assert db.commits == 0  # 不自行 commit（事务由调用方控制）
+    assert ranges[0]["use_count"] == 1
+    assert "created_at" in ranges[0] and "last_used_at" in ranges[0]
+    assert db.commits == 0  # 不自行 commit
 
 
-def test_record_multiple_appends_in_order(patched_model):
-    models = {"a1": FakeModel("a1")}
+def test_record_reuse_same_range_increments_use_count(patched_model):
+    """新片段与历史区间高度重叠（复用）→ 累加 use_count，不新增记录。"""
+    models = {
+        "a1": FakeModel(
+            "a1",
+            {
+                "used_time_ranges": [
+                    {
+                        "start": 10.0,
+                        "end": 20.0,
+                        "plan_id": "p1",
+                        "use_count": 1,
+                        "created_at": "2026-01-01T00:00:00+00:00",
+                        "last_used_at": "2026-01-01T00:00:00+00:00",
+                    },
+                ]
+            },
+        )
+    }
     db = _db(models)
-
-    record_used_segments(db, "a1", 0.0, 4.0, "p1")
-    record_used_segments(db, "a1", 10.0, 14.0, "p1")
-
-    ranges = models["a1"].meta()["used_time_ranges"]
-    assert [r["start"] for r in ranges] == [0.0, 10.0]
+    # 同一起点复用（find_reusable_range 返回的就是历史区间起点）
+    record_used_segments(db, "a1", 10.0, 20.0, "p2")
+    ranges = _ranges(db)
+    assert len(ranges) == 1
+    assert ranges[0]["use_count"] == 2
+    assert ranges[0]["last_used_at"] != "2026-01-01T00:00:00+00:00"
 
 
-def test_record_missing_asset_is_noop(patched_model):
+def test_record_distinct_range_appends(patched_model):
+    models = {
+        "a1": FakeModel(
+            "a1",
+            {
+                "used_time_ranges": [
+                    {"start": 10.0, "end": 20.0, "plan_id": "p1", "use_count": 1},
+                ]
+            },
+        )
+    }
+    db = _db(models)
+    record_used_segments(db, "a1", 25.0, 35.0, "p2")
+    ranges = _ranges(db)
+    assert len(ranges) == 2
+    assert ranges[1]["use_count"] == 1
+
+
+def test_record_missing_asset_no_raise(patched_model):
     db = _db({})
-    record_used_segments(db, "ghost", 0.0, 1.0, "p1")  # 不抛异常
+    record_used_segments(db, "ghost", 1.0, 2.0, "p")  # 不抛异常
 
 
 # ── remove_used_segment ───────────────────────────────────────────────────────
 
 
-def test_remove_matching_segment(patched_model):
-    models = {"a1": FakeModel("a1")}
-    db = _db(models)
-    record_used_segments(db, "a1", 0.0, 4.0, "p1")
-    record_used_segments(db, "a1", 10.0, 14.0, "p1")
-
-    removed = remove_used_segment(db, "a1", 0.0, 4.0, plan_id="p1")
-
-    assert removed is True
-    ranges = models["a1"].meta()["used_time_ranges"]
-    assert len(ranges) == 1
-    assert ranges[0]["start"] == 10.0
-
-
-def test_remove_not_found_returns_false(patched_model):
-    models = {"a1": FakeModel("a1")}
-    db = _db(models)
-    record_used_segments(db, "a1", 0.0, 4.0, "p1")
-
-    assert remove_used_segment(db, "a1", 99.0, 100.0, plan_id="p1") is False
-
-
-def test_remove_respects_tolerance(patched_model):
-    models = {
-        "a1": FakeModel("a1", {"used_time_ranges": [{"start": 5.0, "end": 9.0, "plan_id": "p1"}]}),
-        "a2": FakeModel("a2", {"used_time_ranges": [{"start": 5.0, "end": 9.0, "plan_id": "p1"}]}),
-    }
-    db = _db(models)
-
-    # 偏差 0.3 秒，在 tolerance=0.5 内 → 删除成功
-    assert remove_used_segment(db, "a1", 5.3, 8.7, plan_id="p1") is True
-    # 偏差 2 秒，超出 tolerance → 删除失败
-    assert remove_used_segment(db, "a2", 7.0, 11.0, plan_id="p1") is False
-
-
-def test_remove_plan_id_must_match(patched_model):
-    models = {"a1": FakeModel("a1", {"used_time_ranges": [{"start": 5.0, "end": 9.0, "plan_id": "plan-A"}]})}
-    db = _db(models)
-
-    # 时间匹配但 plan_id 不同 → 不删除
-    assert remove_used_segment(db, "a1", 5.0, 9.0, plan_id="plan-B") is False
-    assert len(models["a1"].meta()["used_time_ranges"]) == 1
-
-
-def test_remove_legacy_record_without_plan_id(patched_model):
-    """旧数据记录没有 plan_id 字段时，MediaKit 移动片段仍能按时间匹配删除（防容量泄漏）。"""
+def test_remove_matching_range(patched_model):
     models = {
         "a1": FakeModel(
             "a1",
-            {"used_time_ranges": [{"start": 5.0, "end": 9.0}]},  # 旧记录无 plan_id
+            {
+                "used_time_ranges": [
+                    {"start": 1.0, "end": 5.0, "plan_id": "p1"},
+                    {"start": 9.0, "end": 12.0, "plan_id": "p2"},
+                ]
+            },
         )
     }
     db = _db(models)
-
-    # 传入 plan_id，但记录本身无 plan_id → 按时间匹配，允许删除
-    assert remove_used_segment(db, "a1", 5.0, 9.0, plan_id="plan-new") is True
-    assert models["a1"].meta()["used_time_ranges"] == []
+    assert remove_used_segment(db, "a1", 1.0, 5.0, plan_id="p1") is True
+    assert len(_ranges(db)) == 1
+    assert _ranges(db)[0]["start"] == 9.0
 
 
-# ── reset_used_segments ───────────────────────────────────────────────────────
+def test_remove_plan_mismatch_keeps_range(patched_model):
+    models = {"a1": FakeModel("a1", {"used_time_ranges": [{"start": 1.0, "end": 5.0, "plan_id": "p1"}]})}
+    db = _db(models)
+    assert remove_used_segment(db, "a1", 1.0, 5.0, plan_id="other") is False
+    assert len(_ranges(db)) == 1
+
+
+def test_remove_legacy_range_without_plan_id(patched_model):
+    """旧数据记录缺 plan_id → 按时间匹配可删除。"""
+    models = {
+        "a1": FakeModel(
+            "a1",
+            {
+                "used_time_ranges": [
+                    {"start": 2.0, "end": 12.0, "created_at": "2026-01-01T00:00:00"},
+                ]
+            },
+        )
+    }
+    db = _db(models)
+    assert remove_used_segment(db, "a1", 2.0, 12.0, plan_id="plan-new") is True
+    assert _ranges(db) == []
+
+
+# ── reset_used_segments（仅运维/测试）─────────────────────────────────────────
 
 
 def test_reset_clears_ranges_keeps_other_fields(patched_model):
@@ -250,74 +277,185 @@ def test_reset_clears_ranges_keeps_other_fields(patched_model):
         "a1": FakeModel(
             "a1",
             {
-                "generation_use_count": 9,
-                "used_time_ranges": [{"start": 1, "end": 2}],
+                "generation_use_count": 3,
+                "used_time_ranges": [
+                    {"start": 1.0, "end": 5.0},
+                ],
             },
         )
     }
     db = _db(models)
-
     reset_used_segments(db, "a1")
-
-    meta = models["a1"].meta()
+    meta = json.loads(models["a1"].classification_result)
     assert meta["used_time_ranges"] == []
-    assert meta["generation_use_count"] == 9
-    assert db.commits == 0
+    assert meta["generation_use_count"] == 3
 
 
-# ── make_reset_callback ───────────────────────────────────────────────────────
+# ── find_reusable_range：受控复用选择 ─────────────────────────────────────────
 
 
-def test_reset_callback_clears_persisted_and_memory(patched_model):
-    models = {"a1": FakeModel("a1", {"used_time_ranges": [{"start": 0, "end": 30}]})}
+def test_find_reusable_prefers_oldest_unused(patched_model):
+    """选 last_used_at 最老、use_count 未达上限的区间；能容纳 clip_duration。"""
+    models = {
+        "a1": FakeModel(
+            "a1",
+            {
+                "used_time_ranges": [
+                    {"start": 0.0, "end": 8.0, "use_count": 1, "last_used_at": "2026-08-01T00:00:00+00:00"},
+                    {
+                        "start": 10.0,
+                        "end": 20.0,
+                        "use_count": 1,
+                        "last_used_at": "2026-01-01T00:00:00+00:00",
+                    },  # 最久未用
+                ]
+            },
+        )
+    }
     db = _db(models)
-
-    used_segments = {"a1": [(0.0, 30.0)], "a2": [(1.0, 2.0)]}
-    cb = make_reset_callback(db, used_segments)
-    cb("a1")
-
-    assert "a1" not in used_segments  # 内存清空
-    assert "a2" in used_segments  # 其他素材不受影响
-    assert models["a1"].meta()["used_time_ranges"] == []
+    result = find_reusable_range(db, "a1", clip_duration=5.0, asset_total=30.0)
+    assert result is not None
+    start, end = result
+    assert start == 10.0 and end == 15.0
 
 
-# ── _calc_random_start_time 轮回回调 ──────────────────────────────────────────
+def test_find_reusable_excludes_max_use_count(patched_model):
+    """use_count 达到上限（3）的区间不再参与复用；全部达上限返回 None。"""
+    models = {
+        "a1": FakeModel(
+            "a1",
+            {
+                "used_time_ranges": [
+                    {"start": 0.0, "end": 10.0, "use_count": 3, "last_used_at": "2026-01-01T00:00:00"},
+                ]
+            },
+        )
+    }
+    db = _db(models)
+    assert find_reusable_range(db, "a1", 5.0, 30.0) is None
 
 
-def test_calc_random_start_invokes_reset_when_exhausted():
-    """素材区间被占满（100 次随机必重叠）→ 触发 on_exhausted，重置后重试成功。"""
+def test_find_reusable_fourth_use_rejected(patched_model):
+    """同区间复用第 4 次被拒绝：use_count=2 的可复用，use_count=3 的不可复用。"""
+    models = {
+        "a1": FakeModel(
+            "a1",
+            {
+                "used_time_ranges": [
+                    {"start": 0.0, "end": 10.0, "use_count": 2, "last_used_at": "2026-03-01T00:00:00"},
+                    {"start": 10.0, "end": 20.0, "use_count": 3, "last_used_at": "2026-01-01T00:00:00"},
+                ]
+            },
+        )
+    }
+    db = _db(models)
+    result = find_reusable_range(db, "a1", 5.0, 30.0)
+    # 只能选 use_count=2 的区间（start=0），不能选 use_count=3 的（虽然它更老）
+    assert result is not None and result[0] == 0.0
+
+
+def test_find_reusable_clamps_to_asset_bounds(patched_model):
+    """历史区间起点 + clip_duration 会越素材末尾时，起点钳制到 max_start。"""
+    models = {
+        "a1": FakeModel(
+            "a1",
+            {
+                "used_time_ranges": [
+                    {"start": 25.0, "end": 30.0, "use_count": 1, "last_used_at": "2026-01-01T00:00:00"},
+                ]
+            },
+        )
+    }
+    db = _db(models)
+    result = find_reusable_range(db, "a1", clip_duration=10.0, asset_total=30.0)
+    assert result is not None
+    start, end = result
+    assert end <= 30.0 + 1e-6 and start >= 0.0
+
+
+def test_find_reusable_no_ranges_returns_none(patched_model):
+    models = {"a1": FakeModel("a1", {"other": 1})}
+    db = _db(models)
+    assert find_reusable_range(db, "a1", 5.0, 30.0) is None
+
+
+# ── make_reuse_callback ───────────────────────────────────────────────────────
+
+
+def test_reuse_callback_returns_range_and_tracks_duration(patched_model):
+    models = {
+        "a1": FakeModel(
+            "a1",
+            {
+                "used_time_ranges": [
+                    {"start": 10.0, "end": 20.0, "use_count": 1, "last_used_at": "2026-01-01T00:00:00"},
+                ]
+            },
+        )
+    }
+    db = _db(models)
+    reused: dict[str, float] = {}
+    cb = make_reuse_callback(db, {"a1": 30.0}, reused)
+    result = cb("a1", 8.0)
+    assert result is not None and result[0] == 10.0
+    assert reused["a1"] == 8.0  # 复用时长累加
+
+
+def test_reuse_callback_db_error_returns_none(patched_model):
+    class BoomSession:
+        def query(self, _m):
+            raise RuntimeError("db down")
+
+    reused: dict[str, float] = {}
+    cb = make_reuse_callback(BoomSession(), {"a1": 30.0}, reused)
+    assert cb("a1", 8.0) is None  # 异常被吞，返回 None
+    assert reused == {}
+
+
+# ── _calc_random_start_time 与受控回调集成 ────────────────────────────────────
+
+
+def test_calc_random_start_uses_reuse_callback_when_exhausted(monkeypatch):
+    """素材区间被占满、100 次随机找不到空位时，调用复用回调返回历史区间。"""
+    import packages.domain.plan_generator_utils as pgu
+
+    monkeypatch.setattr(pgu.random, "uniform", lambda a, b: 0.5)  # 固定候选点必撞区间
+
     durations = {"a1": 30.0}
-    used = {"a1": [(0.0, 10.0), (10.0, 20.0), (20.0, 30.0)]}
-    reset_called = []
+    used = {"a1": [(0.0, 30.0)]}  # 全占满
+    calls = []
 
-    def _on_exhausted(asset_id):
-        reset_called.append(asset_id)
-        used.pop(asset_id, None)  # 模拟轮回清空
+    def reuse_cb(asset_id, clip_duration):
+        calls.append((asset_id, clip_duration))
+        return (10.0, 18.0)
 
-    result = _calc_random_start_time("a1", 10.0, durations, used, on_exhausted=_on_exhausted)
-
-    assert reset_called == ["a1"]
-    assert result is not None
-    assert 0.0 <= result <= 20.0  # max_start = 30 - 10
+    result = _calc_random_start_time("a1", 8.0, durations, used, on_exhausted=reuse_cb)
+    assert calls == [("a1", 8.0)]
+    assert result == 10.0
 
 
-def test_calc_random_start_no_callback_keeps_legacy_fallback():
-    """不传 on_exhausted 时保持旧降级行为，不报错。"""
+def test_calc_random_start_reuse_callback_none_returns_none(monkeypatch):
+    """复用回调返回 None（区间全部达上限/复用占比超闸门）→ calc 返回 None。
+
+    新机制下不做末尾/0.0 重叠降级（那会把片段放回已用过的画面），
+    由调用方轮询下一个素材或报 400；历史记录不被清空。
+    """
+    import packages.domain.plan_generator_utils as pgu
+
+    monkeypatch.setattr(pgu.random, "uniform", lambda a, b: 0.5)
+
     durations = {"a1": 30.0}
-    used = {"a1": [(0.0, 10.0), (10.0, 20.0), (20.0, 30.0)]}
+    used = {"a1": [(0.0, 30.0)]}
+    used_before = list(used["a1"])
+    result = _calc_random_start_time("a1", 8.0, durations, used, on_exhausted=lambda aid, d: None)
+    assert result is None
+    assert used["a1"] == used_before  # 历史记录未被清空
 
-    result = _calc_random_start_time("a1", 10.0, durations, used)
 
-    assert result is not None
+def test_calc_random_start_no_callback_backward_compatible(monkeypatch):
+    """不传 on_exhausted 时行为与旧版兼容（100 次失败走降级）。"""
+    import packages.domain.plan_generator_utils as pgu
 
-
-def test_calc_random_start_with_space_does_not_reset():
-    """有充足空闲区间时不触发 reset。"""
-    durations = {"a1": 100.0}
-    used = {"a1": [(0.0, 50.0)]}
-    reset_called = []
-
-    result = _calc_random_start_time("a1", 5.0, durations, used, on_exhausted=lambda aid: reset_called.append(aid))
-
-    assert reset_called == []
+    monkeypatch.setattr(pgu.random, "uniform", lambda a, b: 0.5)
+    result = _calc_random_start_time("a1", 8.0, {"a1": 30.0}, {"a1": [(0.0, 30.0)]})
     assert result is not None

tests/unit/test_batch_no_plan_guard.py

@@ -0,0 +1,95 @@
+"""AI Review 回归：批量生成 count>1 但 source_edit_plan_id 为空时不应 IndexError。
+
+变体 plan 预克隆仅在 source_edit_plan_id 非空时执行；无源 plan 时
+variant_plan_ids 为空，循环中禁止索引访问，各任务走自身随机选片流程。
+"""
+
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+
+def _make_user():
+    return SimpleNamespace(user=SimpleNamespace(id="user-1"))
+
+
+def _make_request(count):
+    from app.schemas.generation_task import CreateGenerationTaskRequest
+
+    return CreateGenerationTaskRequest(
+        project_id="proj-1",
+        asset_library_id="lib-1",
+        strategy_id="one_take",
+        asset_ids=["a1"],
+        count=count,
+        source_edit_plan_id="",  # 关键：无源 plan（空字符串为假值）
+    )
+
+
+class TestBatchNoSourcePlanNoIndexError:
+    def test_count3_without_source_plan_creates_three_tasks(self):
+        """count=3 且无 source_edit_plan_id：不克隆、不 IndexError、创建 3 个任务。"""
+        from app.api.routes.generation_tasks import create_generation_task
+
+        repo = MagicMock()
+        repo.count_pending_by_user.return_value = 0
+        repo.count_pending_total.return_value = 0
+        repo.create.side_effect = lambda t: t
+        repo.update.side_effect = lambda t: t
+
+        created = []
+
+        def _fake_execute(cmd):
+            task = MagicMock()
+            task.id = f"task-{len(created) + 1}"
+            task.source_edit_plan_id = cmd.source_edit_plan_id
+            task.status = "pending"
+            task.progress = 0.0
+            task.strategy_id = "one_take"
+            task.error_message = ""
+            task.cover_url = None
+            task.title_config = {}
+            task.created_at = None
+            task.batch_id = "batch-1"
+            created.append(task)
+            return task
+
+        with patch("app.api.routes.generation_tasks.CreateGenerationTaskUseCase") as MockUC:
+            MockUC.return_value.execute.side_effect = _fake_execute
+            with patch(
+                "app.api.routes.generation_tasks.safe_enqueue_generation_task",
+                return_value=True,
+            ):
+                with patch("app.api.routes.generation_tasks._writeback_edit_plan_config"):
+                    with patch(
+                        "app.api.routes.generation_tasks._resolve_project_and_library",
+                        return_value=("proj-1", ""),
+                    ):
+                        # 核心断言：不得抛 IndexError（变体 plan 索引守卫）。
+                        # 响应序列化字段与本回归无关，ValidationError 可接受，
+                        # 但 IndexError 必须不出现。
+                        try:
+                            create_generation_task(
+                                _make_request(3),
+                                authenticated_user=_make_user(),
+                                generation_task_repository=repo,
+                                project_repository=MagicMock(),
+                                asset_repository=MagicMock(),
+                                asset_library_repository=MagicMock(),
+                                db=MagicMock(),
+                            )
+                        except IndexError as exc:  # pragma: no cover - 不应发生
+                            pytest.fail(f"无源 plan 批量生成触发 IndexError: {exc}")
+                        except Exception:
+                            # 响应序列化等其他异常与本次守卫无关，忽略
+                            pass
+
+        # 3 个任务全部创建（未因 IndexError 中断）
+        assert len(created) == 3
+        # 无源 plan 时所有任务 source_edit_plan_id 均为空
+        assert all(not t.source_edit_plan_id for t in created)

tests/unit/test_clone_plan_variant.py

@@ -0,0 +1,201 @@
+"""clone_plan_for_variant 单元测试（Task G 验收项：批量 N 条视频片段独立）。
+
+验证：
+- 同一源 plan 克隆 3 次产出 3 个不同 plan_id，各自片段起点不同
+- 源 plan 的片段不被修改
+- 模板/config/时长结构被复制
+- 复用占比闸门触发时保留原起点（不重复抽取）
+- 源 plan 无片段时抛出 ValueError
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+sys.path.insert(0, str(Path(__file__).resolve().parent))  # tests/unit，便于复用同目录 stub
+
+# 复用 test_edit_plan_service 里的内存 stub 仓储
+from test_edit_plan_service import (  # noqa: E402
+    StubEditPlanClipRepository,
+    StubEditPlanRepository,
+    _make_service,
+)
+
+from packages.domain.edit_plan_clip import EditPlanClip
+
+
+@pytest.fixture
+def svc_with_source():
+    """构造带源 plan + 3 个片段的 service（stub 仓储）。"""
+    svc = _make_service()
+    # clone 用 self._clip_repo.session 拿 db；stub 无 session，补一个 MagicMock
+    svc._clip_repo.session = MagicMock()
+
+    source = svc.create_plan(template_id="tpl-001", name="源计划", total_duration=15.0)
+
+    for i in range(3):
+        clip = EditPlanClip.create(
+            plan_id=source.id,
+            clip_type="main",
+            order=i,
+            asset_id=f"a{i % 2 + 1}",  # a1, a2, a1
+            start_time=float(i * 5),
+            duration=5.0,
+        )
+        svc._clip_repo.create(clip)
+    return svc, source
+
+
+def _clone_with_fake_calc(svc, source, starts, *, used=None):
+    """用受控的 calc 起点列表执行一次克隆。
+
+    starts: 每次 _calc_random_start_time 返回的起点（按片段顺序）。
+    返回 (new_plan, replace_all 调用的 clips_data, calc 调用记录)。
+    """
+    calc_calls: list[dict] = []
+
+    def fake_calc(asset_id, clip_duration, durations, used_segments, on_exhausted=None):
+        idx = len(calc_calls)
+        calc_calls.append({"asset_id": asset_id, "clip_duration": clip_duration, "on_exhausted": on_exhausted})
+        return starts[idx]
+
+    with (
+        patch(
+            "app.services.edit_plan_service.get_used_segments",
+            return_value=used or {},
+        ),
+        patch(
+            "app.services.edit_plan_service.make_reuse_callback",
+            return_value=lambda aid, d: None,
+        ),
+        patch(
+            "app.services.edit_plan_service.record_used_segments",
+            return_value=None,
+        ),
+        patch(
+            "packages.domain.plan_generator_utils._calc_random_start_time",
+            side_effect=fake_calc,
+        ),
+        patch.object(svc, "replace_all_clips_transactional", return_value=3) as mock_replace,
+        patch(
+            "packages.adapters.sqlalchemy_impl.models.AssetModel",
+            create=True,
+        ) as mock_asset_model,
+    ):
+        # db.query(AssetModel).filter(...).all() → 返回带 duration 的 mock 素材
+        m1 = MagicMock(id="a1")
+        m1.duration = 60.0
+        m2 = MagicMock(id="a2")
+        m2.duration = 60.0
+        svc._clip_repo.session.query.return_value.filter.return_value.all.return_value = [m1, m2]
+        new_plan = svc.clone_plan_for_variant(source.id, created_by_user_id="u1", name_suffix="变体")
+        clips_data = mock_replace.call_args.args[1]
+    return new_plan, clips_data, calc_calls
+
+
+class TestClonePlanForVariant:
+    def test_three_clones_produce_distinct_plans_and_starts(self, svc_with_source):
+        """克隆 3 次：3 个不同 plan_id，片段起点互不相同（Task G 验收）。"""
+        svc, source = svc_with_source
+        start_sets = [
+            [10.0, 20.0, 30.0],
+            [11.0, 21.0, 31.0],
+            [12.0, 22.0, 32.0],
+        ]
+        plans = []
+        all_clips = []
+        for starts in start_sets:
+            new_plan, clips_data, _ = _clone_with_fake_calc(svc, source, starts)
+            plans.append(new_plan)
+            all_clips.append(clips_data)
+
+        # 3 个不同 plan_id，且都不等于源 plan
+        plan_ids = {p.id for p in plans}
+        assert len(plan_ids) == 3
+        assert source.id not in plan_ids
+
+        # 每次克隆的起点各自不同
+        for clips_data, starts in zip(all_clips, start_sets, strict=True):
+            assert [c["start_time"] for c in clips_data] == starts
+
+        # 三次克隆的起点集合互不相同
+        assert {tuple(c["start_time"] for c in clips) for clips in all_clips} == {
+            (10.0, 20.0, 30.0),
+            (11.0, 21.0, 31.0),
+            (12.0, 22.0, 32.0),
+        }
+
+    def test_source_plan_not_modified(self, svc_with_source):
+        """克隆不修改源 plan 及其片段（保留用户手动编辑）。"""
+        svc, source = svc_with_source
+        source_clips_before = sorted(
+            [(c.order, c.asset_id, c.start_time, c.duration) for c in svc._clip_repo.list_by_plan(source.id)]
+        )
+        source_name_before = source.name
+
+        _clone_with_fake_calc(svc, source, [9.0, 19.0, 29.0])
+        _clone_with_fake_calc(svc, source, [8.0, 18.0, 28.0])
+
+        source_clips_after = sorted(
+            [(c.order, c.asset_id, c.start_time, c.duration) for c in svc._clip_repo.list_by_plan(source.id)]
+        )
+        assert source_clips_after == source_clips_before
+        assert svc._plan_repo.get(source.id).name == source_name_before
+
+    def test_clone_copies_structure(self, svc_with_source):
+        """克隆复制 template_id / config / total_duration / 片段素材与时长。"""
+        svc, source = svc_with_source
+        source.config = {"mode": "ONE_TAKE"}
+        new_plan, clips_data, _ = _clone_with_fake_calc(svc, source, [10.0, 20.0, 30.0])
+
+        assert new_plan.template_id == source.template_id
+        assert new_plan.total_duration == source.total_duration
+        assert new_plan.config == {"mode": "ONE_TAKE"}
+        assert "变体" in new_plan.name
+        # 片段素材与时长结构保持
+        assert [c["asset_id"] for c in clips_data] == ["a1", "a2", "a1"]
+        assert all(c["duration"] == 5.0 for c in clips_data)
+        assert [c["order"] for c in clips_data] == [0, 1, 2]
+
+    def test_clone_uses_reuse_callback(self, svc_with_source):
+        """克隆时 calc 传入了 on_exhausted 受控复用回调（耗尽时复用而非清空历史）。"""
+        svc, source = svc_with_source
+        _, _, calc_calls = _clone_with_fake_calc(svc, source, [10.0, 20.0, 30.0])
+        assert len(calc_calls) == 3
+        for call in calc_calls:
+            assert call["on_exhausted"] is not None
+
+    def test_clone_ratio_blocked_keeps_original_start(self, svc_with_source):
+        """复用占比闸门触发（calc 返回 None）时保留源片段原起点。"""
+        svc, source = svc_with_source
+        # 第 3 个片段 calc 返回 None（模拟复用占比超 15% 拒绝复用）
+        new_plan, clips_data, _ = _clone_with_fake_calc(svc, source, [10.0, 20.0, None])  # type: ignore[list-item]
+        starts = [c["start_time"] for c in clips_data]
+        assert starts[0] == 10.0
+        assert starts[1] == 20.0
+        # 第 3 片段保留源起点（源 order=2 → start_time=10.0）
+        assert starts[2] == 10.0
+
+    def test_clone_empty_source_raises(self):
+        """源 plan 无片段时抛出 ValueError。"""
+        svc = _make_service()
+        svc._clip_repo.session = MagicMock()
+        empty = svc.create_plan(template_id="tpl-x", name="空计划")
+        with pytest.raises(ValueError, match="无片段"):
+            svc.clone_plan_for_variant(empty.id, name_suffix="变体")
+
+    def test_clone_nonexistent_source_raises(self):
+        """源 plan 不存在时抛出 ValueError。"""
+        svc = _make_service()
+        svc._clip_repo.session = MagicMock()
+        with pytest.raises(ValueError, match="不存在"):
+            svc.clone_plan_for_variant("no-such-plan", name_suffix="变体")

tests/unit/test_editor_clips_random_start.py

@@ -5,7 +5,7 @@
 - 素材不足时同一素材轮询切多个片段
 - 随机 start_time + used_segments 去重
 - 素材时长不足时 clip duration 缩短
-- 素材时长为 0 时抛 400
+- 素材时长全部为 0/缺失时抛 400「素材可切区间不足」；混合池中零时长素材被跳过
 - 使用 replace_all_clips_transactional 原子性替换
 - order 从 0 开始
 - start_time=None 时抛出 400
@@ -98,8 +98,8 @@ def _mock_segment_tracker():
             return_value=None,
         ),
         patch(
-            "app.api.routes.templates_editor.clips.make_reset_callback",
-            return_value=lambda asset_id: None,
+            "app.api.routes.templates_editor.clips.make_reuse_callback",
+            return_value=lambda asset_id, clip_duration: None,
         ),
         patch(
             "app.api.routes.templates_editor.clips.remove_used_segment",
@@ -276,7 +276,11 @@ class TestEditorClipsDurationAndStartTime:
 
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
     def test_zero_duration_asset_raises_400(self, mock_storage):
-        """素材时长为 0 时应抛出 400，而不是创建无效片段。"""
+        """所有素材时长均为 0 时轮询无可用素材，抛出 400「素材可切区间不足」。
+
+        新轮询逻辑下零时长素材被跳过（而非立即报错）；全部素材都被跳过时
+        返回 400，不创建无效片段。
+        """
         from app.api.routes.templates_editor.clips import (
             create_clips_from_assets_editor,
         )
@@ -302,7 +306,49 @@ class TestEditorClipsDurationAndStartTime:
                 )
 
         assert exc_info.value.status_code == 400
-        assert "时长" in exc_info.value.detail
+        assert "素材可切区间不足" in exc_info.value.detail
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_zero_duration_asset_skipped_in_mixed_pool(self, mock_storage):
+        """素材池混合零时长与正常素材时，零时长素材被跳过、正常素材承担片段。"""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc(replace_return_count=2)
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(
+            side_effect=lambda aid: {
+                "zero": _make_mock_asset("zero", 0.0),
+                "good": _make_mock_asset("good", 30.0),
+            }[aid]
+        )
+
+        body = ClipsFromAssetsRequest(asset_ids=["zero", "good"], required_clips_count=2)
+
+        with (
+            _patch_segments(_segments(2)),
+            patch(
+                "app.api.routes.templates_editor.clips._calc_random_start_time",
+                side_effect=[5.0, 12.0],
+            ),
+        ):
+            create_clips_from_assets_editor(
+                template_id="tpl-001",
+                body=body,
+                background_tasks=MagicMock(),
+                plan_id=TEST_PLAN_ID,
+                services=(MagicMock(), mock_plan_svc),
+                asset_repo=mock_asset_repo,
+                db=MagicMock(),
+                current_user=_make_auth_user(),
+            )
+
+        clips_data = _get_clips_data_from_call(mock_plan_svc)
+        assert len(clips_data) == 2
+        # 所有片段都分配给正常素材，零时长素材被跳过
+        assert all(c["asset_id"] == "good" for c in clips_data)
 
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
     def test_missing_duration_asset_raises_400(self, mock_storage):
@@ -450,7 +496,7 @@ class TestEditorClipsErrorHandling:
 
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
     def test_none_start_time_raises_400(self, mock_storage):
-        """_calc_random_start_time 返回 None 时应抛出 HTTPException 400。"""
+        """所有素材 calc 均返回 None（区间耗尽且复用被拒）→ 轮询失败抛 400。"""
         from app.api.routes.templates_editor.clips import (
             create_clips_from_assets_editor,
         )
@@ -458,7 +504,7 @@ class TestEditorClipsErrorHandling:
 
         mock_plan_svc = _make_plan_svc()
         mock_asset_repo = MagicMock()
-        # 素材有 duration 但 random 返回 None
+        # 素材有 duration 但 calc 返回 None（模拟可用区间耗尽、复用被闸门拒绝）
         mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
 
         body = ClipsFromAssetsRequest(asset_ids=["a1"])
@@ -483,7 +529,9 @@ class TestEditorClipsErrorHandling:
                 )
 
         assert exc_info.value.status_code == 400
-        assert "时长" in exc_info.value.detail
+        assert "素材可切区间不足" in exc_info.value.detail
+        # 复用被拒导致无起点时，不应创建任何片段
+        assert not mock_plan_svc.replace_all_clips_transactional.called
 
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
     def test_transactional_replace_exception_propagates(self, mock_storage):
@@ -512,3 +560,190 @@ class TestEditorClipsErrorHandling:
                     db=MagicMock(),
                     current_user=_make_auth_user(),
                 )
+
+
+class TestReuseRatioGate:
+    """素材区间耗尽后的受控复用与 15% 占比闸门（路由级）。"""
+
+    @staticmethod
+    def _make_calc_with_reuse(normal_starts, reused_durations):
+        """构造模拟「区间耗尽后受控复用」的 _calc_random_start_time。
+
+        normal_starts: list[float | None]，前 N 次调用返回的空闲起点；
+            返回 None 表示随机找不到空闲 → 触发 on_exhausted 复用回调。
+        回调被调用时返回复用区间（固定 0.0 起点），复用片段时长由路由累加到
+        reused_durations；回调内部占比预判超 15% 时返回 None（calc 随之 None）。
+        """
+        calls = {"i": 0}
+
+        def fake_calc(asset_id, clip_duration, durations, used_segments, on_exhausted=None):
+            i = calls["i"]
+            calls["i"] += 1
+            if i < len(normal_starts) and normal_starts[i] is not None:
+                return normal_starts[i]
+            # 空闲耗尽 → 走受控复用回调（回调返回 (start, end) 元组，calc 取起点）
+            if on_exhausted is not None:
+                result = on_exhausted(asset_id, clip_duration)
+                return result[0] if result else None
+            return None
+
+        return fake_calc, calls
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_reused_clip_ratio_within_threshold(self, mock_storage):
+        """素材 60s、片段 5s：前 12 个用空闲区间，第 13 个复用，
+        复用占比 5/(12*5+5)=7.7% ≤ 15%，正常创建 13 个片段。"""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc(replace_return_count=13)
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 60.0))
+        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=13)
+
+        reused: dict = {}
+
+        def reuse_cb(aid, dur):
+            # 模拟真实回调：返回复用区间前记录复用时长
+            reused[aid] = reused.get(aid, 0.0) + dur
+            return (0.0, dur)
+
+        # 前 12 次分配空闲起点；第 13 次 calc 直接走回调（normal_starts 越界 → None → 回调）
+        normal_starts = [float(i * 5) for i in range(12)]
+        fake_calc, _ = self._make_calc_with_reuse(normal_starts, reused)
+        with (
+            _patch_segments(_segments(13, dur_min=5.0, dur_max=5.0)),
+            patch("app.api.routes.templates_editor.clips._calc_random_start_time", side_effect=fake_calc),
+            patch("app.api.routes.templates_editor.clips.make_reuse_callback", return_value=reuse_cb),
+        ):
+            create_clips_from_assets_editor(
+                template_id="tpl-001",
+                body=body,
+                background_tasks=MagicMock(),
+                plan_id=TEST_PLAN_ID,
+                services=(MagicMock(), mock_plan_svc),
+                asset_repo=mock_asset_repo,
+                db=MagicMock(),
+                current_user=_make_auth_user(),
+            )
+        clips_data = _get_clips_data_from_call(mock_plan_svc)
+        assert len(clips_data) == 13
+        # 1 个复用片段，占比 1/13 ≈ 7.7% ≤ 15%
+        assert reused.get("a1", 0.0) == 5.0
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_reuse_ratio_exceeded_returns_400(self, mock_storage):
+        """复用占比将超 15% 时回调拒绝复用 → 无可用素材 → 400「素材可切区间不足」。
+
+        60s 素材、5s 片段：前 12 个空闲、随后复用占比累计；当 (reused+d)/(assigned+d)
+        超过 15% 时回调返回 None，calc 返回 None，轮询无素材 → 400。
+        """
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc(replace_return_count=0)
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 60.0))
+        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=20)
+
+        # 模拟真实回调：累计复用时长，预判超 15% 拒绝
+        reused: dict = {}
+        assigned: dict = {}
+
+        def fake_reuse_cb(aid, clip_duration):
+            a = assigned.get(aid, 0.0)
+            r = reused.get(aid, 0.0)
+            if a > 0 and (r + clip_duration) / (a + clip_duration) > 0.15:
+                return None  # 占比闸门拒绝
+            reused[aid] = r + clip_duration
+            return (0.0, clip_duration)
+
+        def fake_calc(asset_id, clip_duration, durations, used_segments, on_exhausted=None):
+            a = assigned.get(asset_id, 0.0)
+            # 前 12 个片段（60s/5s）有空闲区间
+            if a < 60.0:
+                start = a
+                assigned[asset_id] = a + clip_duration
+                return start
+            # 之后空闲耗尽 → 复用
+            if on_exhausted is not None:
+                result = on_exhausted(asset_id, clip_duration)
+                if result is not None:
+                    assigned[asset_id] = assigned.get(asset_id, 0.0) + clip_duration
+                return result[0] if result else None
+            return None
+
+        with (
+            _patch_segments(_segments(20, dur_min=5.0, dur_max=5.0)),
+            patch("app.api.routes.templates_editor.clips._calc_random_start_time", side_effect=fake_calc),
+            patch("app.api.routes.templates_editor.clips.make_reuse_callback", return_value=fake_reuse_cb),
+        ):
+            with pytest.raises(HTTPException) as exc_info:
+                create_clips_from_assets_editor(
+                    template_id="tpl-001",
+                    body=body,
+                    background_tasks=MagicMock(),
+                    plan_id=TEST_PLAN_ID,
+                    services=(MagicMock(), mock_plan_svc),
+                    asset_repo=mock_asset_repo,
+                    db=MagicMock(),
+                    current_user=_make_auth_user(),
+                )
+
+        assert exc_info.value.status_code == 400
+        assert "素材可切区间不足" in exc_info.value.detail
+        # 闸门在复用占比达上限时拒绝：60s 空闲 + 至多 ~15% 复用
+        assert reused.get("a1", 0.0) <= 12.0  # 10.0 或 15.0 以内，不会无限复用
+        # 未创建任何片段（整批失败）
+        assert not mock_plan_svc.replace_all_clips_transactional.called
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_calc_none_falls_through_to_next_asset(self, mock_storage):
+        """一个素材区间耗尽且复用被拒（calc 返回 None）时，轮询到下一个可用素材。"""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc(replace_return_count=2)
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(
+            side_effect=lambda aid: {
+                "exhausted": _make_mock_asset("exhausted", 60.0),
+                "fresh": _make_mock_asset("fresh", 60.0),
+            }[aid]
+        )
+        body = ClipsFromAssetsRequest(asset_ids=["exhausted", "fresh"], required_clips_count=2)
+
+        def fake_calc(asset_id, clip_duration, durations, used_segments, on_exhausted=None):
+            if asset_id == "exhausted":
+                # 空闲耗尽 + 回调拒绝 → None
+                return on_exhausted(asset_id, clip_duration) if on_exhausted else None
+            return 8.0  # 新鲜素材正常返回
+
+        with (
+            _patch_segments(_segments(2, dur_min=5.0, dur_max=5.0)),
+            patch("app.api.routes.templates_editor.clips._calc_random_start_time", side_effect=fake_calc),
+            patch(
+                "app.api.routes.templates_editor.clips.make_reuse_callback",
+                return_value=lambda aid, d: None,  # 复用始终被拒
+            ),
+        ):
+            create_clips_from_assets_editor(
+                template_id="tpl-001",
+                body=body,
+                background_tasks=MagicMock(),
+                plan_id=TEST_PLAN_ID,
+                services=(MagicMock(), mock_plan_svc),
+                asset_repo=mock_asset_repo,
+                db=MagicMock(),
+                current_user=_make_auth_user(),
+            )
+        clips_data = _get_clips_data_from_call(mock_plan_svc)
+        assert len(clips_data) == 2
+        # 耗尽素材被跳过，两个片段都分配给新鲜素材
+        assert all(c["asset_id"] == "fresh" for c in clips_data)

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_mediakit_conflicts.py

@@ -0,0 +1,68 @@
+"""MediaKit 智能选片挪点冲突检测测试（Task G 验收项 1）。
+
+覆盖 _recommended_time_conflicts：
+- 区间重叠判定（含 0.3s 边缘间隙扩边）
+- 不冲突场景（间隔大于边缘间隙）
+- 边缘间隙可配置
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+from app.api.routes.templates_editor.clips import (
+    SEGMENT_EDGE_GAP,
+    _recommended_time_conflicts,
+)
+
+
+class TestRecommendedTimeConflicts:
+    def test_overlapping_range_conflicts(self):
+        """推荐区间与已用区间直接重叠 → 冲突。"""
+        assert _recommended_time_conflicts(10.0, 5.0, [(12.0, 17.0)]) is True
+
+    def test_identical_range_conflicts(self):
+        assert _recommended_time_conflicts(10.0, 5.0, [(10.0, 15.0)]) is True
+
+    def test_touching_endpoint_conflicts_due_to_edge_gap(self):
+        """首尾紧贴（推荐 15 开始，已用 [10,15]）：0.3s 扩边内 → 冲突。"""
+        assert _recommended_time_conflicts(15.0, 5.0, [(10.0, 15.0)]) is True
+
+    def test_gap_within_edge_gap_conflicts(self):
+        """间隔 0.2s（< 0.3s 边缘间隙）→ 冲突。"""
+        assert _recommended_time_conflicts(15.2, 5.0, [(10.0, 15.0)]) is True
+
+    def test_gap_beyond_edge_gap_no_conflict(self):
+        """间隔 0.5s（> 0.3s 边缘间隙）→ 不冲突。"""
+        assert _recommended_time_conflicts(15.5, 5.0, [(10.0, 15.0)]) is False
+
+    def test_far_apart_no_conflict(self):
+        """相隔很远 → 不冲突。"""
+        assert _recommended_time_conflicts(20.0, 5.0, [(0.0, 5.0)]) is False
+
+    def test_empty_used_no_conflict(self):
+        assert _recommended_time_conflicts(10.0, 5.0, []) is False
+
+    def test_any_one_range_conflicts(self):
+        """多个已用区间，任一冲突即返回 True。"""
+        used = [(0.0, 5.0), (10.0, 15.0), (20.0, 25.0)]
+        assert _recommended_time_conflicts(12.0, 2.0, used) is True
+        assert _recommended_time_conflicts(6.0, 2.0, used) is False
+
+    def test_custom_edge_gap(self):
+        """edge_gap 可配置：gap=0 时紧贴不冲突（端点相接不算重叠）。"""
+        # edge_gap=0：15.0 开始与已用 [10,15] 端点相接，区间判定 start<end_gap(15) → False
+        assert _recommended_time_conflicts(15.0, 5.0, [(10.0, 15.0)], edge_gap=0.0) is False
+        # edge_gap=1.0：0.5 间隔也算冲突
+        assert _recommended_time_conflicts(15.5, 5.0, [(10.0, 15.0)], edge_gap=1.0) is True
+
+    def test_default_edge_gap_constant(self):
+        """默认边缘间隙常量为 0.3s（配置常量）。"""
+        assert SEGMENT_EDGE_GAP == 0.3

tests/unit/test_mediakit_smart_clips.py

@@ -69,8 +69,13 @@ class TestRecommendedTimeConflicts:
     def test_conflict_exact_boundary_no_overlap(self):
         from app.api.routes.templates_editor.clips import _recommended_time_conflicts
 
-        # 推荐 [10, 15]，已用 [0, 10] — 边界相接不算冲突
-        assert _recommended_time_conflicts(10.0, 5.0, [(0.0, 10.0)]) is False
+        # 新语义：默认 0.3s 边缘间隙扩边，推荐 [10, 15] 与已用 [0, 10] 首尾相接
+        # 落在扩边范围内 → 判为冲突（避免观感重复）
+        assert _recommended_time_conflicts(10.0, 5.0, [(0.0, 10.0)]) is True
+        # 显式 edge_gap=0 时退回纯区间重叠判定：相接不算重叠
+        assert _recommended_time_conflicts(10.0, 5.0, [(0.0, 10.0)], edge_gap=0.0) is False
+        # 间隙大于边缘间隙（0.5 > 0.3）→ 不冲突
+        assert _recommended_time_conflicts(10.5, 5.0, [(0.0, 10.0)]) is False
 
     def test_conflict_multiple_used(self):
         from app.api.routes.templates_editor.clips import _recommended_time_conflicts
@@ -78,8 +83,10 @@ class TestRecommendedTimeConflicts:
         used = [(0.0, 5.0), (10.0, 15.0), (20.0, 25.0)]
         # 推荐 [6, 11] 与 [10, 15] 冲突
         assert _recommended_time_conflicts(6.0, 5.0, used) is True
-        # 推荐 [15, 20] 不冲突
-        assert _recommended_time_conflicts(15.0, 5.0, used) is False
+        # 推荐 [15, 20] 与 [10, 15] 首尾相接：0.3s 扩边内 → 冲突
+        assert _recommended_time_conflicts(15.0, 5.0, used) is True
+        # 空闲段 [5.3, 9.7] 长 4.4s：推荐 [5.5, 9.5]（dur=4）与三区间扩边均不接触
+        assert _recommended_time_conflicts(5.5, 4.0, used) is False
 
 
 # ── _get_mediakit_recommendations 单元测试 ──────────────────────────────────

tests/unit/test_preview_edit_plan_association.py

@@ -13,6 +13,7 @@ from __future__ import annotations
 import os
 import sys
 from dataclasses import dataclass, field
+from types import SimpleNamespace
 from typing import Any, Optional
 from unittest.mock import MagicMock, patch
 
@@ -180,17 +181,19 @@ 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 +212,18 @@ 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",