From 2174e91c48d8a258dc4c2098930cde7e7d649e40 Mon Sep 17 00:00:00 2001
From: CI Bot <ci-bot@xiaoxiajianji.com>
Date: Sat, 29 Aug 2026 20:14:28 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E7=B4=A0=E6=9D=90=E7=89=87=E6=AE=B5?=
 =?UTF-8?q?=E5=8C=BA=E9=97=B4=E6=8C=81=E4=B9=85=E5=8C=96=E5=8E=BB=E9=87=8D?=
 =?UTF-8?q?=20+=20=E5=8C=BA=E9=97=B4=E7=94=A8=E5=B0=BD=E8=87=AA=E5=8A=A8?=
 =?UTF-8?q?=E8=BD=AE=E5=9B=9E?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 新增 asset_segment_tracker 服务：素材 metadata(used_time_ranges) 持久化片段级已用区间
- from-assets 创建片段时读取历史区间，新片段跨任务/跨调用自动避开
- 片段记录与 replace_all_clips_transactional 同事务，失败整体回滚
- MediaKit 异步移动片段起点后同步更新 metadata 区间记录（失败静默）
- _calc_random_start_time 新增 on_exhausted 回调：100次找不到时清空该素材历史区间再重试，实现轮完一圈自动循环

详见 PR body 的 metadata schema 说明
---
 .../app/api/routes/templates_editor/clips.py  |  53 ++-
 .../api/app/services/asset_segment_tracker.py | 171 ++++++++++
 packages/domain/plan_generator_utils.py       |  15 +-
 tests/unit/test_asset_segment_tracker.py      | 310 ++++++++++++++++++
 tests/unit/test_editor_clips_random_start.py  |  30 +-
 5 files changed, 574 insertions(+), 5 deletions(-)
 create mode 100644 apps/api/app/services/asset_segment_tracker.py
 create mode 100644 tests/unit/test_asset_segment_tracker.py

