feat(assets): 素材余量角标 + Step2 过滤已用尽素材 (#1548)

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>

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

@@ -26,7 +26,6 @@ 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
@@ -36,23 +35,6 @@ 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
@@ -97,7 +79,6 @@ 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),
     )
 
 
@@ -588,35 +569,13 @@ 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 filtered_results
+        for r in results
     ]
 
     return SmartMatchResponse(items=items, total_candidates=total_candidates)

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

@@ -358,41 +358,6 @@ 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,54 +418,8 @@ 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 task_index in range(count):
-            # 第 1 条复用源 plan（保留用户编辑结果）；其余使用预克隆的独立变体 plan
-            effective_plan_id = (
-                variant_plan_ids[task_index - 1] if task_index > 0 else request.source_edit_plan_id
-            )
-
+        for _ in range(count):
             task = use_case.execute(
                 CreateGenerationTaskCommand(
                     project_id=project_id,
@@ -477,7 +431,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=effective_plan_id,
+                    source_edit_plan_id=request.source_edit_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,10 +24,8 @@ 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_reuse_callback,
+    make_reset_callback,
     record_used_segments,
     remove_used_segment,
 )
@@ -438,16 +436,11 @@ 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 + edge_gap and end > used_start - edge_gap:
+        if start < used_end and end > used_start:
             return True
     return False
 
@@ -618,98 +611,55 @@ def create_clips_from_assets_editor(
     used_segments: dict[str, list[tuple[float, float]]] = get_used_segments(
         db, unique_asset_ids
     )
-    # 受控复用回调：可用区间耗尽时复用最久未用且未达复用上限（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,
-    )
+    reset_cb = make_reset_callback(db, used_segments)
     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):
-        # 在 segment 的 duration_min ~ duration_max 之间随机取值（保留一位小数）
-        raw_duration = random.uniform(dur_min, dur_max)
+        # 轮询分配素材
+        asset_id = body.asset_ids[i % len(body.asset_ids)]
+        asset_total = asset_durations.get(asset_id, 0.0)
 
-        # 轮询分配素材：跳过时长缺失、复用占比已超 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
-
-        if not asset_id or start_time is None:
-            # 所有素材时长缺失、复用占比超阈值，或区间耗尽且复用被拒 → 素材可切区间不足
+        # 素材时长为 0 或缺失时无法创建有效片段
+        if asset_total <= 0:
             raise HTTPException(
                 status_code=status.HTTP_400_BAD_REQUEST,
-                detail="素材可切区间不足，请补充新素材",
+                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} 时长不足，无法创建有效片段",
+            )
+
+        # 使用随机起始时间（不调用MediaKit，保证接口快速返回）
+        start_time = _calc_random_start_time(
+            asset_id,
+            clip_duration,
+            asset_durations,
+            used_segments,
+            on_exhausted=reset_cb,
+        )
+
+        if start_time is None:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"素材 {asset_id} 时长信息缺失，无法计算起始时间",
             )
 
         # 记录已使用时间段（内存，供本次后续片段避开）
         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
         )
@@ -816,10 +766,6 @@ 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()
         # 已更新的时间段
@@ -861,17 +807,11 @@ 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 片段创建时已写入历史记录
-            other_segments = list(
-                set(other_segments) | set(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,14 +53,6 @@ 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,27 +1,17 @@
-"""素材片段级使用记录追踪与受控复用.
+"""素材片段级使用记录追踪.
 
 在素材 metadata（assets.classification_result JSON）中持久化已使用的片段时间区间，
-供 from-assets 创建片段时避开历史区间，实现跨任务/跨调用的片段去重；
-素材可用区间耗尽后进入受控复用：允许有限次数（MAX_RANGE_USE_COUNT）复用最久未用
-的历史区间，配合调用方的成片复用占比控制（MAX_REUSE_RATIO = 15%），把任意两条
-成片的画面重复率控制在阈值内。
+供 from-assets 创建片段时避开历史区间，实现跨任务/跨调用的片段去重。
 
-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",
-            "use_count": 1,              # 该区间累计被使用次数（复用一次 +1）
-            "last_used_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"},
         ...
     ]
 
 注意：本模块所有函数都不自行 commit，由调用方控制事务边界
 （from-assets 与 replace_all_clips_transactional 同事务；异步任务各自 commit）。
-历史记录永不自动清空（自动轮回重置已下线，reset_used_segments 仅保留给运维/测试）。
 """
 
 from __future__ import annotations
@@ -39,51 +29,24 @@ logger = logging.getLogger(__name__)
 
 USED_RANGES_KEY = "used_time_ranges"
 
-# ── 受控复用配置常量 ─────────────────────────────────────────────────────────
-MAX_RANGE_USE_COUNT = 3
-"""单条历史区间最多被使用次数（含首次），达到后不再参与复用。"""
 
-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:
+def _read_ranges(model: AssetModel) -> list[dict]:
     """从 AssetModel 读取 metadata dict（classification_result 列承载的 JSON）."""
     if not model.classification_result:
         return {}
     try:
-        data = json.loads(model.classification_result)
-        return data if isinstance(data, dict) else {}
+        return json.loads(model.classification_result)
     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`` 参数格式一致，可直接传入。
@@ -93,7 +56,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_meta(model)
+        meta = _read_ranges(model)
         ranges = meta.get(USED_RANGES_KEY) or []
         segments: list[tuple[float, float]] = []
         for r in ranges:
@@ -113,49 +76,19 @@ def record_used_segments(
     end: float,
     plan_id: str,
 ) -> None:
-    """记录一次片段使用（不 commit）.
-
-    若新区间与某条历史区间高度重叠（复用场景，如受控复用回调返回的区间、
-    MediaKit 挪到历史区间），则累加该记录的 ``use_count`` 并刷新 ``last_used_at``，
-    不新增记录；否则追加一条新记录（use_count=1）。
-    """
-    # 行级锁读取：与并发生成任务互斥，保证区间记录读-改-写一致
-    model = _get_model(db, asset_id, for_update=True)
+    """向素材 metadata 追加一条片段使用记录（不 commit）."""
+    model = db.query(AssetModel).filter(AssetModel.id == asset_id).first()
     if model is None:
         logger.warning("[片段追踪] 素材不存在，跳过记录: asset_id=%s", asset_id)
         return
-    meta = _read_meta(model)
+    meta = _read_ranges(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": new_start,
-            "end": new_end,
+            "start": round(float(start), 3),
+            "end": round(float(end), 3),
             "plan_id": plan_id,
-            "created_at": now,
-            "use_count": 1,
-            "last_used_at": now,
+            "created_at": datetime.now(timezone.utc).isoformat(),
         }
     )
     meta[USED_RANGES_KEY] = ranges
@@ -178,10 +111,10 @@ def remove_used_segment(
     Returns:
         是否找到并删除了记录。
     """
