fix(#1743): smart-match排序随机噪声 + 素材使用次数按成片实际片段计数 (#1747)

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

@@ -123,6 +123,7 @@ def _select_assets_from_library(
     assets: list,
     mode: str,
     count: int,
+    rng=None,
 ) -> list[str]:
     """根据选取模式从素材库中选取 ready 状态的视频素材 ID。
 
@@ -130,6 +131,8 @@ def _select_assets_from_library(
         assets: 素材库中所有素材（Asset 实体列表）
         mode: 选取模式 — all=全部, smart=智能匹配（多维度评分+多样性）
         count: 选取数量，0 表示全部（仅 smart 模式有效）
+        rng: 可选随机源（smart 模式排序噪声用），生产环境不传则内部随机；
+            测试可注入固定种子或零噪声随机源获得确定性结果。
 
     Returns:
         选中的素材 ID 列表
@@ -142,8 +145,9 @@ def _select_assets_from_library(
     if mode == "smart":
         # 智能匹配：统一使用 packages/domain/smart_match.py 的多维评分+多样性选取
         # 评分维度：质量分(40%) + 时长适配(30%) + 新鲜度(20%) + 未使用加分(10%)
+        # 排序注入随机噪声（#1743）：同分素材每次选出不同组合，从素材组合层面降重
         limit = count if count > 0 else None
-        results = smart_select_assets(ready_video_assets, limit=limit, kind="video")
+        results = smart_select_assets(ready_video_assets, limit=limit, kind="video", rng=rng)
         return [r.asset.id for r in results]
 
     # 默认 all 模式：返回全部 ready 视频素材

apps/worker/worker_app/core/asset_usage.py

@@ -1,10 +1,20 @@
 from datetime import datetime, timezone
 
 
-def mark_asset_used_for_generation(asset) -> None:
+def mark_asset_used_for_generation(asset, times: int = 1) -> None:
+    """标记素材在成片中被使用，累加使用次数。
+
+    Args:
+        asset: Asset 实体（metadata 就地更新）
+        times: 本次成片实际使用次数（= 最终成片 plan 中引用该素材的片段数）。
+            按「成片实际渲染的片段」计数而非请求传入的 asset_ids 列表——
+            请求列表可能含未被 plan 选用的素材（不应计数），同一素材在多片段
+            复用时应按片段数累加（高频排除/未使用偏好才与真实渲染强度挂钩）。
+    """
+    times = max(1, int(times or 1))
     asset.metadata = {
         **asset.metadata,
-        "generation_use_count": int(asset.metadata.get("generation_use_count") or 0) + 1,
+        "generation_use_count": int(asset.metadata.get("generation_use_count") or 0) + times,
         "last_used_at": datetime.now(timezone.utc).isoformat(),
         "review_status": asset.metadata.get("review_status") or "pending_review",
     }

apps/worker/worker_app/tasks/generation.py

@@ -416,6 +416,76 @@ def pick_batch_cover_index(task_id: str, candidate_count: int) -> int:
     return int(hashlib.md5(task_id.encode()).hexdigest(), 16) % candidate_count
 
 
+def _count_plan_clip_asset_usage(session, plan_id: str) -> dict[str, int]:
+    """统计最终成片 plan 中每个素材被片段引用的次数。
+
+    计数口径（#1743）：以成片实际渲染的 edit_plan_clips 为准——
+    同一素材在多个片段复用按片段数累加；未被 plan 选用的素材（即使
+    出现在请求 asset_ids 中）不计数。
+    """
+    from packages.adapters.sqlalchemy_impl.models import EditPlanClipModel
+
+    rows = session.query(EditPlanClipModel.asset_id).filter(EditPlanClipModel.plan_id == plan_id).all()
+    counts: dict[str, int] = {}
+    for (asset_id,) in rows:
+        if asset_id:
+            counts[asset_id] = counts.get(asset_id, 0) + 1
+    return counts
+
+
+def _record_rendered_asset_usage(
+    session,
+    plan_id: str,
+    task_id: str,
+    fallback_asset_ids: list[str] | None = None,
+) -> int:
+    """按最终成片 plan 的实际片段统计素材使用次数并回写 metadata。
+
+    plan 无有效片段素材（异常数据）时退回 fallback_asset_ids 每个计 1 次，
+    保证使用统计不因数据异常完全丢失。单素材回写失败不影响其他素材。
+
+    Returns: 实际回写次数的素材数量。
+    """
+    from worker_app.core.asset_usage import mark_asset_used_for_generation
+
+    from packages.adapters.sqlalchemy_impl.asset_repository import (
+        SQLAlchemyAssetRepository,
+    )
+
+    used_counts = _count_plan_clip_asset_usage(session, plan_id)
+    if not used_counts and fallback_asset_ids:
+        used_counts = {aid: 1 for aid in fallback_asset_ids if aid}
+    if not used_counts:
+        logger.info("[task_id=%s] 素材使用计数: plan=%s 无有效片段素材，跳过", task_id, plan_id)
+        return 0
+
+    asset_repo = SQLAlchemyAssetRepository(session)
+    written = 0
+    for aid, times in used_counts.items():
+        try:
+            asset = asset_repo.get(aid)
+            if asset:
+                mark_asset_used_for_generation(asset, times=times)
+                asset_repo.update(asset)
+                written += 1
+        except Exception:
+            logger.warning(
+                "[task_id=%s] 更新素材使用次数失败: asset_id=%s times=%d",
+                task_id,
+                aid,
+                times,
+                exc_info=True,
+            )
+    logger.info(
+        "[task_id=%s] 素材使用计数回写完成(plan=%s): %d 个素材, 片段引用 %d 次",
+        task_id,
+        plan_id,
+        written,
+        sum(used_counts.values()),
+    )
+    return written
+
+
 def _upload_rendered_video(
     task_id: str,
     output_path: Path,
@@ -994,35 +1064,22 @@ def generate_video(self, task_id: str) -> dict:
                     logger.warning("[task_id=%s] 更新标题使用次数异常", task_id, exc_info=True)
 
                 # 5.2 更新素材使用次数
+                # 按「最终成片 plan 实际渲染的片段」计数（#1743）：不用请求传入的
+                # task.asset_ids（可能含未被 plan 选用的素材），同一素材多片段复用
+                # 按片段数累加，使 unused_bonus / 高频排除与真实渲染强度挂钩。
+                # current_plan_id 是重渲循环结束后最终成片所用 plan（首版或重渲版）。
                 try:
-                    from worker_app.core.asset_usage import mark_asset_used_for_generation
-
                     _asset_session = SessionLocal()
                     try:
-                        from packages.adapters.sqlalchemy_impl.asset_repository import (
-                            SQLAlchemyAssetRepository,
-                        )
                         from packages.adapters.sqlalchemy_impl.generation_task_repository import (
                             SQLAlchemyGenerationTaskRepository,
                         )
 
-                        _task_repo = SQLAlchemyGenerationTaskRepository(_asset_session)
-                        _asset_repo = SQLAlchemyAssetRepository(_asset_session)
-                        _gen_task = _task_repo.get(task_id)
-                        if _gen_task and _gen_task.asset_ids:
-                            for _aid in _gen_task.asset_ids:
-                                try:
-                                    _asset = _asset_repo.get(_aid)
-                                    if _asset:
-                                        mark_asset_used_for_generation(_asset)
-                                        _asset_repo.update(_asset)
-                                except Exception:
-                                    logger.warning(
-                                        "[task_id=%s] 更新素材使用次数失败: asset_id=%s",
-                                        task_id,
-                                        _aid,
-                                        exc_info=True,
-                                    )
+                        _fallback_ids: list[str] = []
+                        _gt_for_assets = SQLAlchemyGenerationTaskRepository(_asset_session).get(task_id)
+                        if _gt_for_assets:
+                            _fallback_ids = list(_gt_for_assets.asset_ids or [])
+                        _record_rendered_asset_usage(_asset_session, current_plan_id, task_id, _fallback_ids)
                     finally:
                         _asset_session.close()
                 except Exception:

packages/domain/smart_match.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import math
+import random
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from typing import Any
@@ -130,6 +131,7 @@ def smart_select_assets(
     limit: int | None = None,
     kind: str | None = None,
     now: datetime | None = None,
+    rng: random.Random | None = None,
 ) -> list[SmartMatchResult]:
     """从素材列表中智能选取素材。
 
@@ -138,9 +140,11 @@ def smart_select_assets(
         limit: 最大返回数量，None 表示不限制
         kind: 按文件类型过滤（video/image/audio），None 表示不过滤
         now: 当前时间（用于测试注入）
+        rng: 随机数生成器（用于测试注入，控制排序噪声可复现）
 
     Returns:
-        按得分降序排列的 SmartMatchResult 列表
+        按有效得分（综合得分 + 随机噪声）降序排列的 SmartMatchResult 列表。
+        r.score 始终为无噪声的原始综合得分；噪声仅用于排序/分桶顺序。
     """
     # Step 1: 过滤 ready 状态
     ready_assets = [a for a in assets if _get_enum_value(a, "status") == "ready"]
@@ -158,22 +162,39 @@ def smart_select_assets(
         total, breakdown = score_asset(a, now=now)
         scored.append(SmartMatchResult(asset=a, score=total, breakdown=breakdown))
 
-    # Step 4: 按得分降序排序
-    scored.sort(key=lambda r: r.score, reverse=True)
+    # Step 4: 按「得分 + 随机噪声」降序排序
+    # 同分/近分素材（分差 <= SCORE_RANDOM_NOISE_MAX）每次选出的顺序与组合不同，
+    # 从素材组合层面降低成片重复率；分差显著（>20）的高质量素材排名不受影响。
+    # 噪声以 asset.id 为 key 缓存，保证同一次调用内排序与分桶轮询顺序一致。
+    rng = rng or random.Random()
+    noise_by_asset: dict[str, float] = {
+        getattr(a, "id", ""): rng.uniform(0.0, SCORE_RANDOM_NOISE_MAX) for a in ready_assets
+    }
 
-    # Step 5: 多样性保障 — 时长分桶均衡选取
+    def _effective(r: SmartMatchResult) -> float:
+        return r.score + noise_by_asset.get(getattr(r.asset, "id", ""), 0.0)
+
+    scored.sort(key=_effective, reverse=True)
+
+    # Step 5: 多样性保障 — 时长分桶均衡选取（桶内同样按含噪声顺序）
     if limit and limit > 0 and len(scored) > limit:
-        scored = _diversity_select(scored, limit)
+        scored = _diversity_select(scored, limit, effective_key=_effective)
     elif limit and limit > 0:
         scored = scored[:limit]
 
     return scored
 
 
-def _diversity_select(scored: list[SmartMatchResult], limit: int) -> list[SmartMatchResult]:
+def _diversity_select(
+    scored: list[SmartMatchResult],
+    limit: int,
+    effective_key: Any | None = None,
+) -> list[SmartMatchResult]:
     """从已排序的候选中按分桶均衡选取，避免全选中同一时长档。
 
-    策略：轮流从 short/medium/long 桶中按得分顺序取，直到凑满 limit。
+    策略：轮流从 short/medium/long 桶中按顺序取，直到凑满 limit。
+    scored 已按含噪声的有效得分排序，桶内直接继承该顺序；
+    effective_key 给出时最终输出也按有效得分排序（同一次调用内噪声一致）。
     """
     buckets: dict[str, list[SmartMatchResult]] = {
         "short": [],
@@ -209,6 +230,6 @@ def _diversity_select(scored: list[SmartMatchResult], limit: int) -> list[SmartM
         if not added:
             break
 
-    # 按原始得分降序输出
-    selected.sort(key=lambda r: r.score, reverse=True)
+    # 按有效得分（含噪声）降序输出；未传 effective_key 时退回原始得分
+    selected.sort(key=effective_key or (lambda r: r.score), reverse=True)
     return selected

tests/unit/test_asset_availability.py

@@ -420,8 +420,9 @@ class TestSmartMatchFlatStructure:
         """item.id 直接在顶层可读，不存在 item.asset 包装层。"""
         assets = [_fresh_asset("a-flat-1"), _fresh_asset("a-flat-2")]
         resp = TestSmartMatchFiltersExhausted()._call(assets)
-        ids = [item.id for item in resp.items]
-        assert ids == ["a-flat-1", "a-flat-2"]
+        ids = {item.id for item in resp.items}
+        # 同分素材排序含随机噪声（#1743），只断言集合不断言顺序
+        assert ids == {"a-flat-1", "a-flat-2"}
         # 嵌套 asset 字段已移除
         assert all(not hasattr(item, "asset") for item in resp.items)
 

tests/unit/test_asset_select_mode.py

@@ -12,6 +12,7 @@
 
 from __future__ import annotations
 
+import random
 import sys
 from pathlib import Path
 
@@ -19,6 +20,20 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
 
 from app.api.routes.generation_tasks import _select_assets_from_library
 
+from packages.domain.smart_match import SCORE_RANDOM_NOISE_MAX
+
+
+class _ZeroNoiseRandom(random.Random):
+    """零噪声随机源：uniform(0, NOISE_MAX) 恒返回 0，smart 排序确定可复现。"""
+
+    def uniform(self, a, b):
+        if a == 0.0 and b == SCORE_RANDOM_NOISE_MAX:
+            return 0.0
+        return super().uniform(a, b)
+
+
+_ZERO_NOISE = _ZeroNoiseRandom(0)
+
 from packages.domain import Asset, AssetStatus
 
 
@@ -106,7 +121,7 @@ class TestSelectAssetsSmartMode:
             _asset("high", "high.mp4", quality_score=90),
             _asset("mid", "mid.mp4", quality_score=60),
         ]
-        result = _select_assets_from_library(assets, mode="smart", count=0)
+        result = _select_assets_from_library(assets, mode="smart", count=0, rng=_ZERO_NOISE)
         assert result == ["high", "mid", "low"]
 
     def test_smart_duration_optimal_beats_too_short(self):
@@ -115,7 +130,7 @@ class TestSelectAssetsSmartMode:
             _asset("too_short", "short.mp4", quality_score=80, duration=1.0),
             _asset("optimal", "optimal.mp4", quality_score=80, duration=15.0),
         ]
-        result = _select_assets_from_library(assets, mode="smart", count=0)
+        result = _select_assets_from_library(assets, mode="smart", count=0, rng=_ZERO_NOISE)
         # Both: quality=80*0.4=32, recency/unused equal
         # optimal(15s): duration_fitness=30 → total=62+
         # too_short(1s): duration_fitness=20+(1/5)*80=36 → 36*0.3=10.8 → total=42.8+
@@ -127,7 +142,7 @@ class TestSelectAssetsSmartMode:
             _asset("a2", "v2.mp4", quality_score=70),
             _asset("a3", "v3.mp4", quality_score=50),
         ]
-        result = _select_assets_from_library(assets, mode="smart", count=2)
+        result = _select_assets_from_library(assets, mode="smart", count=2, rng=_ZERO_NOISE)
         assert result == ["a1", "a2"]
 
     def test_smart_null_quality_treated_as_default(self):
@@ -136,7 +151,7 @@ class TestSelectAssetsSmartMode:
             _asset("scored", "scored.mp4", quality_score=80),
             _asset("unscored", "unscored.mp4", quality_score=None),
         ]
-        result = _select_assets_from_library(assets, mode="smart", count=0)
+        result = _select_assets_from_library(assets, mode="smart", count=0, rng=_ZERO_NOISE)
         # scored(80): quality=80*0.4=32; unscored(None→50): quality=50*0.4=20
         assert result == ["scored", "unscored"]
 
@@ -146,7 +161,7 @@ class TestSelectAssetsSmartMode:
             _asset("a2", "v2.mp4", quality_score=90),
             _asset("a3", "v3.mp4", quality_score=50),
         ]
-        result = _select_assets_from_library(assets, mode="smart", count=0)
+        result = _select_assets_from_library(assets, mode="smart", count=0, rng=_ZERO_NOISE)
         assert result == ["a2", "a3", "a1"]
 
 

tests/unit/test_asset_usage_count_1743.py

@@ -0,0 +1,194 @@
+"""#1743 素材使用次数回写测试。
+
+计数口径修复：
+- mark_asset_used_for_generation 支持 times 参数，按成片实际片段引用次数累加
+- _count_plan_clip_asset_usage 从最终成片 plan 的 clips 统计 {asset_id: 片段引用次数}
+- _record_rendered_asset_usage 回写 metadata：plan clips 为准、空 plan 兜底任务 asset_ids、
+  未被 plan 选用的素材不计数、单素材失败不影响其他素材
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+for sub in ("apps/worker", "apps/api", "packages", ""):
+    p = str(REPO_ROOT / sub) if sub else str(REPO_ROOT)
+    if p not in sys.path:
+        sys.path.insert(0, p)
+
+import worker_app.tasks.generation as gen_mod  # noqa: E402
+from worker_app.core.asset_usage import mark_asset_used_for_generation  # noqa: E402
+
+
+class FakeAsset:
+    def __init__(self, aid: str, metadata: dict | None = None):
+        self.id = aid
+        self.metadata = metadata or {}
+
+
+# ── mark_asset_used_for_generation ──────────────────────────────────────────
+
+
+class TestMarkAssetUsed:
+    def test_default_times_is_one(self):
+        a = FakeAsset("a1", metadata={})
+        mark_asset_used_for_generation(a)
+        assert a.metadata["generation_use_count"] == 1
+        assert "last_used_at" in a.metadata
+
+    def test_times_accumulates_by_clip_count(self):
+        """同一素材被 3 个片段引用 → 一次回写 +3。"""
+        a = FakeAsset("a1", metadata={"generation_use_count": 2})
+        mark_asset_used_for_generation(a, times=3)
+        assert a.metadata["generation_use_count"] == 5
+
+    def test_times_zero_or_negative_floored_to_one(self):
+        a = FakeAsset("a1", metadata={})
+        mark_asset_used_for_generation(a, times=0)
+        assert a.metadata["generation_use_count"] == 1
+
+    def test_preserves_existing_metadata(self):
+        a = FakeAsset("a1", metadata={"tags": ["travel"], "generation_use_count": 4})
+        mark_asset_used_for_generation(a, times=2)
+        assert a.metadata["tags"] == ["travel"]
+        assert a.metadata["generation_use_count"] == 6
+
+
+# ── _count_plan_clip_asset_usage ────────────────────────────────────────────
+
+
+def _mock_session_with_clips(clip_asset_ids: list[str]):
+    """构造 mock session：query(EditPlanClipModel).filter().all() 返回片段 asset_id 行。"""
+    rows = [(aid,) for aid in clip_asset_ids]
+    session = MagicMock()
+    session.query.return_value.filter.return_value.all.return_value = rows
+    return session
+
+
+class TestCountPlanClipUsage:
+    def test_counts_asset_occurrences_across_clips(self):
+        """plan clips 中 asset-x 出现 2 次、asset-y 1 次 → {x:2, y:1}。"""
+        session = _mock_session_with_clips(["asset-x", "asset-y", "asset-x", ""])
+        with _patch_clip_model():
+            counts = gen_mod._count_plan_clip_asset_usage(session, "plan-1")
+        assert counts == {"asset-x": 2, "asset-y": 1}
+
+    def test_empty_clips_returns_empty(self):
+        session = _mock_session_with_clips([])
+        with _patch_clip_model():
+            counts = gen_mod._count_plan_clip_asset_usage(session, "plan-9")
+        assert counts == {}
+
+    def test_blank_asset_ids_skipped(self):
+        session = _mock_session_with_clips(["", None, "asset-z"])
+        with _patch_clip_model():
+            counts = gen_mod._count_plan_clip_asset_usage(session, "plan-1")
+        assert counts == {"asset-z": 1}
+
+
+class _patch_clip_model:
+    """patch generation 模块内 EditPlanClipModel 的导入路径（函数内 import）。"""
+
+    def __enter__(self):
+        import unittest.mock as mock
+
+        self._patches = [
+            mock.patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel", MagicMock()),
+        ]
+        for p in self._patches:
+            p.start()
+        return self
+
+    def __exit__(self, *exc):
+        for p in self._patches:
+            p.stop()
+        return False
+
+
+# ── _record_rendered_asset_usage ────────────────────────────────────────────
+
+
+class TestRecordRenderedAssetUsage:
+    def _run(self, clip_ids, fallback_ids=None, repo_get_override=None, repo_update_side=None):
+        session = _mock_session_with_clips(clip_ids)
+        assets: dict[str, FakeAsset] = {}
+
+        def fake_repo_init(sess):
+            repo = MagicMock()
+
+            def fake_get(aid):
+                if repo_get_override and aid in repo_get_override:
+                    return repo_get_override[aid]
+                return assets.setdefault(aid, FakeAsset(aid, metadata={}))
+
+            repo.get.side_effect = fake_get
+            if repo_update_side:
+                repo.update.side_effect = repo_update_side
+            else:
+                repo.update.side_effect = lambda a: a
+            return repo
+
+        import unittest.mock as mock
+
+        with (
+            mock.patch(
+                "packages.adapters.sqlalchemy_impl.asset_repository.SQLAlchemyAssetRepository",
+                side_effect=fake_repo_init,
+            ),
+            _patch_clip_model(),
+        ):
+            written = gen_mod._record_rendered_asset_usage(session, "plan-1", "task-1", fallback_asset_ids=fallback_ids)
+        return written, assets
+
+    def test_writes_by_plan_clips_not_request_asset_ids(self):
+        """核心口径：plan clips 用 x×2 + y×1；请求里的 z（未被 plan 选用）不计数。"""
+        written, assets = self._run(
+            clip_ids=["asset-x", "asset-x", "asset-y"],
+            fallback_ids=["asset-x", "asset-y", "asset-z"],
+        )
+        assert written == 2
+        assert assets["asset-x"].metadata["generation_use_count"] == 2
+        assert assets["asset-y"].metadata["generation_use_count"] == 1
+        assert "asset-z" not in assets, "未被 plan 选用的素材不应被计数"
+
+    def test_fallback_to_task_asset_ids_when_plan_empty(self):
+        """plan 无有效片段（异常数据）→ 兜底任务 asset_ids 每个计 1 次。"""
+        written, assets = self._run(clip_ids=[""], fallback_ids=["asset-a", "asset-b"])
+        assert written == 2
+        assert assets["asset-a"].metadata["generation_use_count"] == 1
+        assert assets["asset-b"].metadata["generation_use_count"] == 1
+
+    def test_no_clips_no_fallback_returns_zero(self):
+        written, assets = self._run(clip_ids=[], fallback_ids=[])
+        assert written == 0
+        assert assets == {}
+
+    def test_single_asset_failure_does_not_block_others(self):
+        """单素材 update 抛异常不影响其他素材回写。"""
+
+        def update_side(a):
+            if a.id == "bad":
+                raise RuntimeError("DB boom")
+            return a
+
+        written, assets = self._run(
+            clip_ids=["good1", "bad", "good2"],
+            fallback_ids=None,
+            repo_update_side=update_side,
+        )
+        assert written == 2
+        assert assets["good1"].metadata["generation_use_count"] == 1
+        assert assets["good2"].metadata["generation_use_count"] == 1
+
+    def test_missing_asset_skipped(self):
+        """repo.get 返回 None 的素材跳过，不报错。"""
+        written, assets = self._run(
+            clip_ids=["ghost", "real"],
+            fallback_ids=None,
+            repo_get_override={"ghost": None},
+        )
+        assert written == 1
+        assert assets["real"].metadata["generation_use_count"] == 1

tests/unit/test_smart_match.py

@@ -1,5 +1,6 @@
 """Tests for packages/domain/smart_match.py — 统一智能选素材算法。"""
 
+import random
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 from typing import Any
@@ -7,6 +8,7 @@ from typing import Any
 import pytest
 
 from packages.domain.smart_match import (
+    SCORE_RANDOM_NOISE_MAX,
     SmartMatchResult,
     _diversity_select,
     _duration_bucket,
@@ -46,6 +48,22 @@ class FakeAsset:
 NOW = datetime(2026, 8, 5, 12, 0, 0, tzinfo=timezone.utc)
 
 
+class _ZeroNoiseRandom(random.Random):
+    """零噪声随机源：uniform(0, NOISE_MAX) 恒返回 0，使「按评分降序」类断言确定可复现。
+
+    smart_select_assets 生产环境注入随机噪声（同分素材每次选出不同组合，#1743）；
+    验证纯评分排序的单测用本随机源消除排序随机性。
+    """
+
+    def uniform(self, a, b):
+        if a == 0.0 and b == SCORE_RANDOM_NOISE_MAX:
+            return 0.0
+        return super().uniform(a, b)
+
+
+_ZERO_NOISE = _ZeroNoiseRandom(0)
+
+
 # ── score_asset tests ────────────────────────────────────────────────────────
 
 
@@ -185,7 +203,7 @@ class TestSmartSelectAssets:
             FakeAsset(id="high", quality_score=95, duration=15),
             FakeAsset(id="mid", quality_score=60, duration=15),
         ]
-        results = smart_select_assets(assets)
+        results = smart_select_assets(assets, rng=_ZERO_NOISE)
         scores = [r.score for r in results]
         assert scores == sorted(scores, reverse=True)
         assert results[0].asset.id == "high"
@@ -234,7 +252,7 @@ class TestSmartSelectAssets:
             FakeAsset(id="img1", mime_type="image/jpeg", quality_score=90, duration=None),
             FakeAsset(id="img2", mime_type="image/png", quality_score=70, duration=None),
         ]
-        results = smart_select_assets(assets, kind="image")
+        results = smart_select_assets(assets, kind="image", rng=_ZERO_NOISE)
         assert len(results) == 2
         assert results[0].asset.id == "img1"
 

tests/unit/test_smart_match_fallback.py

@@ -155,7 +155,9 @@ class TestSmartMatchAvailabilityFallback:
         project = Project(id="proj-1", name="Test", owner_user_id="user-1")
         assets = [
             _video_asset("top-exhausted.mp4", quality=100, used_ranges=_exhausted_ranges(15)),
-            _video_asset("second-fresh.mp4", quality=40, used_ranges=None),
+            # second 质量分显著高于 third（质量项差 (90-30)*0.4=24 > 噪声上限 20），
+            # 排除耗尽素材后 second 稳定排首位回补（噪声不影响大分差排名）
+            _video_asset("second-fresh.mp4", quality=90, used_ranges=None),
             _video_asset("third-fresh.mp4", quality=30, used_ranges=None),
         ]
         repo = _StubAssetRepo(assets)

tests/unit/test_smart_match_integration.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import os
+import random
 import sys
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
@@ -41,6 +42,18 @@ class FakeAsset:
     file_type: str = "video"
 
 
+class _ZeroNoiseRandom(random.Random):
+    """零噪声随机源：uniform(0, NOISE_MAX) 恒返回 0，消除排序随机性；
+    其他随机调用（非噪声区间）保持正常随机行为。"""
+
+    def uniform(self, a, b):
+        from packages.domain.smart_match import SCORE_RANDOM_NOISE_MAX
+
+        if a == 0.0 and b == SCORE_RANDOM_NOISE_MAX:
+            return 0.0
+        return super().uniform(a, b)
+
+
 def _asset_with_use_count(asset_id: str, use_count: int) -> FakeAsset:
     """创建指定使用次数的素材，其他维度保持一致。"""
     return FakeAsset(
@@ -112,25 +125,25 @@ class TestSmartSelectAssetsOrdering:
     """验证 smart_select_assets 返回结果按评分降序。"""
 
     def test_less_used_assets_ranked_higher(self):
-        """使用次数少的素材在结果中排名更高。"""
+        """使用次数少的素材在结果中排名更高（注入零噪声 rng 验证纯评分排序）。"""
         assets = [
             _asset_with_use_count("heavily_used", 10),
             _asset_with_use_count("never_used", 0),
             _asset_with_use_count("lightly_used", 2),
         ]
-        results = smart_select_assets(assets)
+        results = smart_select_assets(assets, rng=_ZeroNoiseRandom())
         ids = [r.asset.id for r in results]
         # never_used 排第一，heavily_used 排最后
         assert ids[0] == "never_used"
         assert ids[-1] == "heavily_used"
 
     def test_same_quality_different_use_count(self):
-        """质量相同时，使用次数少的排名更高。"""
+        """质量相同时，使用次数少的排名更高（注入零噪声 rng）。"""
         assets = [
             _asset_with_use_count("used_5", 5),
             _asset_with_use_count("used_0", 0),
         ]
-        results = smart_select_assets(assets)
+        results = smart_select_assets(assets, rng=_ZeroNoiseRandom())
         assert results[0].asset.id == "used_0"
         assert results[1].asset.id == "used_5"
 
@@ -158,22 +171,47 @@ def _make_auth_user():
     return auth
 
 
-def _make_zero_noise_patcher(module):
-    """构造 patch(module.random.uniform)：噪声调用(上界=SCORE_RANDOM_NOISE_MAX)返回 0。
+class _ZeroNoiseCtx:
+    """同时 patch module.random.uniform（排序噪声归零）与 module.random.shuffle
+    （排序后的素材洗牌保持原序），使「按评分排序」类断言确定可复现。
 
-    其他 uniform 调用（如片段时长随机）委托给一个独立的 Random 实例，
-    避免递归回已 patch 的全局函数。
+    生产环境排序噪声与 shuffle 都是降重随机的一部分；测试需要验证纯评分
+    排序时用本上下文消除随机性。其他 uniform 调用（片段时长随机等）委托给
+    独立 Random 实例，避免递归回已 patch 的全局函数。
     """
-    import random as _stdlib_random
 
-    _fallback = _stdlib_random.Random()
+    def __init__(self, module):
+        import random as _stdlib_random
 
-    def _fake_uniform(a, b):
-        if b == SCORE_RANDOM_NOISE_MAX:
-            return 0.0
-        return _fallback.uniform(a, b)
+        self._module = module
+        self._fallback = _stdlib_random.Random()
+        self._patches = []
 
-    return patch.object(module.random, "uniform", _fake_uniform)
+    def __enter__(self):
+        import random as _stdlib_random
+
+        def _fake_uniform(a, b):
+            if b == SCORE_RANDOM_NOISE_MAX:
+                return 0.0
+            return self._fallback.uniform(a, b)
+
+        def _no_shuffle(seq):
+            return None  # 保持原序，不洗牌
+
+        self._patches = [
+            patch.object(self._module.random, "uniform", _fake_uniform).start(),
+            patch.object(self._module.random, "shuffle", _no_shuffle).start(),
+        ]
+        return self
+
+    def __exit__(self, *exc):
+        patch.stopall()
+        return False
+
+
+def _make_zero_noise_patcher(module):
+    """构造零噪声上下文（uniform 噪声归零 + shuffle 保持原序）。"""
+    return _ZeroNoiseCtx(module)
 
 
 def _patch_zero_noise_clips():

tests/unit/test_smart_match_noise_1743.py

@@ -0,0 +1,134 @@
+"""#1743 smart-match 排序随机噪声测试。
+
+smart_select_assets 排序注入 0~SCORE_RANDOM_NOISE_MAX 随机噪声后：
+- 同分/近分素材每次调用选出的组合与顺序不同（修复"每次只选同样几个素材"）
+- 分差 > 噪声上限的高质量素材保持稳定优先级
+- r.score 始终为无噪声原始分；噪声只影响排序
+- 同一次调用内排序与多样性分桶使用一致噪声（结果稳定可复现）
+"""
+
+from __future__ import annotations
+
+import random
+import sys
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+for sub in ("packages",):
+    p = str(REPO_ROOT / sub)
+    if p not in sys.path:
+        sys.path.insert(0, p)
+
+from packages.domain.smart_match import (  # noqa: E402
+    SCORE_RANDOM_NOISE_MAX,
+    score_asset,
+    smart_select_assets,
+)
+
+NOW = datetime(2026, 9, 6, 12, 0, 0, tzinfo=timezone.utc)
+
+
+@dataclass
+class FakeAsset:
+    id: str
+    duration: float = 15.0
+    quality_score: float | None = None  # None → 按 50 计，模拟 staging 真实情况
+    status: str = "ready"
+    metadata: dict = field(default_factory=dict)
+    created_at: datetime = NOW
+
+    @property
+    def file_type(self) -> str:
+        return "video"
+
+
+def _make_tied_assets(n: int) -> list[FakeAsset]:
+    """构造 n 个综合得分完全相同的素材（quality NULL 按 50 + 时长 15s 满分 + 同创建时间）。"""
+    return [FakeAsset(id=f"a{i}", duration=15.0, quality_score=None) for i in range(n)]
+
+
+class TestScoreNoiseInjection:
+    def test_tied_assets_same_raw_score(self):
+        """前置校验：同分素材原始得分确实一致。"""
+        assets = _make_tied_assets(6)
+        scores = {score_asset(a, now=NOW)[0] for a in assets}
+        assert len(scores) == 1, f"测试前提不成立：同分素材得分不一致 {scores}"
+
+    def test_tied_assets_order_varies_across_calls(self):
+        """同分素材：不同随机种子选出的顺序/组合不同（核心修复点）。"""
+        orderings = set()
+        for seed in range(8):
+            results = smart_select_assets(_make_tied_assets(8), rng=random.Random(seed))
+            orderings.add(tuple(r.asset.id for r in results))
+        # 8 个不同种子应产生多种不同排序（若零随机噪声则只有 1 种）
+        assert len(orderings) >= 4, f"同分素材排序几乎不变: {len(orderings)} 种"
+
+    def test_tied_assets_top_n_varies_with_limit(self):
+        """同分素材 + limit 截断：不同种子选出的 Top-N 组合不同。"""
+        top_sets = set()
+        for seed in range(10):
+            results = smart_select_assets(_make_tied_assets(10), limit=3, rng=random.Random(seed))
+            top_sets.add(frozenset(r.asset.id for r in results))
+        assert len(top_sets) >= 4, f"Top-N 组合几乎不变: {len(top_sets)} 种"
+
+    def test_large_score_gap_keeps_priority(self):
+        """分差 > 噪声上限（20）时：低质素材即使噪声拉满也排不到高质素材前面。"""
+        # 高质：quality=100 → quality_component=40；低质：quality=0 → 0，仅质量项就差 40 分
+        high = [FakeAsset(id="high", quality_score=100, duration=15.0)]
+        low = [FakeAsset(id=f"low{i}", quality_score=0, duration=15.0) for i in range(6)]
+        for seed in range(20):
+            results = smart_select_assets(high + low, limit=3, rng=random.Random(seed))
+            assert results[0].asset.id == "high", f"seed={seed} 低质素材靠噪声排到首位"
+
+    def test_score_field_is_raw_without_noise(self):
+        """r.score 始终是无噪声原始分（噪声只影响排序，不污染返回分值）。"""
+        assets = _make_tied_assets(5)
+        raw_scores = {score_asset(a, now=NOW)[0] for a in assets}
+        results = smart_select_assets(assets, rng=random.Random(42))
+        for r in results:
+            assert r.score in raw_scores
+
+    def test_rng_deterministic_same_seed(self):
+        """同一种子多次调用结果完全一致（可复现，测试可依赖）。"""
+        run = lambda: tuple(  # noqa: E731
+            r.asset.id for r in smart_select_assets(_make_tied_assets(8), rng=random.Random(123))
+        )
+        assert run() == run()
+
+    def test_diversity_bucket_respects_noise(self):
+        """多样性分桶路径（候选数 > limit）：同分素材跨种子入选组合不同。"""
+        # 构造短/中/长三档同分素材各 4 个，limit=6 触发分桶轮询
+        assets = []
+        for i in range(4):
+            assets.append(FakeAsset(id=f"short{i}", duration=6.0))
+        for i in range(4):
+            assets.append(FakeAsset(id=f"med{i}", duration=15.0))
+        for i in range(4):
+            assets.append(FakeAsset(id=f"long{i}", duration=45.0))
+        # 同档内时长接近 → 得分接近
+        combos = set()
+        for seed in range(10):
+            results = smart_select_assets(assets, limit=6, rng=random.Random(seed))
+            combos.add(frozenset(r.asset.id for r in results))
+        assert len(combos) >= 3, f"分桶选取组合几乎不变: {len(combos)} 种"
+
+    def test_noise_constant_matches_from_assets(self):
+        """噪声上限与 from-assets 片段分配的 SCORE_RANDOM_NOISE_MAX 同源（20 分）。"""
+        assert SCORE_RANDOM_NOISE_MAX == 20.0
+
+    def test_returns_all_when_no_limit(self):
+        """无 limit 时返回全部候选（噪声只改顺序，不丢素材）。"""
+        assets = _make_tied_assets(7)
+        results = smart_select_assets(assets, rng=random.Random(1))
+        assert len(results) == 7
+        assert {r.asset.id for r in results} == {a.id for a in assets}
+
+    def test_non_ready_assets_excluded_before_noise(self):
+        """非 ready 素材不参与排序（噪声不影响状态过滤）。"""
+        assets = _make_tied_assets(4)
+        assets[0].status = "processing"
+        results = smart_select_assets(assets, rng=random.Random(1))
+        assert all(r.asset.status == "ready" for r in results)
+        assert len(results) == 3