diff --git a/apps/api/app/api/routes/templates_editor/clips.py b/apps/api/app/api/routes/templates_editor/clips.py
index c92d2955d..7e443765f 100755
--- a/apps/api/app/api/routes/templates_editor/clips.py
+++ b/apps/api/app/api/routes/templates_editor/clips.py
@@ -23,6 +23,12 @@ import re
 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 (
+    get_used_segments,
+    make_reset_callback,
+    record_used_segments,
+    remove_used_segment,
+)
 from app.services.edit_plan_service import EditPlanService
 from app.services.edit_template_service import EditTemplateService
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status
@@ -600,7 +606,12 @@ def create_clips_from_assets_editor(
             asset_durations[asset_id] = float(asset.duration or 0.0)
 
     # 3. 在内存中计算所有片段数据（使用随机起始时间，不调用MediaKit）
-    used_segments: dict[str, list[tuple[float, float]]] = {}
+    # 读取素材 metadata 中持久化的历史已用区间（跨任务/跨调用去重），
+    # 格式与 _calc_random_start_time 的 used_segments 参数一致
+    used_segments: dict[str, list[tuple[float, float]]] = get_used_segments(
+        db, unique_asset_ids
+    )
+    reset_cb = make_reset_callback(db, used_segments)
     clips_data: list[dict] = []
 
     for i, (_seg_order, dur_min, dur_max) in enumerate(segments):
@@ -630,7 +641,11 @@ def create_clips_from_assets_editor(
 
         # 使用随机起始时间（不调用MediaKit，保证接口快速返回）
         start_time = _calc_random_start_time(
-            asset_id, clip_duration, asset_durations, used_segments
+            asset_id,
+            clip_duration,
+            asset_durations,
+            used_segments,
+            on_exhausted=reset_cb,
         )
 
         if start_time is None:
@@ -639,10 +654,15 @@ def create_clips_from_assets_editor(
                 detail=f"素材 {asset_id} 时长信息缺失，无法计算起始时间",
             )
 
-        # 记录已使用时间段
+        # 记录已使用时间段（内存，供本次后续片段避开）
         used_segments.setdefault(asset_id, []).append(
             (start_time, start_time + clip_duration)
         )
+        # 同步写入素材 metadata（不 commit，与下方 replace_all_clips_transactional
+        # 处于同一事务，任一步失败整体回滚，不留脏数据）
+        record_used_segments(
+            db, asset_id, start_time, start_time + clip_duration, plan_id
+        )
 
         clips_data.append(
             {
@@ -799,8 +819,35 @@ def _update_mediakit_recommendations_async(  # pragma: no cover
 
             # 逐个更新并捕获异常（单点失败不影响其他片段）
             try:
+                old_start = clip.start_time
+                old_end = old_start + clip_duration
                 plan_svc.update_clip(clip.id, start_time=recommended_start)
                 db.commit()
+                # MediaKit 移动了片段起点 → 同步素材 metadata 的区间记录：
+                # 删除旧区间记录（按 plan_id + 旧 start 匹配），写入新区间。
+                # 异步任务，失败静默，不影响已更新的片段。
+                try:
+                    if remove_used_segment(
+                        db, asset_id, old_start, old_end, plan_id=plan_id
+                    ):
+                        record_used_segments(
+                            db,
+                            asset_id,
+                            recommended_start,
+                            recommended_start + clip_duration,
+                            plan_id,
+                        )
+                        db.commit()
+                except Exception as me:
+                    logger.warning(
+                        "后台任务: 同步素材区间记录失败: clip_id=%s error=%s",
+                        clip.id,
+                        me,
+                    )
+                    try:
+                        db.rollback()
+                    except Exception:
+                        pass
                 updated_count += 1
                 updated_clip_ids.add(clip.id)
             except Exception as ue:
diff --git a/apps/api/app/services/asset_segment_tracker.py b/apps/api/app/services/asset_segment_tracker.py
new file mode 100644
index 000000000..cb485816b
--- /dev/null
+++ b/apps/api/app/services/asset_segment_tracker.py
@@ -0,0 +1,171 @@
+"""素材片段级使用记录追踪.
+
+在素材 metadata（assets.classification_result JSON）中持久化已使用的片段时间区间，
+供 from-assets 创建片段时避开历史区间，实现跨任务/跨调用的片段去重。
+
+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"},
+        ...
+    ]
+
+注意：本模块所有函数都不自行 commit，由调用方控制事务边界
+（from-assets 与 replace_all_clips_transactional 同事务；异步任务各自 commit）。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Callable
+
+from sqlalchemy.orm import Session
+
+from packages.adapters.sqlalchemy_impl.models import AssetModel
+
+logger = logging.getLogger(__name__)
+
+USED_RANGES_KEY = "used_time_ranges"
+
+
+def _read_ranges(model: AssetModel) -> list[dict]:
+    """从 AssetModel 读取 metadata dict（classification_result 列承载的 JSON）."""
+    if not model.classification_result:
+        return {}
+    try:
+        return json.loads(model.classification_result)
+    except Exception:
+        return {}
+
+
+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`` 参数格式一致，可直接传入。
+    """
+    if not asset_ids:
+        return {}
+    result: dict[str, list[tuple[float, float]]] = {}
+    models = db.query(AssetModel).filter(AssetModel.id.in_(list(set(asset_ids)))).all()
+    for model in models:
+        meta = _read_ranges(model)
+        ranges = meta.get(USED_RANGES_KEY) or []
+        segments: list[tuple[float, float]] = []
+        for r in ranges:
+            try:
+                segments.append((float(r["start"]), float(r["end"])))
+            except (KeyError, TypeError, ValueError):
+                continue
+        if segments:
+            result[model.id] = segments
+    return result
+
+
+def record_used_segments(
+    db: Session,
+    asset_id: str,
+    start: float,
+    end: float,
+    plan_id: str,
+) -> None:
+    """向素材 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_ranges(model)
+    ranges = list(meta.get(USED_RANGES_KEY) or [])
+    ranges.append(
+        {
+            "start": round(float(start), 3),
+            "end": round(float(end), 3),
+            "plan_id": plan_id,
+            "created_at": datetime.now(timezone.utc).isoformat(),
+        }
+    )
+    meta[USED_RANGES_KEY] = ranges
+    model.classification_result = json.dumps(meta, ensure_ascii=False)
+    model.updated_at = datetime.now(timezone.utc)
+
+
+def remove_used_segment(
+    db: Session,
+    asset_id: str,
+    start: float,
+    end: float,
+    plan_id: str | None = None,
+    tolerance: float = 0.5,
+) -> bool:
+    """删除素材 metadata 中匹配的一条使用记录（不 commit）.
+
+    匹配规则：start/end 与记录值相差不超过 tolerance 秒；plan_id 非空时还需相等。
+    Returns:
+        是否找到并删除了记录。
+    """
+    model = db.query(AssetModel).filter(AssetModel.id == asset_id).first()
+    if model is None:
+        return False
+    meta = _read_ranges(model)
+    ranges = list(meta.get(USED_RANGES_KEY) or [])
+    remaining: list[dict] = []
+    removed = False
+    for r in ranges:
+        try:
+            match = (
+                abs(float(r["start"]) - float(start)) <= tolerance
+                and abs(float(r["end"]) - float(end)) <= tolerance
+            )
+        except (KeyError, TypeError, ValueError):
+            remaining.append(r)
+            continue
+        if plan_id is not None and r.get("plan_id") != plan_id:
+            match = False
+        if match and not removed:
+            removed = True
+            continue
+        remaining.append(r)
+    if removed:
+        meta[USED_RANGES_KEY] = remaining
+        model.classification_result = json.dumps(meta, ensure_ascii=False)
+        model.updated_at = datetime.now(timezone.utc)
+    return removed
+
+
+def reset_used_segments(db: Session, asset_id: str) -> None:
+    """清空单个素材的历史片段使用记录（不 commit）.
+
+    单个素材的可用区间被全部占用（轮回一圈）后调用，使后续片段可重新使用整段素材。
+    """
+    model = db.query(AssetModel).filter(AssetModel.id == asset_id).first()
+    if model is None:
+        return
+    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)
+
+
+def make_reset_callback(db: Session, used_segments: dict) -> Callable[[str], None]:
+    """构造给 _calc_random_start_time 用的 reset 回调.
+
+    回调同时清空持久化 metadata 和内存中的 used_segments，使重试随机能覆盖全素材。
+    """
+
+    def _reset(asset_id: str) -> None:
+        try:
+            reset_used_segments(db, asset_id)
+        except Exception:
+            logger.warning("[片段追踪] reset 持久化记录失败: asset_id=%s", asset_id, exc_info=True)
+        used_segments.pop(asset_id, None)
+
+    return _reset
diff --git a/packages/domain/plan_generator_utils.py b/packages/domain/plan_generator_utils.py
index d87ebf2af..4ba918fc3 100755
--- a/packages/domain/plan_generator_utils.py
+++ b/packages/domain/plan_generator_utils.py
@@ -12,7 +12,7 @@
 from __future__ import annotations
 
 import random
-from typing import List
+from typing import Callable, List
 
 from packages.domain.edit_plan_clip import EditPlanClip
 from packages.domain.editing_mode import EditingMode
@@ -204,6 +204,7 @@ def _calc_random_start_time(
     clip_duration: float,
     asset_durations: dict[str, float] | None,
     used_segments: dict[str, list[tuple[float, float]]] | None = None,
+    on_exhausted: Callable[[str], None] | None = None,
 ) -> float | None:
     """计算随机 start_time，避开已使用的时间段.
 
@@ -216,6 +217,8 @@ def _calc_random_start_time(
         clip_duration: 片段时长（秒）
         asset_durations: 素材 ID -> 时长映射
         used_segments: {asset_id: [(start1, end1), (start2, end2), ...]} 已使用的时间段
+        on_exhausted: 100 次随机都找不到空闲区间时的回调（入参 asset_id）。
+            通常用于清空该素材的历史使用记录实现“轮回重置”；回调后会再随机重试一次。
 
     Returns:
         随机 start_time 或 None
@@ -255,6 +258,16 @@ def _calc_random_start_time(
         if not overlap:
             return candidate
 
+    # 100 次都找不到空闲区间：触发轮回重置回调（清空历史使用记录）后再随机重试一次
+    if on_exhausted is not None:
+        try:
+            on_exhausted(asset_id)
+        except Exception:
+            pass
+        retry = random.uniform(0.0, max_start)
+        if not used_segments or asset_id not in used_segments:
+            return retry
+
     # 如果尝试多次仍找不到，缩短时长使用素材末尾
     # 找到最后一个已使用段之后的可用空间
     last_used_end = 0.0
diff --git a/tests/unit/test_asset_segment_tracker.py b/tests/unit/test_asset_segment_tracker.py
new file mode 100644
index 000000000..043e51b1f
--- /dev/null
+++ b/tests/unit/test_asset_segment_tracker.py
@@ -0,0 +1,310 @@
+"""素材片段使用记录追踪服务测试（asset_segment_tracker）.
+
+覆盖：
+- get_used_segments 聚合 metadata 中持久化的区间
+- record_used_segments 追加记录（不 commit，保留原有 metadata 字段）
+- remove_used_segment 匹配删除（tolerance + plan_id）
+- reset_used_segments 清空轮回（其他字段不动）
+- make_reset_callback 同时清持久化和内存
+- _calc_random_start_time 的 on_exhausted 轮回回调
+"""
+
+from __future__ import annotations
+
+import json
+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"))
+
+import pytest
+
+from app.services import asset_segment_tracker as ast
+from app.services.asset_segment_tracker import (
+    get_used_segments,
+    make_reset_callback,
+    record_used_segments,
+    remove_used_segment,
+    reset_used_segments,
+)
+from packages.domain.plan_generator_utils import _calc_random_start_time
+
+
+class FakeModel:
+    """模拟 AssetModel：id + classification_result（JSON Text）+ updated_at。"""
+
+    def __init__(self, asset_id: str, meta: dict | None = None):
+        self.id = asset_id
+        self.classification_result = json.dumps(meta, ensure_ascii=False) if meta else None
+        self.updated_at = None
+
+    def meta(self) -> dict:
+        return json.loads(self.classification_result) if self.classification_result else {}
+
+
+class _InExpr:
+    def __init__(self, ids, models):
+        self._ids = ids
+        self._models = models
+
+    def all(self):
+        return [self._models[i] for i in self._ids if i in self._models]
+
+
+class _EqExpr:
+    def __init__(self, target_id, models):
+        self._target_id = target_id
+        self._models = models
+
+    def first(self):
+        return self._models.get(self._target_id)
+
+
+class FakeSession:
+    """模拟 db：db.query(Model).filter(Model.id.in_(ids)).all() / .filter(Model.id == id).first()。
+
+    tracker 模块里的 AssetModel 被 monkeypatch 为 FakeModel 类，
+    这里用挂在类上的伪 column 对象接住 in_ / __eq__。
+    """
+
+    class _Col:
+        def __init__(self, models):
+            self._models = models
+
+        def in_(self, ids):
+            return _InExpr(list(ids), self._models)
+
+        def __eq__(self, other):
+            return _EqExpr(other, self._models)
+
+    def __init__(self, models: dict[str, FakeModel]):
+        self._models = models
+        self.commits = 0
+
+    def query(self, _model):
+        col = self._Col(self._models)
+
+        class _Q:
+            def filter(self_inner, expr):
+                return expr
+
+        q = _Q()
+        # 让 tracker 里 AssetModel.id 能取到伪 column
+        _model.id = col
+        return q
+
+    def commit(self):
+        self.commits += 1
+
+
+@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)
+
+
+# ── get_used_segments ─────────────────────────────────────────────────────────
+
+
+def test_get_used_segments_aggregates_ranges(patched_model):
+    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"},
+        ]}),
+        "a2": FakeModel("a2", {"other": 1}),            # 无区间记录
+        "a3": FakeModel("a3"),                           # metadata 为空
+    }
+    db = _db(models)
+
+    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_input(patched_model):
+    assert get_used_segments(_db({}), []) == {}
+
+
+# ── record_used_segments ──────────────────────────────────────────────────────
+
+
+def test_record_appends_and_no_commit(patched_model):
+    models = {"a1": FakeModel("a1", {"generation_use_count": 3})}
+    db = _db(models)
+
+    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"] == 2.0
+    assert ranges[0]["end"] == 6.5
+    assert ranges[0]["plan_id"] == "plan-x"
+    assert "created_at" in ranges[0]
+    assert db.commits == 0  # 不自行 commit（事务由调用方控制）
+
+
+def test_record_multiple_appends_in_order(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")
+
+    ranges = models["a1"].meta()["used_time_ranges"]
+    assert [r["start"] for r in ranges] == [0.0, 10.0]
+
+
+def test_record_missing_asset_is_noop(patched_model):
+    db = _db({})
+    record_used_segments(db, "ghost", 0.0, 1.0, "p1")  # 不抛异常
+
+
+# ── remove_used_segment ───────────────────────────────────────────────────────
+
+
+def test_remove_matching_segment(patched_model):
+    models = {"a1": FakeModel("a1")}
+    db = _db(models)
+    record_used_segments(db, "a1", 0.0, 4.0, "p1")
+    record_used_segments(db, "a1", 10.0, 14.0, "p1")
+
+    removed = remove_used_segment(db, "a1", 0.0, 4.0, plan_id="p1")
+
+    assert removed is True
+    ranges = models["a1"].meta()["used_time_ranges"]
+    assert len(ranges) == 1
+    assert ranges[0]["start"] == 10.0
+
+
+def test_remove_not_found_returns_false(patched_model):
+    models = {"a1": FakeModel("a1")}
+    db = _db(models)
+    record_used_segments(db, "a1", 0.0, 4.0, "p1")
+
+    assert remove_used_segment(db, "a1", 99.0, 100.0, plan_id="p1") is False
+
+
+def test_remove_respects_tolerance(patched_model):
+    models = {
+        "a1": FakeModel("a1", {"used_time_ranges": [
+            {"start": 5.0, "end": 9.0, "plan_id": "p1"}
+        ]}),
+        "a2": FakeModel("a2", {"used_time_ranges": [
+            {"start": 5.0, "end": 9.0, "plan_id": "p1"}
+        ]}),
+    }
+    db = _db(models)
+
+    # 偏差 0.3 秒，在 tolerance=0.5 内 → 删除成功
+    assert remove_used_segment(db, "a1", 5.3, 8.7, plan_id="p1") is True
+    # 偏差 2 秒，超出 tolerance → 删除失败
+    assert remove_used_segment(db, "a2", 7.0, 11.0, plan_id="p1") is False
+
+
+def test_remove_plan_id_must_match(patched_model):
+    models = {"a1": FakeModel("a1", {"used_time_ranges": [
+        {"start": 5.0, "end": 9.0, "plan_id": "plan-A"}
+    ]})}
+    db = _db(models)
+
+    # 时间匹配但 plan_id 不同 → 不删除
+    assert remove_used_segment(db, "a1", 5.0, 9.0, plan_id="plan-B") is False
+    assert len(models["a1"].meta()["used_time_ranges"]) == 1
+
+
+# ── reset_used_segments ───────────────────────────────────────────────────────
+
+
+def test_reset_clears_ranges_keeps_other_fields(patched_model):
+    models = {"a1": FakeModel("a1", {
+        "generation_use_count": 9,
+        "used_time_ranges": [{"start": 1, "end": 2}],
+    })}
+    db = _db(models)
+
+    reset_used_segments(db, "a1")
+
+    meta = models["a1"].meta()
+    assert meta["used_time_ranges"] == []
+    assert meta["generation_use_count"] == 9
+    assert db.commits == 0
+
+
+# ── make_reset_callback ───────────────────────────────────────────────────────
+
+
+def test_reset_callback_clears_persisted_and_memory(patched_model):
+    models = {"a1": FakeModel("a1", {"used_time_ranges": [
+        {"start": 0, "end": 30}
+    ]})}
+    db = _db(models)
+
+    used_segments = {"a1": [(0.0, 30.0)], "a2": [(1.0, 2.0)]}
+    cb = make_reset_callback(db, used_segments)
+    cb("a1")
+
+    assert "a1" not in used_segments     # 内存清空
+    assert "a2" in used_segments         # 其他素材不受影响
+    assert models["a1"].meta()["used_time_ranges"] == []
+
+
+# ── _calc_random_start_time 轮回回调 ──────────────────────────────────────────
+
+
+def test_calc_random_start_invokes_reset_when_exhausted():
+    """素材区间被占满（100 次随机必重叠）→ 触发 on_exhausted，重置后重试成功。"""
+    durations = {"a1": 30.0}
+    used = {"a1": [(0.0, 10.0), (10.0, 20.0), (20.0, 30.0)]}
+    reset_called = []
+
+    def _on_exhausted(asset_id):
+        reset_called.append(asset_id)
+        used.pop(asset_id, None)  # 模拟轮回清空
+
+    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_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)
+
+    assert result is not None
+
+
+def test_calc_random_start_with_space_does_not_reset():
+    """有充足空闲区间时不触发 reset。"""
+    durations = {"a1": 100.0}
+    used = {"a1": [(0.0, 50.0)]}
+    reset_called = []
+
+    result = _calc_random_start_time(
+        "a1", 5.0, durations, used, on_exhausted=lambda aid: reset_called.append(aid)
+    )
+
+    assert reset_called == []
+    assert result is not None
diff --git a/tests/unit/test_editor_clips_random_start.py b/tests/unit/test_editor_clips_random_start.py
index 60ed288dd..459200a92 100644
--- a/tests/unit/test_editor_clips_random_start.py
+++ b/tests/unit/test_editor_clips_random_start.py
@@ -81,6 +81,34 @@ def _get_clips_data_from_call(mock_plan_svc):
     return call_args.kwargs.get("clips_data", [])
 
 
+@pytest.fixture(autouse=True)
+def _mock_segment_tracker():
+    """from-assets 现在会读/写素材 metadata 的片段区间记录，测试中 mock 掉避免依赖真实 DB。
+
+    get_used_segments 返回空 dict（等价历史行为：无历史区间）；
+    record/remove/reset 回调均无副作用。
+    """
+    with (
+        patch(
+            "app.api.routes.templates_editor.clips.get_used_segments",
+            return_value={},
+        ),
+        patch(
+            "app.api.routes.templates_editor.clips.record_used_segments",
+            return_value=None,
+        ),
+        patch(
+            "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",
+            return_value=False,
+        ),
+    ):
+        yield
+
+
 class TestEditorClipsBySegments:
     """测试按 segment 数量创建片段 + 素材轮询。"""
 
@@ -387,7 +415,7 @@ class TestEditorClipsDurationAndStartTime:
 
         captured_used_segments = []
 
-        def fake_calc(asset_id, clip_duration, asset_durations, used_segments):
+        def fake_calc(asset_id, clip_duration, asset_durations, used_segments, on_exhausted=None):
             captured_used_segments.append({aid: list(segs) for aid, segs in (used_segments or {}).items()})
             return (len(captured_used_segments) - 1) * 5.0