-    model = _get_model(db, asset_id)
+    model = db.query(AssetModel).filter(AssetModel.id == asset_id).first()
     if model is None:
         return False
-    meta = _read_meta(model)
+    meta = _read_ranges(model)
     ranges = list(meta.get(USED_RANGES_KEY) or [])
     remaining: list[dict] = []
     removed = False
@@ -194,7 +127,8 @@ 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:
@@ -211,219 +145,30 @@ def remove_used_segment(
 def reset_used_segments(db: Session, asset_id: str) -> None:
     """清空单个素材的历史片段使用记录（不 commit）.
 
-    仅供运维/测试使用；正常生成流程中历史记录永不自动清空（受控复用取代自动轮回）。
+    单个素材的可用区间被全部占用（轮回一圈）后调用，使后续片段可重新使用整段素材。
     """
-    model = _get_model(db, asset_id)
+    model = db.query(AssetModel).filter(AssetModel.id == asset_id).first()
     if model is None:
         return
-    meta = _read_meta(model)
+    meta = _read_ranges(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)
 
 
-# ── 素材余量/可用性计算（Task H：素材库角标 + smart-match 过滤）──────────────
+def make_reset_callback(db: Session, used_segments: dict) -> Callable[[str], None]:
+    """构造给 _calc_random_start_time 用的 reset 回调.
 
-# 判定「是否还有空闲可切区间」时使用的最小片段时长（秒）：空闲段长于此值才视为可切
-_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，不影响主生成流程。
+    回调同时清空持久化 metadata 和内存中的 used_segments，使重试随机能覆盖全素材。
     """
 
-    def _reuse(asset_id: str, clip_duration: float) -> tuple[float, float] | None:
+    def _reset(asset_id: str) -> None:
         try:
-            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)
+            reset_used_segments(db, asset_id)
         except Exception:
-            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
+            logger.warning("[片段追踪] reset 持久化记录失败: asset_id=%s", asset_id, exc_info=True)
+        used_segments.pop(asset_id, None)
 
-    return _reuse
+    return _reset

apps/api/app/services/edit_plan_service.py

@@ -9,12 +9,6 @@ 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 (
@@ -459,116 +453,6 @@ 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]:

apps/web/src/api/assets/index.ts

