apps/api/app/api/routes/upload.py

@@ -290,6 +290,43 @@ def _submit_ingest_job(
     return job
 
 
+def _find_active_ingest_job(ingest_job_repository: Any, asset_id: str) -> Any | None:
+    """查询 asset 上是否存在"仍在跑或已成功"的 ingest job（FAILED 视为不存在，需重提）。"""
+    if not asset_id:
+        return None
+    find = getattr(ingest_job_repository, "find_by_asset_id", None)
+    if not callable(find):
+        # 旧仓储未实现 find_by_asset_id，无法判断 → 保守返回 None（走正常流程，
+        # _submit_ingest_job 自身有数据库唯一约束/幂等兜底，不会重复建 job）
+        return None
+    try:
+        return find(asset_id)
+    except Exception:  # noqa: BLE001
+        logger.warning("[upload] find_by_asset_id 查询失败，按无 job 处理: asset=%s", asset_id, exc_info=True)
+        return None
+
+
+def _is_true_duplicate(existing_asset: Asset, ingest_job_repository: Any) -> tuple[bool, Any | None]:
+    """判断 `existing_asset` 是真重复（应短路返 duplicated）还是占位（应补提 ingest）。
+
+    返回 (is_duplicate, existing_job):
+    - READY 素材：真重复，job 可能为 None（已就绪不需要 job_id）
+    - PROCESSING/UPLOADING 且已有在跑/已完成 ingest job：幂等重试，真重复，job 返回给前端轮询
+    - PROCESSING/UPLOADING 且无 job：prepare 建的占位 / 之前 ingest 创建失败 → 非重复，需补提 ingest
+    - ERROR/DELETED：非重复（允许重新上传覆盖）
+    """
+    status = getattr(existing_asset, "status", None)
+    if status == AssetStatus.READY:
+        return True, None
+    if status in (AssetStatus.PROCESSING, AssetStatus.UPLOADING):
+        job = _find_active_ingest_job(ingest_job_repository, existing_asset.id)
+        if job is not None:
+            return True, job
+        return False, None
+    # ERROR / DELETED / 其它：走正常流程重新 ingest
+    return False, None
+
+
 @router.post("/direct/prepare", response_model=DirectUploadPrepareResponse)
 async def prepare_direct_upload(
     request: DirectUploadPrepareRequest,
@@ -444,12 +481,25 @@ async def complete_direct_upload(
         file_size=request.file_size,
     )
     if existing is not None:
-        return DirectUploadCompleteResponse(
-            storage_key=existing.storage_key,
-            ingest_job_id="",
-            duplicated=True,
-            asset_id=existing.id,
-            url=storage_service.get_url(existing.storage_key),
+        is_dup, existing_job = _is_true_duplicate(existing, ingest_job_repository)
+        if is_dup:
+            logger.info(
+                "[upload] complete 幂等命中真重复: asset=%s status=%s job=%s",
+                existing.id,
+                getattr(existing, "status", None),
+                getattr(existing_job, "id", None),
+            )
+            return DirectUploadCompleteResponse(
+                storage_key=existing.storage_key,
+                ingest_job_id=getattr(existing_job, "id", "") or "",
+                duplicated=True,
+                asset_id=existing.id,
+                url=storage_service.get_url(existing.storage_key),
+            )
+        logger.info(
+            "[upload] complete 命中占位 asset（status=%s 无 ingest job），继续补提 ingest: asset=%s",
+            getattr(existing, "status", None),
+            existing.id,
         )
 
     try:
@@ -479,14 +529,24 @@ async def complete_direct_upload(
     )
     # Issue #1776: 计数由 asset_repository.create() 自动维护
 
-    job = _submit_ingest_job(
-        project_id=request.project_id,
-        library_id=request.library_id,
-        storage_key=normalized_key,
-        ingest_job_repository=ingest_job_repository,
-        file_hash=request.file_hash,
-        asset_id=pending_asset.id,
-    )
+    # 幂等保护：补提占位场景下可能已有 job（极端竞态），先查一次
+    existing_job = _find_active_ingest_job(ingest_job_repository, pending_asset.id)
+    if existing_job is not None:
+        logger.info(
+            "[upload] complete 补提时发现 job 已存在（竞态/并发重试），复用: asset=%s job=%s",
+            pending_asset.id,
+            existing_job.id,
+        )
+        job = existing_job
+    else:
+        job = _submit_ingest_job(
+            project_id=request.project_id,
+            library_id=request.library_id,
+            storage_key=normalized_key,
+            ingest_job_repository=ingest_job_repository,
+            file_hash=request.file_hash,
+            asset_id=pending_asset.id,
+        )
     return DirectUploadCompleteResponse(
         storage_key=normalized_key,
         ingest_job_id=job.id,
@@ -533,12 +593,27 @@ async def upload_asset(
         file_size=0,
     )
     if existing is not None:
-        return UploadAssetResponse(
-            storage_key=existing.storage_key,
-            ingest_job_id="",
-            url="",
-            duplicated=True,
-            asset_id=existing.id,
+        is_dup, existing_job = _is_true_duplicate(existing, ingest_job_repository)
+        if is_dup:
+            logger.info(
+                "[upload] multipart 幂等命中真重复: asset=%s status=%s job=%s",
+                existing.id,
+                getattr(existing, "status", None),
+                getattr(existing_job, "id", None),
+            )
+            return UploadAssetResponse(
+                storage_key=existing.storage_key,
+                ingest_job_id=getattr(existing_job, "id", "") or "",
+                url=storage_service.get_url(existing.storage_key)
+                if getattr(existing, "status", None) == AssetStatus.READY
+                else "",
+                duplicated=True,
+                asset_id=existing.id,
+            )
+        logger.info(
+            "[upload] multipart 命中占位 asset（status=%s 无 ingest job），继续补提 ingest: asset=%s",
+            getattr(existing, "status", None),
+            existing.id,
         )
 
     file_id = uuid4().hex[:8]
@@ -574,14 +649,23 @@ async def upload_asset(
     )
     # Issue #1776: 计数由 asset_repository.create() 自动维护
 
-    job = _submit_ingest_job(
-        project_id=project_id,
-        library_id=library_id,
-        storage_key=storage_key,
-        ingest_job_repository=ingest_job_repository,
-        file_hash=file_hash,
-        asset_id=pending_asset.id,
-    )
+    existing_job = _find_active_ingest_job(ingest_job_repository, pending_asset.id)
+    if existing_job is not None:
+        logger.info(
+            "[upload] multipart 补提时发现 job 已存在（竞态/并发重试），复用: asset=%s job=%s",
+            pending_asset.id,
+            existing_job.id,
+        )
+        job = existing_job
+    else:
+        job = _submit_ingest_job(
+            project_id=project_id,
+            library_id=library_id,
+            storage_key=storage_key,
+            ingest_job_repository=ingest_job_repository,
+            file_hash=file_hash,
+            asset_id=pending_asset.id,
+        )
 
     return UploadAssetResponse(
         storage_key=storage_key,

apps/worker/video_processing/thumbnail_generator.py

@@ -606,6 +606,20 @@ def extract_and_upload_cover_frames(
                 if len(candidates) >= num_frames:
                     logger.info("[thumbnail] MediaKit 抽帧完成: %d 帧", len(candidates))
 
+                # MediaKit 路径帧在 NamedTemporaryFile 中持久存在（finally 清理），在进入本地 ffmpeg 前评分
+                if len(candidates) > 1:
+                    try:
+                        from packages.shared.cover_frame_scorer import score_frames
+
+                        candidates = score_frames(candidates)
+                        logger.info(
+                            "[thumbnail] MediaKit 封面帧评分完成: count=%d best_score=%.1f",
+                            len(candidates),
+                            candidates[0].get("score", 0.0) if candidates else 0.0,
+                        )
+                    except Exception:
+                        logger.warning("[thumbnail] MediaKit 封面帧质量评分失败，保持原始顺序", exc_info=True)
+
         # ── 默认路径：本地 ffmpeg 单次 select 抽帧 + 并发上传 ──────────────
         if len(candidates) < num_frames:
             if candidates:
@@ -651,54 +665,89 @@ def extract_and_upload_cover_frames(
                         except Exception as e:
                             logger.warning("[thumbnail] 标题叠加失败 %s: %s", fp, e)
 
-                # 5) 并发上传 OSS（线程池并发）
+                # 5) 质量评分（必须在 TemporaryDirectory 内，帧文件还在磁盘上）
+                t_score = time.monotonic()
+                local_candidates: list[dict] = [{"position": st, "image_path": fp} for (st, fp) in frame_results]
+                scored: list[dict] = local_candidates
+                if len(local_candidates) > 1:
+                    try:
+                        from packages.shared.cover_frame_scorer import score_frames
+
+                        scored = score_frames(local_candidates)
+                        logger.info(
+                            "[thumbnail] 封面评分耗时: %.2fs (best_score=%.1f, count=%d)",
+                            time.monotonic() - t_score,
+                            scored[0].get("score", 0.0) if scored else 0.0,
+                            len(scored),
+                        )
+                    except Exception:
+                        logger.warning(
+                            "[thumbnail] 封面帧质量评分失败，保持 seek 点原始顺序",
+                            exc_info=True,
+                        )
+                        scored = local_candidates
+
+                # 6) 按评分顺序并发上传 OSS（best 帧先上传；best 已是 scored[0]）
                 t2 = time.monotonic()
 
-                def _upload_one(idx: int, st: float, fp: str) -> dict | None:
+                def _upload_one(rank: int, st: float, fp: str, score: float) -> dict | None:
                     try:
-                        storage_key = f"covers/{plan_id}/{task_id}/frame_{idx}.jpg"
+                        storage_key = f"covers/{plan_id}/{task_id}/frame_{rank}.jpg"
                         url = upload_to_oss(fp, storage_key)
                         if url:
-                            return {"url": url, "position": st, "image_path": fp}
-                        logger.warning("[thumbnail] 上传失败 idx=%d", idx)
+                            return {
+                                "url": url,
+                                "position": st,
+                                "image_path": fp,
+                                "score": score,
+                                "is_best": rank == 0,
+                            }
+                        logger.warning("[thumbnail] 上传失败 rank=%d t=%.2f", rank, st)
                     except Exception as e:
-                        logger.warning("[thumbnail] 上传异常 idx=%d t=%.2f: %s", idx, st, e)
+                        logger.warning("[thumbnail] 上传异常 rank=%d t=%.2f: %s", rank, st, e)
                     return None
 
-                upload_results: list[dict | None] = [None] * len(frame_results)
-                max_workers = min(8, max(2, len(frame_results)))
+                upload_results: list[dict | None] = [None] * len(scored)
+                max_workers = min(8, max(2, len(scored)))
                 with ThreadPoolExecutor(max_workers=max_workers) as pool:
-                    future_map = {pool.submit(_upload_one, i, st, fp): i for i, (st, fp) in enumerate(frame_results)}
+                    future_map = {
+                        pool.submit(
+                            _upload_one,
+                            i,
+                            float(c.get("position", 0.0)),
+                            str(c["image_path"]),
+                            float(c.get("score", 0.0)),
+                        ): i
+                        for i, c in enumerate(scored)
+                    }
                     for fut in as_completed(future_map):
                         i = future_map[fut]
                         try:
                             upload_results[i] = fut.result()
                         except Exception as e:
-                            logger.warning("[thumbnail] 上传 feature 异常 idx=%d: %s", i, e)
+                            logger.warning("[thumbnail] 上传 future 异常 rank=%d: %s", i, e)
                 logger.info("[thumbnail] 并发上传耗时: %.2fs", time.monotonic() - t2)
 
                 for r in upload_results:
                     if r is not None:
-                        _temp_paths.append(r["image_path"])
+                        # 本地帧在 TemporaryDirectory 内，with 退出自动删除，无需进 _temp_paths
                         candidates.append(r)
 
-        # ── 阶段 2：质量评分 ────────────────────────────────────────────
+        # 如果本地 ffmpeg 路径产生了候选（已评分）但未经过 MediaKit 路径，candidates 已按评分顺序排好。
+        # 混合场景下（MediaKit + 本地 ffmpeg 都产出），统一按 score 降序排列；缺失 score 的（理论上不应出现）排末尾。
         if len(candidates) > 1:
-            try:
-                from packages.shared.cover_frame_scorer import score_frames
-
-                candidates = score_frames(candidates)
+            candidates.sort(key=lambda c: c.get("score", -1.0), reverse=True)
+            if candidates:
+                candidates[0]["is_best"] = True
                 elapsed = time.monotonic() - t0
                 logger.info(
-                    "[thumbnail] 封面帧评分完成: plan_id=%s count=%d best_score=%.1f elapsed=%.2fs path=%s",
+                    "[thumbnail] 封面完成: plan_id=%s count=%d best=t%.2fs score=%.1f elapsed=%.2fs",
                     plan_id,
                     len(candidates),
-                    candidates[0].get("score", 0.0) if candidates else 0.0,
+                    candidates[0].get("position", 0.0),
+                    candidates[0].get("score", 0.0),
                     elapsed,
-                    "mediakit" if use_mediakit else "local",
                 )
-            except Exception:
-                logger.warning("[thumbnail] 封面帧质量评分失败，保持原始顺序", exc_info=True)
 
         for c in candidates:
             c.pop("image_path", None)

packages/adapters/in_memory/ingest_job_repository.py

@@ -14,6 +14,24 @@ class InMemoryIngestJobRepository:
     def get(self, job_id: str) -> IngestJob | None:
         return self._items.get(job_id)
 
+    def find_by_asset_id(self, asset_id: str) -> IngestJob | None:
+        if not asset_id:
+            return None
+        from packages.domain.classification import IngestJobStatus
+
+        running: IngestJob | None = None
+        completed: IngestJob | None = None
+        for job in self._items.values():
+            if getattr(job, "asset_id", "") != asset_id:
+                continue
+            if job.status in (IngestJobStatus.PENDING, IngestJobStatus.PROCESSING):
+                if running is None or job.created_at > running.created_at:
+                    running = job
+            elif job.status == IngestJobStatus.COMPLETED:
+                if completed is None or job.created_at > completed.created_at:
+                    completed = job
+        return running or completed
+
     def update(self, job: IngestJob) -> IngestJob:
         self._items[job.id] = job
         return job

packages/adapters/sqlalchemy_impl/ingest_job_repository.py

@@ -46,6 +46,43 @@ class SQLAlchemyIngestJobRepository:
             updated_at=model.updated_at,
         )
 
+    def find_by_asset_id(self, asset_id: str) -> IngestJob | None:
+        """返回 asset 最近一条未失败的 ingest job（PENDING/PROCESSING/COMPLETED 均算存在，用于幂等判断）。"""
+        if not asset_id:
+            return None
+        # 优先返回仍在跑的 (PENDING/PROCESSING)，否则返回最新一条 COMPLETED
+        model = (
+            self.session.query(IngestJobModel)
+            .filter(IngestJobModel.asset_id == asset_id)
+            .filter(IngestJobModel.status.in_([IngestJobStatus.PENDING.value, IngestJobStatus.PROCESSING.value]))
+            .order_by(IngestJobModel.created_at.desc())
+            .first()
+        )
+        if model is None:
+            model = (
+                self.session.query(IngestJobModel)
+                .filter(IngestJobModel.asset_id == asset_id)
+                .filter(IngestJobModel.status == IngestJobStatus.COMPLETED.value)
+                .order_by(IngestJobModel.created_at.desc())
+                .first()
+            )
+        if model is None:
+            return None
+        return IngestJob(
+            id=model.id,
+            project_id=model.project_id,
+            library_id=model.library_id,
+            storage_key=model.storage_key,
+            status=IngestJobStatus(model.status),
+            error_message=model.error_message,
+            result_asset_id=model.result_asset_id,
+            file_hash=model.file_hash or "",
+            asset_id=getattr(model, "asset_id", "") or "",
+            celery_task_id=getattr(model, "celery_task_id", "") or "",
+            created_at=model.created_at,
+            updated_at=model.updated_at,
+        )
+
     def list_by_project(self, project_id: str) -> list[IngestJob]:
         models = self.session.query(IngestJobModel).filter(IngestJobModel.project_id == project_id).all()
         return [self.get(model.id) for model in models if self.get(model.id) is not None]

packages/ports/ingest_job_repository.py

@@ -12,5 +12,8 @@ class IngestJobRepository(Protocol):
     def get(self, job_id: str) -> IngestJob | None:
         """Retrieve an ingest job by ID."""
 
+    def find_by_asset_id(self, asset_id: str) -> IngestJob | None:
+        """Retrieve the most recent ingest job for a given asset (幂等判断)."""
+
     def update(self, job: IngestJob) -> IngestJob:
         """Update an ingest job and return it."""

tests/unit/test_thumbnail_generator.py

@@ -282,3 +282,130 @@ class TestSeekPointBlackAvoidance:
         assert out[2] == 8.0
         # 中点必须不在黑屏内
         assert not (4.5 <= out[1] <= 5.5)
+
+
+class TestScorerRunsInsideTempDir:
+    """P1 修复：scorer 必须在 TemporaryDirectory 块内调用（帧文件还在时）。"""
+
+    def _setup_mocks(self, monkeypatch, tmp_path, *, scorer_should_read=True):
+        from video_processing.thumbnail_generator import extract_and_upload_cover_frames
+
+        class FakeClient:
+            is_available = False
+
+        monkeypatch.setattr(
+            "packages.shared.mediakit_client.get_mediakit_client",
+            lambda: FakeClient(),
+        )
+
+        self._frames_on_disk_when_called = []
+
+        def _fake_single_pass(video_path, seek_points, out_dir, prefix="frame", **kw):
+            results = []
+            for i, st in enumerate(seek_points):
+                fp = Path(out_dir) / f"{prefix}_{i + 1:02d}.jpg"
+                fp.write_bytes(b"\xff\xd8\xff\xe0" + b"X" * 200)
+                results.append((st, str(fp)))
+            return results
+
+        monkeypatch.setattr(
+            "video_processing.thumbnail_generator._extract_frames_single_pass",
+            _fake_single_pass,
+        )
+        monkeypatch.setattr(
+            "video_processing.thumbnail_generator._detect_black_intervals",
+            lambda *a, **kw: [],
+        )
+        monkeypatch.setattr(
+            "video_processing.ffmpeg_utils.probe_duration",
+            lambda path: 60.0,
+        )
+        monkeypatch.setattr(
+            "video_processing.oss_helpers.upload_to_oss",
+            lambda path, key: f"https://oss.example.com/{key}",
+        )
+        # 标题叠加 no-op
+        monkeypatch.setattr(
+            "video_processing.thumbnail_generator.apply_title_overlay",
+            lambda *a, **kw: None,
+        )
+
+        # 记录 scorer 被调用时各 image_path 是否存在
+        def _fake_scorer(candidates):
+            for c in candidates:
+                self._frames_on_disk_when_called.append(Path(c["image_path"]).exists())
+            # 给个假评分：倒序排，验证顺序被应用
+            scored = list(candidates)
+            for i, c in enumerate(scored):
+                c["score"] = float(len(scored) - i)
+            scored.sort(key=lambda c: c["score"], reverse=True)
+            return scored
+
+        monkeypatch.setattr(
+            "packages.shared.cover_frame_scorer.score_frames",
+            _fake_scorer,
+        )
+        return extract_and_upload_cover_frames
+
+    def test_scorer_reads_files_while_they_exist(self, tmp_path, monkeypatch):
+        """核心 P1：评分时帧文件必须还在磁盘上（在 TemporaryDirectory 内调用）。"""
+        extract = self._setup_mocks(monkeypatch, tmp_path)
+        video_file = tmp_path / "t.mp4"
+        video_file.write_bytes(b"fake")
+        result = extract(str(video_file), "plan1", num_frames=3)
+        # scorer 看到的 3 个文件都必须存在
+        assert len(self._frames_on_disk_when_called) == 3
+        assert all(self._frames_on_disk_when_called), f"scorer 调用时有文件已被删除: {self._frames_on_disk_when_called}"
+        # 结果按评分降序排列（is_best 在第一个）
+        assert len(result) == 3
+        assert result[0].get("is_best") is True
+        # 结果中不应该再暴露 image_path
+        assert all("image_path" not in c for c in result)
+
+    def test_scorer_failure_falls_back_gracefully(self, tmp_path, monkeypatch):
+        """评分抛异常时不应中断上传，仍返回所有候选帧。"""
+        from video_processing.thumbnail_generator import extract_and_upload_cover_frames
+
+        class FakeClient:
+            is_available = False
+
+        monkeypatch.setattr("packages.shared.mediakit_client.get_mediakit_client", lambda: FakeClient())
+
+        def _fake_single_pass(video_path, seek_points, out_dir, prefix="frame", **kw):
+            results = []
+            for i, st in enumerate(seek_points):
+                fp = Path(out_dir) / f"{prefix}_{i + 1:02d}.jpg"
+                fp.write_bytes(b"\xff\xd8\xff\xe0" + b"X" * 100)
+                results.append((st, str(fp)))
+            return results
+
+        monkeypatch.setattr("video_processing.thumbnail_generator._extract_frames_single_pass", _fake_single_pass)
+        monkeypatch.setattr("video_processing.thumbnail_generator._detect_black_intervals", lambda *a, **kw: [])
+        monkeypatch.setattr("video_processing.ffmpeg_utils.probe_duration", lambda p: 60.0)
+        monkeypatch.setattr(
+            "video_processing.oss_helpers.upload_to_oss",
+            lambda path, key: f"https://oss/{key}",
+        )
+        monkeypatch.setattr("video_processing.thumbnail_generator.apply_title_overlay", lambda *a, **kw: None)
+
+        def _boom(candidates):
+            raise RuntimeError("cv2 crashed")
+
+        monkeypatch.setattr("packages.shared.cover_frame_scorer.score_frames", _boom)
+
+        video_file = tmp_path / "t.mp4"
+        video_file.write_bytes(b"fake")
+        # 不应抛出
+        result = extract_and_upload_cover_frames(str(video_file), "plan1", num_frames=3)
+        assert len(result) == 3
+        assert all("url" in c for c in result)
+
+    def test_best_frame_is_first_after_scoring(self, tmp_path, monkeypatch):
+        """评分后 best 帧（score 最高）在 candidates[0]，is_best=True。"""
+        extract = self._setup_mocks(monkeypatch, tmp_path)
+        video_file = tmp_path / "t.mp4"
+        video_file.write_bytes(b"fake")
+        result = extract(str(video_file), "plan1", num_frames=5)
+        assert result[0]["is_best"] is True
+        scores = [c.get("score", 0.0) for c in result]
+        assert scores == sorted(scores, reverse=True)

tests/unit/test_upload_complete_idempotency_1714.py

@@ -128,6 +128,25 @@ class StubIngestJobRepository:
     def get(self, job_id: str) -> IngestJob | None:
         return self._jobs.get(job_id)
 
+    def find_by_asset_id(self, asset_id: str) -> IngestJob | None:
+        """返回 asset 最近一条在跑/已完成 job（FAILED 不返回，允许重提）。"""
+        from packages.domain.classification import IngestJobStatus as S
+
+        if not asset_id:
+            return None
+        running = None
+        completed = None
+        for job in self._jobs.values():
+            if getattr(job, "asset_id", "") != asset_id:
+                continue
+            if job.status in (S.PENDING, S.PROCESSING):
+                if running is None or job.created_at > running.created_at:
+                    running = job
+            elif job.status == S.COMPLETED:
+                if completed is None or job.created_at > completed.created_at:
+                    completed = job
+        return running or completed
+
     def update(self, job: IngestJob) -> IngestJob:
         self._jobs[job.id] = job
         return job
@@ -420,3 +439,172 @@ class TestMultipartUploadIdempotency:
         assert ingest_repo.created_count == 1
         # OSS 上传只发生一次（第二次在幂等检查处直接返回）
         assert storage.upload_file.call_count == 1
+
+
+# ── P1 修复（#2092）：占位 asset 被误判为 duplicate → 素材永久 processing ──
+
+
+class TestPlaceholderAssetNotTreatedAsDuplicate:
+    """prepare 建了 PROCESSING 占位但还没 ingest，complete 必须补提 ingest 而不是短路返 duplicated。"""
+
+    def test_complete_hits_prepare_placeholder_without_job_submits_ingest(self):
+        """场景：/direct/prepare 建了 PROCESSING 占位（同 client_upload_id），complete 命中后应补提 ingest。"""
+        from packages.domain.classification import IngestJobStatus
+
+        # 预建占位 asset（prepare 建的，PROCESSING，无 ingest job）
+        placeholder = Asset(
+            id="asset-placeholder",
+            project_id="proj-1",
+            library_id="lib-1",
+            name="IMG_9999.MOV",
+            storage_key="uploads/prepare/IMG_9999.MOV",
+            mime_type="video/quicktime",
+            status=AssetStatus.PROCESSING,
+            client_upload_id="tok-prep",
+        )
+        # 注意：占位的 storage_key 是 prepare 生成的 key，complete 传入的是用户实际上传的 key
+        client, asset_repo, ingest_repo, storage = _client(
+            asset_repo=StubAssetRepository([placeholder]),
+        )
+        storage._normalize_storage_key = lambda k: k  # complete 用自己的 key
+        storage.get_url = lambda k: f"https://oss/{k}"
+
+        body = {
+            "project_id": "proj-1",
+            "library_id": "lib-1",
+            "storage_key": "uploads/actual/IMG_9999.MOV",  # complete 用真实上传 key
+            "client_upload_id": "tok-prep",  # 命中占位
+            "file_size": 1024,
+        }
+        r = client.post("/api/v1/direct/complete", json=body)
+        assert r.status_code == 200, r.text
+        b = r.json()
+        # 关键：不是 duplicate；返回 job_id；占位被复用（不新建 asset）
+        assert b["duplicated"] is False, f"占位被误判为 duplicate: {b}"
+        assert b["ingest_job_id"], "应补提 ingest job"
+        assert b["asset_id"] == "asset-placeholder"
+        # 不新建 asset（占位复用）
+        assert len(asset_repo.created) == 0
+        # job 被提交
+        assert ingest_repo.created_count == 1
+
+    def test_ready_asset_treated_as_true_duplicate(self):
+        """READY 素材命中 → 真重复，返 duplicated 且不提交新 job。"""
+        ready = Asset(
+            id="asset-ready",
+            project_id="proj-1",
+            library_id="lib-1",
+            name="done.MOV",
+            storage_key="uploads/done/done.MOV",
+            mime_type="video/quicktime",
+            status=AssetStatus.READY,
+            client_upload_id="tok-done",
+        )
+        client, asset_repo, ingest_repo, _storage = _client(asset_repo=StubAssetRepository([ready]))
+        body = {
+            "project_id": "proj-1",
+            "library_id": "lib-1",
+            "storage_key": "uploads/done/done.MOV",
+            "client_upload_id": "tok-done",
+        }
+        r = client.post("/api/v1/direct/complete", json=body)
+        assert r.status_code == 200
+        b = r.json()
+        assert b["duplicated"] is True
+        assert b["asset_id"] == "asset-ready"
+        assert ingest_repo.created_count == 0
+
+    def test_processing_asset_with_existing_job_is_idempotent_duplicate(self):
+        """PROCESSING 但已有在跑 job → 幂等重试，返 duplicated + 已有 job_id，不重复提交。"""
+        processing = Asset(
+            id="asset-running",
+            project_id="proj-1",
+            library_id="lib-1",
+            name="running.MOV",
+            storage_key="uploads/run/running.MOV",
+            mime_type="video/quicktime",
+            status=AssetStatus.PROCESSING,
+            client_upload_id="tok-run",
+        )
+        client, asset_repo, ingest_repo, _ = _client(asset_repo=StubAssetRepository([processing]))
+        # 预置一个在跑 job
+        existing_job = IngestJob(
+            id="job-existing",
+            project_id="proj-1",
+            library_id="lib-1",
+            storage_key="uploads/run/running.MOV",
+            asset_id="asset-running",
+        )
+        ingest_repo.create(existing_job)
+        before = ingest_repo.created_count
+
+        body = {
+            "project_id": "proj-1",
+            "library_id": "lib-1",
+            "storage_key": "uploads/run/running.MOV",
+            "client_upload_id": "tok-run",
+        }
+        r = client.post("/api/v1/direct/complete", json=body)
+        assert r.status_code == 200
+        b = r.json()
+        assert b["duplicated"] is True
+        assert b["ingest_job_id"] == "job-existing"
+        # 没有新建 job
+        assert ingest_repo.created_count == before
+        assert len(asset_repo.created) == 0
+
+    def test_error_asset_allows_reingest(self):
+        """ERROR 状态素材命中 → 不视为 duplicate，重新走 ingest。"""
+        errored = Asset(
+            id="asset-err",
+            project_id="proj-1",
+            library_id="lib-1",
+            name="err.MOV",
+            storage_key="uploads/err/err.MOV",
+            mime_type="video/quicktime",
+            status=AssetStatus.ERROR,
+            client_upload_id="tok-err",
+        )
+        client, asset_repo, ingest_repo, _ = _client(asset_repo=StubAssetRepository([errored]))
+        body = {
+            "project_id": "proj-1",
+            "library_id": "lib-1",
+            "storage_key": "uploads/err/err.MOV",
+            "client_upload_id": "tok-err",
+        }
+        r = client.post("/api/v1/direct/complete", json=body)
+        assert r.status_code == 200
+        b = r.json()
+        assert b["duplicated"] is False
+        assert b["ingest_job_id"]
+        assert ingest_repo.created_count == 1
+
+    def test_multipart_placeholder_without_job_submits_ingest(self):
+        """multipart 上传命中 PROCESSING 占位且无 job → 补提 ingest（不短路返 duplicated）。"""
+        placeholder = Asset(
+            id="asset-mp-placeholder",
+            project_id="proj-1",
+            library_id="lib-1",
+            name="mp.MOV",
+            storage_key="uploads/mp/mp.MOV",
+            mime_type="video/quicktime",
+            status=AssetStatus.PROCESSING,
+            client_upload_id="tok-mp",
+        )
+        client, asset_repo, ingest_repo, storage = _client(asset_repo=StubAssetRepository([placeholder]))
+        storage.upload_file = MagicMock(return_value="https://oss/mp.MOV")
+        storage.get_url = MagicMock(return_value="https://oss/mp.MOV")
+        r = client.post(
+            "/api/v1",
+            data={
+                "project_id": "proj-1",
+                "library_id": "lib-1",
+                "client_upload_id": "tok-mp",
+            },
+            files={"file": ("mp.MOV", b"data", "video/quicktime")},
+        )
+        assert r.status_code == 200, r.text
+        b = r.json()
+        assert b["duplicated"] is False, f"multipart 占位被误判为 duplicate: {b}"
+        assert b["ingest_job_id"]
+        assert ingest_repo.created_count == 1