@@ -20,6 +20,10 @@ export type {
 // 素材诊断
 export { getAssetDiagnosis } from "./diagnosis"
 
+// 素材余量/可用性判断
+export { isAssetUsable } from "./usage"
+export type { AssetUsageLike } from "./usage"
+
 // 素材库
 export {
   getAssetLibraries,

apps/web/src/api/assets/types.ts

@@ -40,6 +40,10 @@ export interface AssetItem {
   thumbnail_url?: string
   /** 时长（秒），视频/音频素材由后端从 metadata 提取到顶层 */
   duration?: number
+  /** 已切片段占用时长占比（0~1，后端片段重复率控制机制返回；字段缺失视为未统计） */
+  used_ratio?: number | null
+  /** 是否已彻底用尽（无新区间且历史区间复用次数均达上限）；false 的素材不参与生成选片 */
+  usable?: boolean | null
   status?: string
   classification_status?: AssetClassificationStatus | null
   quality_score?: number | null

apps/web/src/api/assets/usage.ts

@@ -0,0 +1,18 @@
+/**
+ * 素材余量/可用性判断
+ * 后端片段重复率控制机制（任意两条成片画面重复率 ≤15%）上线后，
+ * 素材列表会附加 usable / used_ratio 字段。字段未上线前一律按可用处理。
+ */
+
+/** 仅依赖素材余量相关字段的最小结构，api 层与 pages 层 AssetItem 均可传入 */
+export interface AssetUsageLike {
+  usable?: boolean | null
+  used_ratio?: number | null
+}
+
+/**
+ * 素材是否仍可参与生成选片。
+ * usable === false 表示已彻底用尽（无新区间且复用次数全部达上限）；
+ * 字段缺失（undefined/null）时降级为可用，保证后端字段上线前零影响。
+ */
+export const isAssetUsable = (asset: AssetUsageLike): boolean => asset.usable !== false

apps/web/src/pages/assets/assets.css

@@ -402,6 +402,44 @@
   margin-bottom: var(--space-sm);
 }
 
+/* 状态标签 + 余量角标行 */
+.xx-asset-meta-left {
+  display: inline-flex;
+  align-items: center;
+  gap: var(--space-xs);
+  min-width: 0;
+}
+
+/* 视频素材余量角标（仅状态展示，不影响卡片操作） */
+.xx-asset-usage-badge {
+  display: inline-flex;
+  align-items: center;
+  padding: var(--space-xxs) var(--space-sm);
+  border-radius: var(--radius-full);
+  font-size: var(--font-size-xs);
+  font-weight: var(--font-weight-medium);
+  line-height: 1.5;
+  white-space: nowrap;
+}
+
+/* 已用尽：红色实心 */
+.xx-asset-usage-badge-exhausted {
+  background: var(--error-color);
+  color: var(--text-inverse);
+}
+
+/* 即将用尽：红色软底 */
+.xx-asset-usage-badge-warning {
+  background: var(--error-soft);
+  color: var(--error-color);
+}
+
+/* 已用 xx%：橙色软底 */
+.xx-asset-usage-badge-ratio {
+  background: var(--warning-soft);
+  color: var(--warning-color);
+}
+
 /* 诊断按钮 */
 .xx-asset-diagnose-btn {
   width: 100%;

apps/web/src/pages/assets/components/AssetCard.tsx

@@ -8,7 +8,7 @@ import {
   CloseCircleOutlined,
 } from "@ant-design/icons"
 import { Popconfirm } from "antd"
-import type { AssetItem } from "@/pages/assets/types"
+import { getUsageBadge, type AssetItem } from "@/pages/assets/types"
 import { thumbGradient } from "@/pages/assets/utils/asset"
 import { kindIcon } from "@/pages/assets/utils/kindIcon"
 import { StatusPill } from "./AssetSkeleton"
@@ -34,95 +34,106 @@ const AssetCard: React.FC<AssetCardProps> = ({
   onDiagnose,
   onPlay,
   onDelete,
-}) => (
-  <div className={`xx-asset-card${selected ? " xx-asset-card-selected" : ""}`} onClick={onToggle}>
-    {/* 缩略图区 */}
-    <div className="xx-asset-thumb" style={{ background: thumbGradient(asset.kind) }}>
-      {asset.thumbUrl ? (
-        <img src={asset.thumbUrl} alt={asset.name} />
-      ) : (
-        <span className="xx-asset-thumb-placeholder">
-          {asset.loading ? <LoadingOutlined /> : kindIcon(asset.kind)}
-        </span>
-      )}
+}) => {
+  // 视频素材余量角标（已用尽/即将用尽/已用 xx%）；非视频或字段缺失返回 null
+  const usageBadge = getUsageBadge(asset)
+  return (
+    <div className={`xx-asset-card${selected ? " xx-asset-card-selected" : ""}`} onClick={onToggle}>
+      {/* 缩略图区 */}
+      <div className="xx-asset-thumb" style={{ background: thumbGradient(asset.kind) }}>
+        {asset.thumbUrl ? (
+          <img src={asset.thumbUrl} alt={asset.name} />
+        ) : (
+          <span className="xx-asset-thumb-placeholder">
+            {asset.loading ? <LoadingOutlined /> : kindIcon(asset.kind)}
+          </span>
+        )}
 
-      {/* 处理中遮罩 */}
-      {asset.loading && (
-        <div className="xx-asset-thumb-overlay xx-asset-thumb-processing">
-          <LoadingOutlined />
-          <span>处理中</span>
+        {/* 处理中遮罩 */}
+        {asset.loading && (
+          <div className="xx-asset-thumb-overlay xx-asset-thumb-processing">
+            <LoadingOutlined />
+            <span>处理中</span>
+          </div>
+        )}
+
+        {/* 失败状态标识 */}
+        {asset.status === "bad" && asset.statusLabel === "处理失败" && (
+          <div className="xx-asset-thumb-overlay xx-asset-thumb-failed">
+            <CloseCircleOutlined />
+            <span>处理失败</span>
+          </div>
+        )}
+
+        {/* 视频/配音类显示播放按钮（处理中/失败不显示） */}
+        {asset.kind === "video" && !asset.loading && asset.status !== "bad" && (
+          <span
+            className="xx-asset-play"
+            onClick={(e) => {
+              e.stopPropagation()
+              onPlay()
+            }}
+          >
+            <PlayCircleOutlined />
+          </span>
+        )}
+
+        {/* 删除按钮 */}
+        <Popconfirm
+          title="确认删除"
+          description="删除后不可恢复，确定要删除这个素材吗？"
+          onConfirm={(e) => {
+            e?.stopPropagation()
+            onDelete()
+          }}
+          onCancel={(e) => e?.stopPropagation()}
+          okText="删除"
+          cancelText="取消"
+          okButtonProps={{ danger: true }}
+        >
+          <span className="xx-asset-delete" onClick={(e) => e.stopPropagation()}>
+            <DeleteOutlined />
+          </span>
+        </Popconfirm>
+
+        {/* 选中态勾选 */}
+        {selected && (
+          <span className="xx-asset-check">
+            <CheckOutlined />
+          </span>
+        )}
+      </div>
+
+      {/* 信息区 */}
+      <div className="xx-asset-info">
+        <p className="xx-asset-name" title={asset.name}>
+          {asset.name}
+        </p>
+        <div className="xx-asset-meta">
+          <span className="xx-asset-meta-left">
+            <StatusPill status={asset.status} label={asset.statusLabel} />
+            {usageBadge && (
+              <span className={`xx-asset-usage-badge xx-asset-usage-badge-${usageBadge.variant}`}>
+                {usageBadge.label}
+              </span>
+            )}
+          </span>
+          {asset.duration && <span>{asset.duration}</span>}
         </div>
-      )}
-
-      {/* 失败状态标识 */}
-      {asset.status === "bad" && asset.statusLabel === "处理失败" && (
-        <div className="xx-asset-thumb-overlay xx-asset-thumb-failed">
-          <CloseCircleOutlined />
-          <span>处理失败</span>
-        </div>
-      )}
-
-      {/* 视频/配音类显示播放按钮（处理中/失败不显示） */}
-      {asset.kind === "video" && !asset.loading && asset.status !== "bad" && (
-        <span
-          className="xx-asset-play"
+        <button
+          className={`xx-asset-diagnose-btn${diagnosing ? " xx-asset-diagnose-btn-loading" : ""}`}
+          disabled={diagnosing || asset.loading || asset.status === "bad"}
           onClick={(e) => {
             e.stopPropagation()
-            onPlay()
+            onDiagnose()
           }}
         >
-          <PlayCircleOutlined />
-        </span>
-      )}
-
-      {/* 删除按钮 */}
-      <Popconfirm
-        title="确认删除"
-        description="删除后不可恢复，确定要删除这个素材吗？"
-        onConfirm={(e) => {
-          e?.stopPropagation()
-          onDelete()
-        }}
-        onCancel={(e) => e?.stopPropagation()}
-        okText="删除"
-        cancelText="取消"
-        okButtonProps={{ danger: true }}
-      >
-        <span className="xx-asset-delete" onClick={(e) => e.stopPropagation()}>
-          <DeleteOutlined />
-        </span>
-      </Popconfirm>
-
-      {/* 选中态勾选 */}
-      {selected && (
-        <span className="xx-asset-check">
-          <CheckOutlined />
-        </span>
-      )}
-    </div>
-
-    {/* 信息区 */}
-    <div className="xx-asset-info">
-      <p className="xx-asset-name" title={asset.name}>
-        {asset.name}
-      </p>
-      <div className="xx-asset-meta">
-        <StatusPill status={asset.status} label={asset.statusLabel} />
-        {asset.duration && <span>{asset.duration}</span>}
+          {diagnosing ? <LoadingOutlined /> : <ExperimentOutlined />}
+          {diagnosing ? "诊断中..." : "诊断"}
+        </button>
       </div>
-      <button
-        className={`xx-asset-diagnose-btn${diagnosing ? " xx-asset-diagnose-btn-loading" : ""}`}
-        disabled={diagnosing || asset.loading || asset.status === "bad"}
-        onClick={(e) => {
-          e.stopPropagation()
-          onDiagnose()
-        }}
-      >
-        {diagnosing ? <LoadingOutlined /> : <ExperimentOutlined />}
-        {diagnosing ? "诊断中..." : "诊断"}
-      </button>
     </div>
-  </div>
-)
+  )
+}
 
 export default AssetCard

apps/web/src/pages/assets/types.ts

@@ -27,6 +27,36 @@ export interface AssetItem {
   duration?: string
   size: number
   createdAt: string
+  /** 已切片段占用时长占比（0~1），后端字段缺失时为 undefined */
+  usedRatio?: number
+  /** 是否已彻底用尽（false 的素材不参与生成选片），字段缺失时视为可用 */
+  usable?: boolean
+}
+
+/** 素材余量角标状态（仅视频素材） */
+export interface UsageBadge {
+  /** 角标文案 */
+  label: string
+  /** 样式变体：exhausted=红色实心，warning=红色软底，ratio=橙色软底 */
+  variant: "exhausted" | "warning" | "ratio"
+}
+
+/**
+ * 根据后端余量字段计算视频素材的余量角标；
+ * 非视频、字段缺失或已用占比 <50% 时不显示（返回 null）。
+ */
+export const getUsageBadge = (asset: {
+  kind?: AssetKind
+  usable?: boolean
+  usedRatio?: number
+}): UsageBadge | null => {
+  if (asset.kind && asset.kind !== "video") return null
+  if (asset.usable === false) return { label: "已用尽", variant: "exhausted" }
+  const ratio = asset.usedRatio
+  if (ratio == null) return null
+  if (ratio >= 0.85) return { label: "即将用尽", variant: "warning" }
+  if (ratio >= 0.5) return { label: `已用 ${Math.round(ratio * 100)}%`, variant: "ratio" }
+  return null
 }
 
 /** 根据 mime_type 推断前端 AssetKind */
@@ -111,5 +141,7 @@ export const mapAsset = (item: ApiAssetItem): AssetItem => {
     duration: metadata.duration != null ? formatDuration(metadata.duration as number) : undefined,
     size: item.file_size ? +(item.file_size / (1024 * 1024)).toFixed(1) : 0,
     createdAt: item.created_at ? new Date(item.created_at).toISOString().slice(0, 10) : "—",
+    usedRatio: item.used_ratio ?? undefined,
+    usable: item.usable ?? undefined,
   }
 }

apps/web/src/pages/generate/components/Step2MaterialSelect.tsx

@@ -62,8 +62,9 @@ const Step2MaterialSelect: React.FC<Step2MaterialSelectProps> = (props) => {
           </div>
 
           <ManualMaterialList
-            materials={m.materials}
+            materials={m.selectableMaterials}
             materialsLoading={m.materialsLoading}
+            allExhausted={m.allMaterialsExhausted}
             selectedMaterials={m.selectedMaterials}
             onToggle={m.handleToggleMaterial}
           />
@@ -77,7 +78,7 @@ const Step2MaterialSelect: React.FC<Step2MaterialSelectProps> = (props) => {
             onMatch={m.handleSmartMatch}
             hasMatched={m.hasMatched}
             onRefresh={m.handleRefreshMatch}
-            materialsCount={m.materials.items.length}
+            materialsCount={m.selectableMaterials.items.length}
             loading={m.materialsLoading}
           />
 

apps/web/src/pages/generate/components/material/ManualMaterialList.tsx

@@ -11,6 +11,8 @@ const { Text } = Typography
 interface ManualMaterialListProps {
   materials: { items: AssetItem[]; total: number }
   materialsLoading: boolean
+  /** 库内有素材但全部已用尽（usable === false），用于区分空状态文案 */
+  allExhausted?: boolean
   selectedMaterials: string[]
   onToggle: (materialId: string) => void
 }
@@ -247,6 +249,7 @@ const MaterialCard: React.FC<{
 const ManualMaterialList: React.FC<ManualMaterialListProps> = ({
   materials,
   materialsLoading,
+  allExhausted,
   selectedMaterials,
   onToggle,
 }) => {
@@ -256,7 +259,9 @@ const ManualMaterialList: React.FC<ManualMaterialListProps> = ({
         <Text style={{ color: "var(--text-secondary)", padding: "16px 0" }}>加载素材中…</Text>
       ) : materials.items.length === 0 ? (
         <Text style={{ color: "var(--text-secondary)", padding: "16px 0" }}>
-          暂无素材，请先在视频库中上传
+          {allExhausted
+            ? "暂无可选素材（素材可能已用尽，请先上传新素材）"
+            : "暂无素材，请先在视频库中上传"}
         </Text>
       ) : (
         <div

apps/web/src/pages/generate/hooks/step2-materials/useMaterialLibrary.ts

@@ -1,6 +1,6 @@
-import { useState, useEffect } from "react"
+import { useState, useEffect, useMemo } from "react"
 import { useQuery } from "@tanstack/react-query"
-import { getAssets, getAssetLibraries } from "@/api/assets"
+import { getAssets, getAssetLibraries, isAssetUsable } from "@/api/assets"
 import type { AssetItem } from "@/api/assets"
 
 /**
@@ -31,11 +31,26 @@ export function useMaterialLibrary() {
     enabled: !!selectedLibraryId,
   })
 
+  // 生成选片只展示仍可切出不重复片段的素材（usable !== false）；
+  // 后端字段未上线时 isAssetUsable 恒为 true，过滤为 no-op
+  const selectableMaterials = useMemo(
+    () => ({
+      items: materials.items.filter(isAssetUsable),
+      total: materials.total,
+    }),
+    [materials],
+  )
+
+  // 库内有素材但全部已用尽（用于区分空状态文案）
+  const allMaterialsExhausted = materials.items.length > 0 && selectableMaterials.items.length === 0
+
   return {
     libraries,
     selectedLibraryId,
     setSelectedLibraryId,
     materials,
+    selectableMaterials,
+    allMaterialsExhausted,
     materialsLoading,
   }
 }

apps/web/src/pages/generate/hooks/step2-materials/useSmartMatch.ts

@@ -1,7 +1,7 @@
 import { useState, useCallback, useMemo } from "react"
 import { message } from "antd"
 import type { AssetItem } from "@/api/assets"
-import { smartMatchAssets } from "@/api/assets"
+import { smartMatchAssets, isAssetUsable } from "@/api/assets"
 
 interface UseSmartMatchOptions {
   libraryId: string
@@ -32,38 +32,41 @@ export function useSmartMatch({
       return
     }
 
-    if (materials.items.length === 0) {
-      message.warning("当前视频库暂无素材")
+    // 已用尽素材（usable === false）不参与智能匹配；
+    // 后端字段未上线时 isAssetUsable 恒为 true，过滤为 no-op
+    const usableItems = materials.items.filter(isAssetUsable)
+    if (usableItems.length === 0) {
+      message.warning(
+        materials.items.length === 0 ? "当前视频库暂无素材" : "素材可能已用尽，请先上传新素材",
+      )
       return
     }
 
     setSmartMatching(true)
 
     try {
-      // 调用后端智能匹配 API
+      // 调用后端智能匹配 API（后端也会排除已用尽素材，这里前端兜底过滤）
       const result = await smartMatchAssets(libraryId)
-      const matchedIds = result.items?.map((a: AssetItem) => a.id) ?? []
+      const matched = (result.items ?? []).filter(isAssetUsable)
+      const matchedIds = matched.map((a: AssetItem) => a.id)
 
       if (matchedIds.length > 0) {
         onSmartSelectedIdsChange(matchedIds)
         // 保存 API 返回的完整素材列表
-        const resolved = result.items?.length
-          ? result.items
-          : materials.items.filter((a) => matchedIds.includes(a.id))
-        setSmartMatchedResults(resolved)
+        setSmartMatchedResults(matched)
         setHasMatched(true)
         message.success(`AI 已为你选择 ${matchedIds.length} 个素材`)
       } else {
-        // 后端返回空结果，回退到全选
-        onSmartSelectedIdsChange(materials.items.map((a) => a.id))
-        setSmartMatchedResults(materials.items)
+        // 后端返回空结果，回退到全选可用素材
+        onSmartSelectedIdsChange(usableItems.map((a) => a.id))
+        setSmartMatchedResults(usableItems)
         setHasMatched(true)
         message.info("AI 暂未找到匹配素材，已全选当前库素材")
       }
     } catch {
-      // 后端 API 尚未就绪时，回退到全选当前库素材
-      onSmartSelectedIdsChange(materials.items.map((a) => a.id))
-      setSmartMatchedResults(materials.items)
+      // 后端 API 尚未就绪时，回退到全选当前库可用素材
+      onSmartSelectedIdsChange(usableItems.map((a) => a.id))
+      setSmartMatchedResults(usableItems)
       setHasMatched(true)
       message.info("已为你全选当前库素材（智能匹配功能即将上线）")
     } finally {
@@ -77,7 +80,7 @@ export function useSmartMatch({
   }, [handleSmartMatch])
 
   const handleSelectAllMatched = useCallback(() => {
-    onSmartSelectedIdsChange(materials.items.map((a) => a.id))
+    onSmartSelectedIdsChange(materials.items.filter(isAssetUsable).map((a) => a.id))
   }, [materials.items, onSmartSelectedIdsChange])
 
   const handleClearSmartSelect = useCallback(() => {
@@ -99,7 +102,7 @@ export function useSmartMatch({
   const smartSelectedTotalDuration = useMemo(
     () =>
       materials.items
-        .filter((a) => smartSelectedIds.includes(a.id))
+        .filter((a) => isAssetUsable(a) && smartSelectedIds.includes(a.id))
         .reduce((sum, a) => sum + (a.duration || 0), 0),
     [materials.items, smartSelectedIds],
   )

apps/web/src/pages/generate/hooks/useStep2Materials.ts

@@ -38,12 +38,19 @@ export function useStep2Materials({
   templateSegments,
   onServerClipsChange,
 }: UseStep2MaterialsProps) {
-  const { libraries, selectedLibraryId, setSelectedLibraryId, materials, materialsLoading } =
-    useMaterialLibrary()
+  const {
+    libraries,
+    selectedLibraryId,
+    setSelectedLibraryId,
+    materials,
+    selectableMaterials,
+    allMaterialsExhausted,
+    materialsLoading,
+  } = useMaterialLibrary()
 
   const smartMatch = useSmartMatch({
     libraryId: selectedLibraryId,
-    materials,
+    materials: selectableMaterials,
     smartSelectedIds,
     onSmartSelectedIdsChange,
   })
@@ -59,13 +66,19 @@ export function useStep2Materials({
     }
     if (!selectedLibraryId) return
     if (materialsLoading) return
-    if (materials.items.length === 0) return
+    if (selectableMaterials.items.length === 0) return
     // 防止同一视频库重复触发
     if (autoTriggeredRef.current === selectedLibraryId) return
 
     autoTriggeredRef.current = selectedLibraryId
     handleSmartMatch()
-  }, [selectedLibraryId, materialMode, materialsLoading, materials.items, handleSmartMatch])
+  }, [
+    selectedLibraryId,
+    materialMode,
+    materialsLoading,
+    selectableMaterials.items,
+    handleSmartMatch,
+  ])
 
   /* ── Step2 选择素材后自动保存草稿 asset_ids（防抖 500ms，失败静默） ── */
   const { scheduleSave } = useDraftAutoSave(selectedTemplate)
@@ -169,6 +182,8 @@ export function useStep2Materials({
     selectedLibraryId,
     setSelectedLibraryId,
     materials,
+    selectableMaterials,
+    allMaterialsExhausted,
     materialsLoading,
     // 模式
     materialMode,

apps/web/src/test/api/asset-usage.test.ts

@@ -0,0 +1,24 @@
+import { describe, expect, it } from "vitest"
+import { isAssetUsable } from "@/api/assets"
+
+describe("isAssetUsable", () => {
+  it("usable 字段缺失时降级为可用（后端字段未上线零影响）", () => {
+    expect(isAssetUsable({})).toBe(true)
+    expect(isAssetUsable({ usable: undefined })).toBe(true)
+    expect(isAssetUsable({ usable: null })).toBe(true)
+  })
+
+  it("usable === true 时可用", () => {
+    expect(isAssetUsable({ usable: true })).toBe(true)
+  })
+
+  it("usable === false 时不可用（已彻底用尽）", () => {
+    expect(isAssetUsable({ usable: false })).toBe(false)
+    expect(isAssetUsable({ usable: false, used_ratio: 1 })).toBe(false)
+  })
+
+  it("used_ratio 不影响可用性判断（只影响角标展示）", () => {
+    expect(isAssetUsable({ used_ratio: 0.99 })).toBe(true)
+    expect(isAssetUsable({ used_ratio: 0 })).toBe(true)
+  })
+})

apps/web/src/test/pages/assets/usage-badge.test.ts

@@ -0,0 +1,53 @@
+import { describe, expect, it } from "vitest"
+import { getUsageBadge } from "@/pages/assets/types"
+
+describe("getUsageBadge", () => {
+  it("非视频素材不显示角标", () => {
+    expect(getUsageBadge({ kind: "voice", usable: false })).toBeNull()
+    expect(getUsageBadge({ kind: "image", usable: false })).toBeNull()
+  })
+
+  it("字段缺失时不显示角标（降级零影响）", () => {
+    expect(getUsageBadge({ kind: "video" })).toBeNull()
+    expect(getUsageBadge({ kind: "video", usable: true, usedRatio: undefined })).toBeNull()
+  })
+
+  it("usable === false 显示红色实心「已用尽」", () => {
+    expect(getUsageBadge({ kind: "video", usable: false, usedRatio: 1 })).toEqual({
+      label: "已用尽",
+      variant: "exhausted",
+    })
+    // usable === false 优先级最高，即使 usedRatio 字段缺失
+    expect(getUsageBadge({ kind: "video", usable: false })).toEqual({
+      label: "已用尽",
+      variant: "exhausted",
+    })
+  })
+
+  it("used_ratio >= 0.85 显示红色「即将用尽」", () => {
+    expect(getUsageBadge({ kind: "video", usable: true, usedRatio: 0.85 })).toEqual({
+      label: "即将用尽",
+      variant: "warning",
+    })
+    expect(getUsageBadge({ kind: "video", usable: true, usedRatio: 0.97 })).toEqual({
+      label: "即将用尽",
+      variant: "warning",
+    })
+  })
+
+  it("used_ratio >= 0.5 显示橙色「已用 xx%」", () => {
+    expect(getUsageBadge({ kind: "video", usable: true, usedRatio: 0.5 })).toEqual({
+      label: "已用 50%",
+      variant: "ratio",
+    })
+    expect(getUsageBadge({ kind: "video", usable: true, usedRatio: 0.84 })).toEqual({
+      label: "已用 84%",
+      variant: "ratio",
+    })
+  })
+
+  it("used_ratio < 0.5 不显示角标", () => {
+    expect(getUsageBadge({ kind: "video", usable: true, usedRatio: 0.49 })).toBeNull()
+    expect(getUsageBadge({ kind: "video", usable: true, usedRatio: 0 })).toBeNull()
+  })
+})

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, float], tuple[float, float] | None] | None = None,
+    on_exhausted: Callable[[str], None] | None = None,
 ) -> float | None:
     """计算随机 start_time，避开已使用的时间段.
 
@@ -220,10 +220,8 @@ def _calc_random_start_time(
         clip_duration: 片段时长（秒）
         asset_durations: 素材 ID -> 时长映射
         used_segments: {asset_id: [(start1, end1), (start2, end2), ...]} 已使用的时间段
-        on_exhausted: 100 次随机都找不到空闲区间时的受控复用回调，入参为
-            (asset_id, clip_duration)，返回 (start, end) 复用区间或 None。
-            历史记录永不自动清空；回调返回 None（全部达上限/复用占比超闸门）时
-            本函数返回 None，由调用方轮询下一个素材或报错，不做重叠降级。
+        on_exhausted: 100 次随机都找不到空闲区间时的回调（入参 asset_id）。
+            通常用于清空该素材的历史使用记录实现“轮回重置”；回调后会再随机重试一次。
 
     Returns:
         随机 start_time 或 None
@@ -263,36 +261,22 @@ def _calc_random_start_time(
         if not overlap:
             return candidate
 
-    # 100 次都找不到空闲区间：进入受控复用，回调从历史区间中选最久未用且
-    # 使用次数未达上限的区间返回（历史记录永不自动清空）
+    # 100 次都找不到空闲区间：触发轮回重置回调（清空历史使用记录）后再随机重试一次
     if on_exhausted is not None:
         try:
-            reused = on_exhausted(asset_id, clip_duration)
+            on_exhausted(asset_id)
         except Exception:
             logger.warning(
-                "on_exhausted 受控复用回调异常: asset_id=%s",
+                "on_exhausted 轮回重置回调异常: asset_id=%s",
                 asset_id,
                 exc_info=True,
             )
-            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
+        retry = random.uniform(0.0, max_start)
+        if not used_segments or asset_id not in used_segments:
+            return retry
 
-    # 未提供 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)
@@ -301,6 +285,7 @@ def _calc_random_start_time(
         # 返回从最后使用点开始的位置
         return min(last_used_end, max_start)
 
+    # 实在没有空间，返回0（可能会重叠，但至少能执行）
     return 0.0
 
 

tests/unit/test_asset_availability.py

@@ -1,376 +0,0 @@
-"""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,14 +1,12 @@
-"""素材片段使用记录追踪 + 受控复用机制测试（asset_segment_tracker）.
+"""素材片段使用记录追踪服务测试（asset_segment_tracker）.
 
 覆盖：
 - get_used_segments 聚合 metadata 中持久化的区间
-- 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 次避不开时调用复用回调返回历史区间（不再清空历史）
+- record_used_segments 追加记录（不 commit，保留原有 metadata 字段）
+- remove_used_segment 匹配删除（tolerance + plan_id）
+- reset_used_segments 清空轮回（其他字段不动）
+- make_reset_callback 同时清持久化和内存
+- _calc_random_start_time 的 on_exhausted 轮回回调
 """
 
 from __future__ import annotations
@@ -26,12 +24,8 @@ 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_reuse_callback,
+    make_reset_callback,
     record_used_segments,
     remove_used_segment,
     reset_used_segments,
@@ -66,16 +60,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()。"""
+    """模拟 db：db.query(Model).filter(Model.id.in_(ids)).all() / .filter(Model.id == id).first()。
+
+    tracker 模块里的 AssetModel 被 monkeypatch 为 FakeModel 类，
+    这里用挂在类上的伪 column 对象接住 in_ / __eq__。
+    """
 
     class _Col:
         def __init__(self, models):
@@ -99,6 +93,7 @@ class FakeSession:
                 return expr
 
         q = _Q()
+        # 让 tracker 里 AssetModel.id 能取到伪 column
         _model.id = col
         return q
 
@@ -108,27 +103,19 @@ 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 ─────────────────────────────────────────────────────────
 
 
@@ -138,138 +125,124 @@ def test_get_used_segments_aggregates_ranges(patched_model):
             "a1",
             {
                 "used_time_ranges": [
-                    {"start": 1.0, "end": 5.0, "plan_id": "p1", "use_count": 2},
+                    {"start": 1.0, "end": 5.0, "plan_id": "p1"},
                     {"start": 9.0, "end": 12.0, "plan_id": "p2"},
                 ]
             },
         ),
-        "a2": FakeModel("a2", {"other": 1}),
-        "a3": FakeModel("a3"),
+        "a2": FakeModel("a2", {"other": 1}),  # 无区间记录
+        "a3": FakeModel("a3"),  # metadata 为空
     }
     db = _db(models)
-    assert get_used_segments(db, ["a1", "a2", "a3", "missing"]) == {"a1": [(1.0, 5.0), (9.0, 12.0)]}
+
+    result = get_used_segments(db, ["a1", "a2", "a3", "missing"])
+
+    assert result == {"a1": [(1.0, 5.0), (9.0, 12.0)]}
 
 
-def test_get_used_segments_empty(patched_model):
+def test_get_used_segments_empty_input(patched_model):
     assert get_used_segments(_db({}), []) == {}
 
 
 # ── record_used_segments ──────────────────────────────────────────────────────
 
 
-def test_record_appends_new_range_with_use_count_one(patched_model):
-    models = {"a1": FakeModel("a1", {"generation_use_count": 48, "review_status": "pending_review"})}
+def test_record_appends_and_no_commit(patched_model):
+    models = {"a1": FakeModel("a1", {"generation_use_count": 3})}
     db = _db(models)
-    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"
+
+    record_used_segments(db, "a1", 2.0, 6.5, "plan-x")
+
+    meta = models["a1"].meta()
+    assert meta["generation_use_count"] == 3  # 原有字段保留
     ranges = meta["used_time_ranges"]
     assert len(ranges) == 1
-    assert ranges[0]["start"] == 12.5 and ranges[0]["end"] == 20.3
+    assert ranges[0]["start"] == 2.0
+    assert ranges[0]["end"] == 6.5
     assert ranges[0]["plan_id"] == "plan-x"
-    assert ranges[0]["use_count"] == 1
-    assert "created_at" in ranges[0] and "last_used_at" in ranges[0]
-    assert db.commits == 0  # 不自行 commit
+    assert "created_at" in ranges[0]
+    assert db.commits == 0  # 不自行 commit（事务由调用方控制）
 
 
-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",
-                    },
-                ]
-            },
-        )
-    }
+def test_record_multiple_appends_in_order(patched_model):
+    models = {"a1": FakeModel("a1")}
     db = _db(models)
-    # 同一起点复用（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"
+
+    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]
 
 
-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):
+def test_record_missing_asset_is_noop(patched_model):
     db = _db({})
-    record_used_segments(db, "ghost", 1.0, 2.0, "p")  # 不抛异常
+    record_used_segments(db, "ghost", 0.0, 1.0, "p1")  # 不抛异常
 
 
 # ── remove_used_segment ───────────────────────────────────────────────────────
 
 
-def test_remove_matching_range(patched_model):
+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 移动片段仍能按时间匹配删除（防容量泄漏）。"""
     models = {
         "a1": FakeModel(
             "a1",
-            {
-                "used_time_ranges": [
-                    {"start": 1.0, "end": 5.0, "plan_id": "p1"},
-                    {"start": 9.0, "end": 12.0, "plan_id": "p2"},
-                ]
-            },
+            {"used_time_ranges": [{"start": 5.0, "end": 9.0}]},  # 旧记录无 plan_id
         )
     }
     db = _db(models)
-    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
+
+    # 传入 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"] == []
 
 
-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（仅运维/测试）─────────────────────────────────────────
+# ── reset_used_segments ───────────────────────────────────────────────────────
 
 
 def test_reset_clears_ranges_keeps_other_fields(patched_model):
@@ -277,185 +250,74 @@ def test_reset_clears_ranges_keeps_other_fields(patched_model):
         "a1": FakeModel(
             "a1",
             {
-                "generation_use_count": 3,
-                "used_time_ranges": [
-                    {"start": 1.0, "end": 5.0},
-                ],
+                "generation_use_count": 9,
+                "used_time_ranges": [{"start": 1, "end": 2}],
             },
         )
     }
     db = _db(models)
+
     reset_used_segments(db, "a1")
-    meta = json.loads(models["a1"].classification_result)
+
+    meta = models["a1"].meta()
     assert meta["used_time_ranges"] == []
-    assert meta["generation_use_count"] == 3
+    assert meta["generation_use_count"] == 9
+    assert db.commits == 0
 
 
-# ── find_reusable_range：受控复用选择 ─────────────────────────────────────────
+# ── make_reset_callback ───────────────────────────────────────────────────────
 
 
-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",
-                    },  # 最久未用
-                ]
-            },
-        )
-    }
+def test_reset_callback_clears_persisted_and_memory(patched_model):
+    models = {"a1": FakeModel("a1", {"used_time_ranges": [{"start": 0, "end": 30}]})}
     db = _db(models)
-    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
+
+    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"] == []
 
 
-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
+# ── _calc_random_start_time 轮回回调 ──────────────────────────────────────────
 
 
-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)  # 固定候选点必撞区间
-
+def test_calc_random_start_invokes_reset_when_exhausted():
+    """素材区间被占满（100 次随机必重叠）→ 触发 on_exhausted，重置后重试成功。"""
     durations = {"a1": 30.0}
-    used = {"a1": [(0.0, 30.0)]}  # 全占满
-    calls = []
+    used = {"a1": [(0.0, 10.0), (10.0, 20.0), (20.0, 30.0)]}
+    reset_called = []
 
-    def reuse_cb(asset_id, clip_duration):
-        calls.append((asset_id, clip_duration))
-        return (10.0, 18.0)
+    def _on_exhausted(asset_id):
+        reset_called.append(asset_id)
+        used.pop(asset_id, None)  # 模拟轮回清空
 
-    result = _calc_random_start_time("a1", 8.0, durations, used, on_exhausted=reuse_cb)
-    assert calls == [("a1", 8.0)]
-    assert result == 10.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
 
-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
+def test_calc_random_start_no_callback_keeps_legacy_fallback():
+    """不传 on_exhausted 时保持旧降级行为，不报错。"""
+    durations = {"a1": 30.0}
+    used = {"a1": [(0.0, 10.0), (10.0, 20.0), (20.0, 30.0)]}
+
+    result = _calc_random_start_time("a1", 10.0, durations, used)
 
-    monkeypatch.setattr(pgu.random, "uniform", lambda a, b: 0.5)
+    assert result is not None
 
-    durations = {"a1": 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  # 历史记录未被清空
 
+def test_calc_random_start_with_space_does_not_reset():
+    """有充足空闲区间时不触发 reset。"""
+    durations = {"a1": 100.0}
+    used = {"a1": [(0.0, 50.0)]}
+    reset_called = []
 
-def test_calc_random_start_no_callback_backward_compatible(monkeypatch):
-    """不传 on_exhausted 时行为与旧版兼容（100 次失败走降级）。"""
-    import packages.domain.plan_generator_utils as pgu
+    result = _calc_random_start_time("a1", 5.0, durations, used, on_exhausted=lambda aid: reset_called.append(aid))
 
-    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 reset_called == []
     assert result is not None

tests/unit/test_clone_plan_variant.py

@@ -1,201 +0,0 @@
-"""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_reuse_callback",
-            return_value=lambda asset_id, clip_duration: None,
+            "app.api.routes.templates_editor.clips.make_reset_callback",
+            return_value=lambda asset_id: None,
         ),
         patch(
             "app.api.routes.templates_editor.clips.remove_used_segment",
@@ -276,11 +276,7 @@ class TestEditorClipsDurationAndStartTime:
 
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
     def test_zero_duration_asset_raises_400(self, mock_storage):
-        """所有素材时长均为 0 时轮询无可用素材，抛出 400「素材可切区间不足」。
-
-        新轮询逻辑下零时长素材被跳过（而非立即报错）；全部素材都被跳过时
-        返回 400，不创建无效片段。
-        """
+        """素材时长为 0 时应抛出 400，而不是创建无效片段。"""
         from app.api.routes.templates_editor.clips import (
             create_clips_from_assets_editor,
         )
@@ -306,49 +302,7 @@ class TestEditorClipsDurationAndStartTime:
                 )
 
         assert exc_info.value.status_code == 400
-        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)
+        assert "时长" in exc_info.value.detail
 
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
     def test_missing_duration_asset_raises_400(self, mock_storage):
@@ -496,7 +450,7 @@ class TestEditorClipsErrorHandling:
 
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
     def test_none_start_time_raises_400(self, mock_storage):
-        """所有素材 calc 均返回 None（区间耗尽且复用被拒）→ 轮询失败抛 400。"""
+        """_calc_random_start_time 返回 None 时应抛出 HTTPException 400。"""
         from app.api.routes.templates_editor.clips import (
             create_clips_from_assets_editor,
         )
@@ -504,7 +458,7 @@ class TestEditorClipsErrorHandling:
 
         mock_plan_svc = _make_plan_svc()
         mock_asset_repo = MagicMock()
-        # 素材有 duration 但 calc 返回 None（模拟可用区间耗尽、复用被闸门拒绝）
+        # 素材有 duration 但 random 返回 None
         mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
 
         body = ClipsFromAssetsRequest(asset_ids=["a1"])
@@ -529,9 +483,7 @@ class TestEditorClipsErrorHandling:
                 )
 
         assert exc_info.value.status_code == 400
-        assert "素材可切区间不足" in exc_info.value.detail
-        # 复用被拒导致无起点时，不应创建任何片段
-        assert not mock_plan_svc.replace_all_clips_transactional.called
+        assert "时长" in exc_info.value.detail
 
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
     def test_transactional_replace_exception_propagates(self, mock_storage):
@@ -560,190 +512,3 @@ 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,8 +612,6 @@ 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

@@ -1,68 +0,0 @@
-"""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,13 +69,8 @@ class TestRecommendedTimeConflicts:
     def test_conflict_exact_boundary_no_overlap(self):
         from app.api.routes.templates_editor.clips import _recommended_time_conflicts
 
-        # 新语义：默认 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
+        # 推荐 [10, 15]，已用 [0, 10] — 边界相接不算冲突
+        assert _recommended_time_conflicts(10.0, 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
@@ -83,10 +78,8 @@ 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] 与 [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
+        # 推荐 [15, 20] 不冲突
+        assert _recommended_time_conflicts(15.0, 5.0, used) is False
 
 
 # ── _get_mediakit_recommendations 单元测试 ──────────────────────────────────

tests/unit/test_preview_edit_plan_association.py

@@ -13,7 +13,6 @@ 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
 
@@ -181,19 +180,17 @@ class TestPreviewEditPlanAutoAssociation:
             "packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository",
             return_value=fake_plan_repo,
         ):
-            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=""),
-                )
+            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 被设置（自动关联后再克隆为独立 plan）
+        # 找到 store 中的 task 并验证 source_edit_plan_id 被设置
         tasks = list(gen_task_repo._store.values())
         assert len(tasks) == 1
         task = tasks[0]
-        assert task.source_edit_plan_id == "plan-clone-001"
+        assert task.source_edit_plan_id == "plan-auto-001"
 
     @patch(
         "app.api.routes.generation_preview._resolve_strategy_id_from_template",
@@ -212,18 +209,16 @@ class TestPreviewEditPlanAutoAssociation:
         client: TestClient,
         gen_task_repo: StubGenerationTaskRepository,
     ):
-        """前端已传 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"),
-            )
+        """前端已传 source_edit_plan_id 时，不应触发自动关联"""
+        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-clone-explicit"
+        assert tasks[0].source_edit_plan_id == "plan-explicit-001"
 
     @patch(
         "app.api.routes.generation_preview._resolve_strategy_id_from_template",