alembic/versions/066_asset_client_upload_id_ingest_job_asset_id.py

@@ -0,0 +1,34 @@
+"""add client_upload_id to assets and asset_id to ingest_jobs
+
+Issue #1714：上传 complete 幂等 + worker 转码回写关联。
+- assets.client_upload_id：客户端幂等 token（complete 去重）
+- ingest_jobs.asset_id：complete 阶段创建的占位 asset id（worker 回写关联，
+  防止 HEVC 转码改写 storage_key 后找不到占位而兜底新建 READY 记录）
+
+Revision ID: 066_upload_idempotency
+Revises: 065_dup_record_sim_match
+Create Date: 2026-09-05
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "066_upload_idempotency"
+down_revision = "065_dup_record_sim_match"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("assets", sa.Column("client_upload_id", sa.String(64), nullable=True))
+    op.create_index("ix_assets_client_upload_id", "assets", ["client_upload_id"])
+    op.add_column("ingest_jobs", sa.Column("asset_id", sa.String(36), nullable=False, server_default=""))
+    op.create_index("ix_ingest_jobs_asset_id", "ingest_jobs", ["asset_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_ingest_jobs_asset_id", table_name="ingest_jobs")
+    op.drop_column("ingest_jobs", "asset_id")
+    op.drop_index("ix_assets_client_upload_id", table_name="assets")
+    op.drop_column("assets", "client_upload_id")

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

@@ -85,12 +85,22 @@ def _infer_mime_type_from_storage_key(storage_key: str) -> str:
     """从 storage_key 推断 MIME 类型（与 worker 端保持一致）。"""
     lower_filename = storage_key.rsplit("/", 1)[-1].lower()
     _MIME_MAP = {
-        ".mov": "video/quicktime", ".mp4": "video/mp4", ".avi": "video/x-msvideo",
-        ".mkv": "video/x-matroska", ".webm": "video/webm",
-        ".png": "image/png", ".gif": "image/gif", ".bmp": "image/bmp",
-        ".svg": "image/svg+xml", ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
-        ".mp3": "audio/mpeg", ".wav": "audio/wav", ".ogg": "audio/ogg",
-        ".flac": "audio/flac", ".m4a": "audio/x-m4a",
+        ".mov": "video/quicktime",
+        ".mp4": "video/mp4",
+        ".avi": "video/x-msvideo",
+        ".mkv": "video/x-matroska",
+        ".webm": "video/webm",
+        ".png": "image/png",
+        ".gif": "image/gif",
+        ".bmp": "image/bmp",
+        ".svg": "image/svg+xml",
+        ".jpg": "image/jpeg",
+        ".jpeg": "image/jpeg",
+        ".mp3": "audio/mpeg",
+        ".wav": "audio/wav",
+        ".ogg": "audio/ogg",
+        ".flac": "audio/flac",
+        ".m4a": "audio/x-m4a",
     }
     for ext, mime in _MIME_MAP.items():
         if lower_filename.endswith(ext):
@@ -98,8 +108,85 @@ def _infer_mime_type_from_storage_key(storage_key: str) -> str:
     return "video/mp4"  # default
 
 
+# 兜底去重：无 file_hash / client_upload_id 时，同库同名近期活动记录视为重复
+FALLBACK_DEDUP_WINDOW_MINUTES = 30
+ACTIVE_ASSET_STATUSES = (AssetStatus.UPLOADING, AssetStatus.PROCESSING)
+
+
+def _find_duplicate_asset(
+    asset_repository: Any,
+    *,
+    library_id: str,
+    file_hash: str,
+    client_upload_id: str,
+    filename: str,
+    file_size: int = 0,
+) -> Any:
+    """complete/上传幂等去重，按优先级查找已存在的素材。
+
+    1. client_upload_id（客户端幂等 token，同一次上传的重试保持一致）
+    2. file_hash（内容哈希，不同上传只要内容相同即去重）
+    3. 兜底：同库 + 同文件名（+同大小）且 30 分钟内仍处 uploading/processing
+       的记录——旧客户端不传 hash/token 时，防止 complete 超时重试反复建占位。
+
+    全部为鸭子类型调用：旧仓储无对应方法时静默跳过，不破坏既有实现。
+    """
+    if client_upload_id:
+        find = getattr(asset_repository, "find_by_library_and_client_upload_id", None)
+        if callable(find):
+            existing = find(library_id=library_id, client_upload_id=client_upload_id)
+            if existing is not None:
+                logger.info(
+                    "素材幂等命中(client_upload_id): library=%s token=%s asset=%s",
+                    library_id,
+                    client_upload_id,
+                    getattr(existing, "id", "?"),
+                )
+                return existing
+    if file_hash:
+        existing = asset_repository.find_by_library_and_file_hash(
+            library_id=library_id,
+            file_hash=file_hash,
+        )
+        if existing is not None:
+            logger.info(
+                "素材去重命中(file_hash): library=%s hash=%s asset=%s",
+                library_id,
+                file_hash,
+                existing.id,
+            )
+            return existing
+    if filename:
+        find_recent = getattr(asset_repository, "find_recent_active_by_library_and_name", None)
+        if callable(find_recent):
+            existing = find_recent(
+                library_id=library_id,
+                name=filename,
+                within_minutes=FALLBACK_DEDUP_WINDOW_MINUTES,
+                file_size=file_size or 0,
+            )
+            if existing is not None and getattr(existing, "status", None) in ACTIVE_ASSET_STATUSES:
+                logger.info(
+                    "素材幂等兜底命中(近期活动同名记录): library=%s name=%s asset=%s status=%s",
+                    library_id,
+                    filename,
+                    getattr(existing, "id", "?"),
+                    getattr(existing, "status", "?"),
+                )
+                return existing
+    return None
+
+
 def _create_pending_asset(
-    asset_repository, project_id, library_id, storage_key, filename, mime_type, user_id, file_hash=""
+    asset_repository,
+    project_id,
+    library_id,
+    storage_key,
+    filename,
+    mime_type,
+    user_id,
+    file_hash="",
+    client_upload_id="",
 ):
     """立即创建一条 PROCESSING 状态的 Asset 记录，使前端能马上看到新素材。"""
     asset = Asset.create(
@@ -111,6 +198,7 @@ def _create_pending_asset(
         status=AssetStatus.PROCESSING,
         uploaded_by_user_id=user_id,
         file_hash=file_hash,
+        client_upload_id=client_upload_id,
     )
     return asset_repository.create(asset)
 
@@ -121,6 +209,7 @@ def _submit_ingest_job(
     storage_key: str,
     ingest_job_repository: Any,
     file_hash: str = "",
+    asset_id: str = "",
 ) -> Any:
     use_case = SubmitIngestJobUseCase(ingest_job_repository)
     job = use_case.execute(
@@ -129,6 +218,7 @@ def _submit_ingest_job(
             library_id=library_id,
             storage_key=storage_key,
             file_hash=file_hash,
+            asset_id=asset_id,
         )
     )
     celery_app.send_task("worker.ingest_asset", args=[job.id])
@@ -202,7 +292,7 @@ async def complete_direct_upload(
     asset_repository: Any = Depends(get_asset_repository),
     storage_service: OSSStorageService = Depends(get_storage_service),
 ) -> DirectUploadCompleteResponse:
-    """确认浏览器直传完成并创建导入任务。"""
+    """确认浏览器直传完成并创建导入任务（幂等：重复 complete 返回同一素材）。"""
     require_project_and_library(
         request.project_id,
         request.library_id,
@@ -212,6 +302,29 @@ async def complete_direct_upload(
     normalized_key = storage_service._normalize_storage_key(request.storage_key)
     if not normalized_key.startswith("uploads/"):
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid upload key")
+
+    filename = normalized_key.rsplit("/", 1)[-1]
+
+    # ── 幂等去重（放在 OSS 检查之前）：complete 超时后前端重试时，
+    # 第一次 complete 可能已建好占位记录，此时即使 OSS 检查失败也必须返回
+    # 已存在记录，绝不能再建第二条。─
+    existing = _find_duplicate_asset(
+        asset_repository,
+        library_id=request.library_id,
+        file_hash=request.file_hash,
+        client_upload_id=request.client_upload_id,
+        filename=filename,
+        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),
+        )
+
     try:
         file_exists = storage_service.file_exists(normalized_key)
     except Exception as error:
@@ -223,29 +336,7 @@ async def complete_direct_upload(
     if not file_exists:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Uploaded file not found")
 
-    # ── 素材去重检测：同素材库 + 同 file_hash 视为重复 ──
-    if request.file_hash:
-        existing = asset_repository.find_by_library_and_file_hash(
-            library_id=request.library_id,
-            file_hash=request.file_hash,
-        )
-        if existing is not None:
-            logger.info(
-                "素材去重命中: library=%s hash=%s existing_asset=%s",
-                request.library_id,
-                request.file_hash,
-                existing.id,
-            )
-            return DirectUploadCompleteResponse(
-                storage_key=normalized_key,
-                ingest_job_id="",
-                duplicated=True,
-                asset_id=existing.id,
-                url=storage_service.get_url(normalized_key),
-            )
-
     # 立即创建 Asset 记录（PROCESSING 状态），使前端刷新后即可看到新素材
-    filename = normalized_key.rsplit("/", 1)[-1]
     mime_type = _infer_mime_type_from_storage_key(normalized_key)
     pending_asset = _create_pending_asset(
         asset_repository=asset_repository,
@@ -256,6 +347,7 @@ async def complete_direct_upload(
         mime_type=mime_type,
         user_id=authenticated_user.user.id,
         file_hash=request.file_hash,
+        client_upload_id=request.client_upload_id,
     )
 
     job = _submit_ingest_job(
@@ -264,6 +356,7 @@ async def complete_direct_upload(
         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,
@@ -283,7 +376,8 @@ async def upload_asset(
     project_id: str = Form(..., min_length=1, description="项目 ID"),
     library_id: str = Form(..., min_length=1, description="素材库 ID"),
     file: UploadFile = File(..., description="要上传的文件（视频、音频、图片等）"),
-    file_hash: str = Form(default="", description="文件 MD5 哈希，用于去重检测"),
+    file_hash: str = Form(default="", description="文件哈希，用于去重检测"),
+    client_upload_id: str = Form(default="", description="客户端幂等 token（同一次上传的重试保持一致）"),
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     ingest_job_repository: Any = Depends(get_ingest_job_repository),
     project_repository: Any = Depends(get_project_repository),
@@ -294,32 +388,31 @@ async def upload_asset(
     """上传素材文件并触发导入流水线。"""
     require_project_and_library(project_id, library_id, project_repository, asset_library_repository)
 
-    # ── 素材去重检测：上传前检查同素材库 + 同 file_hash ──
-    if file_hash:
-        existing = asset_repository.find_by_library_and_file_hash(
-            library_id=library_id,
-            file_hash=file_hash,
-        )
-        if existing is not None:
-            logger.info(
-                "素材去重命中(multipart): library=%s hash=%s existing_asset=%s",
-                library_id,
-                file_hash,
-                existing.id,
-            )
-            return UploadAssetResponse(
-                storage_key=existing.storage_key,
-                ingest_job_id="",
-                url="",
-                duplicated=True,
-                asset_id=existing.id,
-            )
-
-    # P2-5: 服务端验证 MIME 类型
+    # P2-5: 服务端验证 MIME 类型（先验证，再幂等去重，避免非法类型绕过）
     validated_content_type = _validate_mime_type(file.content_type)
 
-    file_id = uuid4().hex[:8]
     safe_filename = file.filename.replace("/", "_").replace("\\", "_") if file.filename else "unknown"
+
+    # ── 幂等去重：client_upload_id → file_hash → 近期活动同名记录兜底 ──
+    # 放在 OSS 上传之前：重复提交直接返回，不占 OSS 流量、不建新记录。
+    existing = _find_duplicate_asset(
+        asset_repository,
+        library_id=library_id,
+        file_hash=file_hash,
+        client_upload_id=client_upload_id,
+        filename=safe_filename,
+        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,
+        )
+
+    file_id = uuid4().hex[:8]
     storage_key = f"uploads/{file_id}/{safe_filename}"
 
     try:
@@ -348,6 +441,7 @@ async def upload_asset(
         mime_type=validated_content_type,
         user_id=authenticated_user.user.id,
         file_hash=file_hash,
+        client_upload_id=client_upload_id,
     )
 
     job = _submit_ingest_job(
@@ -356,6 +450,7 @@ async def upload_asset(
         storage_key=storage_key,
         ingest_job_repository=ingest_job_repository,
         file_hash=file_hash,
+        asset_id=pending_asset.id,
     )
 
     return UploadAssetResponse(

apps/api/app/schemas/upload.py

@@ -31,14 +31,16 @@ class DirectUploadCompleteRequest(BaseModel):
     project_id: str = Field(..., min_length=1)
     library_id: str = Field(..., min_length=1)
     storage_key: str = Field(..., min_length=1, max_length=255)
-    file_hash: str = Field(default="", max_length=64, description="文件 MD5 哈希，用于去重检测")
+    file_hash: str = Field(default="", max_length=64, description="文件哈希，用于去重检测")
+    client_upload_id: str = Field(default="", max_length=64, description="客户端幂等 token（同一次上传的重试保持一致）")
+    file_size: int = Field(default=0, ge=0, description="文件大小（字节），用于无 hash 时的兜底去重")
 
 
 class DirectUploadCompleteResponse(BaseModel):
     storage_key: str
     ingest_job_id: str
-    duplicated: bool = Field(default=False, description="是否为重复素材（命中去重）")
-    asset_id: str = Field(default="", description="重复素材的 asset_id（duplicated=true 时返回）")
+    duplicated: bool = Field(default=False, description="是否为重复素材/重复 complete（命中幂等去重）")
+    asset_id: str = Field(default="", description="素材 asset_id（重复 complete 时返回已存在记录）")
     url: str = Field(default="", description="Public URL of uploaded file")
 
 
@@ -46,5 +48,5 @@ class UploadAssetResponse(BaseModel):
     storage_key: str
     ingest_job_id: str
     url: str = Field(..., description="Public URL of uploaded file")
-    duplicated: bool = Field(default=False, description="是否为重复素材（命中去重）")
-    asset_id: str = Field(default="", description="重复素材的 asset_id（duplicated=true 时返回）")
+    duplicated: bool = Field(default=False, description="是否为重复素材/重复提交（命中幂等去重）")
+    asset_id: str = Field(default="", description="素材 asset_id（重复提交时返回已存在记录）")

apps/worker/worker_app/tasks/ingest.py

@@ -359,6 +359,54 @@ def validate_transcode_output(
     return True
 
 
+def _original_key_from_storage_key(storage_key: str) -> str:
+    """从可能被 HEVC 转码改写的 storage_key 还原原始 key。
+
+    转码成功后 key 形如 uploads/<id>/IMG_2282_h264.MOV，
+    占位 asset 以原始 key uploads/<id>/IMG_2282.MOV 创建。
+    """
+    if not storage_key:
+        return storage_key
+    _p = Path(storage_key)
+    if _p.stem.endswith("_h264"):
+        return str(_p.parent / (_p.stem[: -len("_h264")] + _p.suffix))
+    return storage_key
+
+
+def _resolve_placeholder_asset(asset_repo, job, original_storage_key):
+    """找到 complete 阶段创建的 PROCESSING 占位 asset（Issue #1714）。
+
+    HEVC 转码成功后 job.storage_key 会被改写为 *_h264 新 key，旧实现用新 key
+    回查占位必然落空，进而兜底新建一条 READY 记录，导致原占位永久卡 processing。
+
+    查找优先级：
+    1. job.asset_id（complete 派单时透传的占位 id，最可靠，不依赖 key）；
+    2. 原始 storage_key（占位记录以原始 key 创建）；
+    3. 当前 job.storage_key（未转码/降级场景与原始 key 相同）。
+
+    找不到返回 None（旧链路兼容，由调用方兜底新建并告警）。
+    """
+    asset_id = getattr(job, "asset_id", "") or ""
+    if asset_id:
+        try:
+            found = asset_repo.find_by_id(asset_id)
+            if found is not None:
+                return found
+        except Exception as find_err:
+            logger.warning("占位 asset 按 id 查询失败 asset_id=%s: %s", asset_id, find_err)
+    for key in (original_storage_key, getattr(job, "storage_key", "")):
+        if not key:
+            continue
+        try:
+            found = asset_repo.find_by_storage_key(key)
+        except Exception:
+            logger.warning("find_by_storage_key not available, trying fallback lookup")
+            found = None
+        if found is not None:
+            return found
+    return None
+
+
 @celery_app.task(name="worker.ingest_asset")
 def ingest_asset(job_id: str) -> dict:
     """
@@ -380,6 +428,10 @@ def ingest_asset(job_id: str) -> dict:
         if job is None:
             return {"status": "failed", "error": "job not found"}
 
+        # 记录原始 storage_key：HEVC 转码成功后 job.storage_key 会改写为 *_h264，
+        # 而 complete 阶段的占位 asset 始终以原始 key 创建，关联回写必须保留它。
+        original_storage_key = job.storage_key
+
         # Update job status to PROCESSING
         job.status = IngestJobStatus.PROCESSING
         job.updated_at = datetime.now(timezone.utc)
@@ -624,22 +676,41 @@ def ingest_asset(job_id: str) -> dict:
                 error_reason,
             )
 
-            asset = Asset.create(
-                project_id=job.project_id,
-                library_id=job.library_id,
-                name=filename,
-                storage_key=job.storage_key,
-                mime_type=mime_type,
-                metadata={"source": "upload", "ingest_error": error_reason},
-                file_size=int(metadata.get("size_bytes", 0)),
-                duration=float(metadata.get("duration", 0)),
-                width=int(metadata.get("width", 0)),
-                height=int(metadata.get("height", 0)),
-                codec=metadata.get("codec") or None,
-                status=AssetStatus.ERROR,
-                file_hash=job.file_hash,
-            )
-            asset_repo.create(asset)
+            placeholder = _resolve_placeholder_asset(asset_repo, job, original_storage_key)
+            if placeholder is not None:
+                # 回写占位记录：标 ERROR（Issue #1714：禁止新建第二条导致占位孤儿）
+                asset = placeholder
+                asset.mime_type = mime_type
+                asset.metadata = {"source": "upload", "ingest_error": error_reason}
+                asset.file_size = int(metadata.get("size_bytes", 0))
+                asset.duration = float(metadata.get("duration", 0)) or None
+                asset.width = int(metadata.get("width", 0)) or None
+                asset.height = int(metadata.get("height", 0)) or None
+                codec_val = metadata.get("codec")
+                if codec_val:
+                    asset.codec = str(codec_val)
+                asset.status = AssetStatus.ERROR
+                asset.updated_at = datetime.now(timezone.utc)
+                asset_repo.update(asset)
+            else:
+                # 旧链路兜底：无占位记录（如历史 job 重跑）才新建
+                logger.warning("无效素材且未找到占位记录，兜底新建 ERROR asset: job_id=%s", job_id)
+                asset = Asset.create(
+                    project_id=job.project_id,
+                    library_id=job.library_id,
+                    name=filename,
+                    storage_key=job.storage_key,
+                    mime_type=mime_type,
+                    metadata={"source": "upload", "ingest_error": error_reason},
+                    file_size=int(metadata.get("size_bytes", 0)),
+                    duration=float(metadata.get("duration", 0)),
+                    width=int(metadata.get("width", 0)),
+                    height=int(metadata.get("height", 0)),
+                    codec=metadata.get("codec") or None,
+                    status=AssetStatus.ERROR,
+                    file_hash=job.file_hash,
+                )
+                asset_repo.create(asset)
 
             # Update job status to FAILED
             job.status = IngestJobStatus.FAILED
@@ -656,16 +727,19 @@ def ingest_asset(job_id: str) -> dict:
                 "error": error_reason,
             }
 
-        # 查找已存在的 Asset 记录（由 API 端在上传完成时立即创建为 PROCESSING 状态）
-        existing_asset = None
-        try:
-            existing_asset = asset_repo.find_by_storage_key(job.storage_key)
-        except Exception:
-            logger.warning("find_by_storage_key not available, trying fallback lookup")
+        # 查找 complete 阶段创建的占位 Asset 记录（Issue #1714）。
+        # 必须用原始 storage_key / job.asset_id 关联——HEVC 转码后 job.storage_key
+        # 已改写为 *_h264，用新 key 回查占位必然落空，旧实现因此兜底新建 READY 记录，
+        # 导致原 PROCESSING 占位永久卡住（每个 HEVC 视频产生两条记录）。
+        existing_asset = _resolve_placeholder_asset(asset_repo, job, original_storage_key)
 
         if existing_asset is None:
-            # 兜底：如果 API 端没有预先创建 Asset（旧版本兼容），则创建新记录
-            logger.info("No pre-created asset found for storage_key=%s, creating new", job.storage_key)
+            # 兜底：仅当确实没有占位记录（旧版本 API / 历史 job 重跑）才新建。
+            logger.warning(
+                "No placeholder asset found for job_id=%s original_key=%s, creating new",
+                job_id,
+                original_storage_key,
+            )
             metadata["source"] = "upload"
             asset = Asset.create(
                 project_id=job.project_id,
@@ -685,8 +759,13 @@ def ingest_asset(job_id: str) -> dict:
             )
             asset_repo.create(asset)
         else:
-            # 更新已有的 Asset 记录，补充元数据并将状态改为 READY
+            # 更新占位记录：补充元数据、置 READY。转码成功时 storage_key 同步改写为
+            # *_h264（播放/下载走转码产物），原始 key 记入 metadata 可溯源。
             asset = existing_asset
+            if job.storage_key != asset.storage_key:
+                metadata["original_storage_key"] = asset.storage_key
+                metadata["hevc_transcoded"] = True
+                asset.storage_key = job.storage_key
             asset.mime_type = mime_type
             metadata["source"] = "upload"
             asset.metadata = metadata
@@ -737,9 +816,28 @@ def ingest_asset(job_id: str) -> dict:
                 job_repo.update(job)
 
                 # 将上传时创建的占位 Asset（PROCESSING/UPLOADING）标记为 ERROR，
-                # 避免素材永远卡在中间状态
+                # 避免素材永远卡在中间状态。转码可能已把 job.storage_key 改写为
+                # *_h264，需用 asset_id / 原始 key 多路径关联占位（Issue #1714）。
                 try:
-                    existing = asset_repo.find_by_storage_key(job.storage_key)
+                    existing = None
+                    _asset_id = getattr(job, "asset_id", "") or ""
+                    if _asset_id:
+                        try:
+                            existing = asset_repo.find_by_id(_asset_id)
+                        except Exception:
+                            existing = None
+                    if existing is None:
+                        _candidate_keys = [
+                            _original_key_from_storage_key(job.storage_key),
+                            job.storage_key,
+                        ]
+                        for _key in _candidate_keys:
+                            try:
+                                existing = asset_repo.find_by_storage_key(_key)
+                            except Exception:
+                                existing = None
+                            if existing is not None:
+                                break
                     if existing and existing.status in (
                         AssetStatus.PROCESSING,
                         AssetStatus.UPLOADING,

packages/adapters/in_memory/asset_repository.py

@@ -146,3 +146,44 @@ class InMemoryAssetRepository:
             if asset.library_id == library_id and asset.file_hash == file_hash:
                 return asset
         return None
+
+    def find_by_library_and_client_upload_id(
+        self,
+        library_id: str,
+        client_upload_id: str,
+    ) -> Asset | None:
+        """按素材库 + 客户端幂等 token 查找已有素材。"""
+        if not client_upload_id:
+            return None
+        for asset in self._assets.values():
+            if asset.library_id == library_id and getattr(asset, "client_upload_id", "") == client_upload_id:
+                return asset
+        return None
+
+    def find_recent_active_by_library_and_name(
+        self,
+        library_id: str,
+        name: str,
+        within_minutes: int = 30,
+        file_size: int = 0,
+    ) -> Asset | None:
+        """兜底去重：同库 + 同文件名（+同大小）且近期活动状态的素材。"""
+        from datetime import datetime, timedelta, timezone
+
+        if not name:
+            return None
+        from packages.domain import AssetStatus
+
+        cutoff = datetime.now(timezone.utc) - timedelta(minutes=within_minutes)
+        candidates = [
+            a
+            for a in self._assets.values()
+            if a.library_id == library_id
+            and a.name == name
+            and a.status in (AssetStatus.UPLOADING, AssetStatus.PROCESSING)
+            and a.created_at >= cutoff
+            and (not file_size or file_size <= 0 or a.file_size == file_size)
+        ]
+        if not candidates:
+            return None
+        return max(candidates, key=lambda a: a.created_at)

packages/adapters/sqlalchemy_impl/asset_repository.py

@@ -134,6 +134,7 @@ class SQLAlchemyAssetRepository:
             quality_score=asset.quality_score,
             uploaded_by_user_id=asset.uploaded_by_user_id or "system",
             file_hash=asset.file_hash or None,
+            client_upload_id=asset.client_upload_id or None,
             created_at=asset.created_at,
             updated_at=now,
         )
@@ -163,6 +164,8 @@ class SQLAlchemyAssetRepository:
         model.quality_score = asset.quality_score
         model.uploaded_by_user_id = asset.uploaded_by_user_id or model.uploaded_by_user_id
         model.file_hash = asset.file_hash or model.file_hash
+        if getattr(model, "client_upload_id", None) is None and asset.client_upload_id:
+            model.client_upload_id = asset.client_upload_id
         model.updated_at = datetime.now(timezone.utc)
         self.session.flush()
         self._sync_asset_tags(asset.id, asset.tag_ids)
@@ -388,6 +391,7 @@ class SQLAlchemyAssetRepository:
             quality_score=model.quality_score,
             uploaded_by_user_id=model.uploaded_by_user_id,
             file_hash=model.file_hash or "",
+            client_upload_id=getattr(model, "client_upload_id", None) or "",
             metadata=metadata,
             tag_ids=tag_ids,
             created_at=model.created_at,
@@ -452,3 +456,53 @@ class SQLAlchemyAssetRepository:
         if model is None:
             return None
         return self._to_domain(model)
+
+    def find_by_library_and_client_upload_id(
+        self,
+        library_id: str,
+        client_upload_id: str,
+    ) -> Asset | None:
+        """按素材库 + 客户端幂等 token 查找已有素材（complete 幂等）。"""
+        if not client_upload_id:
+            return None
+        model = (
+            self.session.query(AssetModel)
+            .filter(
+                AssetModel.asset_library_id == library_id,
+                AssetModel.client_upload_id == client_upload_id,
+            )
+            .first()
+        )
+        if model is None:
+            return None
+        return self._to_domain(model)
+
+    def find_recent_active_by_library_and_name(
+        self,
+        library_id: str,
+        name: str,
+        within_minutes: int = 30,
+        file_size: int = 0,
+    ) -> Asset | None:
+        """兜底去重：同库 + 同文件名（+同大小）且近期仍处活动状态（uploading/processing）的素材。
+
+        用于旧客户端未传 file_hash/client_upload_id 时，防止 complete 超时重试
+        反复创建 PROCESSING 占位记录。只命中"活动中"的近期记录，READY 历史素材不拦。
+        """
+        from datetime import datetime, timedelta, timezone
+
+        if not name:
+            return None
+        cutoff = datetime.now(timezone.utc) - timedelta(minutes=within_minutes)
+        query = self.session.query(AssetModel).filter(
+            AssetModel.asset_library_id == library_id,
+            AssetModel.name == name,
+            AssetModel.status.in_([AssetStatus.UPLOADING.value, AssetStatus.PROCESSING.value]),
+            AssetModel.created_at >= cutoff,
+        )
+        if file_size and file_size > 0:
+            query = query.filter(AssetModel.file_size == file_size)
+        model = query.order_by(AssetModel.created_at.desc()).first()
+        if model is None:
+            return None
+        return self._to_domain(model)

packages/adapters/sqlalchemy_impl/ingest_job_repository.py

@@ -18,6 +18,7 @@ class SQLAlchemyIngestJobRepository:
             error_message=job.error_message,
             result_asset_id=job.result_asset_id,
             file_hash=job.file_hash,
+            asset_id=job.asset_id or "",
             created_at=job.created_at,
             updated_at=job.updated_at,
         )
@@ -38,6 +39,7 @@ class SQLAlchemyIngestJobRepository:
             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 "",
             created_at=model.created_at,
             updated_at=model.updated_at,
         )
@@ -54,6 +56,9 @@ class SQLAlchemyIngestJobRepository:
         model.error_message = job.error_message
         model.result_asset_id = job.result_asset_id
         model.file_hash = job.file_hash
+        model.storage_key = job.storage_key
+        if job.asset_id:
+            model.asset_id = job.asset_id
         model.updated_at = job.updated_at
         self.session.commit()
         return job

packages/adapters/sqlalchemy_impl/models.py

@@ -94,6 +94,7 @@ class AssetModel(Base):
     quality_score = Column(Float, nullable=True)
     uploaded_by_user_id = Column(String(36), nullable=False)
     file_hash = Column(String(64), nullable=True, index=True)
+    client_upload_id = Column(String(64), nullable=True, index=True)
     extra_meta = Column("metadata", JSON, nullable=False, default=dict)
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc), index=True)
     updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
@@ -244,6 +245,7 @@ class IngestJobModel(Base):
     error_message = Column(Text, nullable=False, default="")
     result_asset_id = Column(String(36), nullable=False, default="")
     file_hash = Column(String(64), nullable=True, index=True)
+    asset_id = Column(String(36), nullable=False, default="", index=True)
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
     updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
 

packages/application/ingest_jobs.py

@@ -12,6 +12,7 @@ class SubmitIngestJobCommand:
     library_id: str
     storage_key: str
     file_hash: str = ""
+    asset_id: str = ""
 
 
 class SubmitIngestJobUseCase:
@@ -24,5 +25,6 @@ class SubmitIngestJobUseCase:
             library_id=command.library_id,
             storage_key=command.storage_key,
             file_hash=command.file_hash,
+            asset_id=command.asset_id,
         )
         return self.ingest_job_repository.create(job)

packages/domain/entities.py

@@ -174,6 +174,7 @@ class Asset:
     quality_score: float | None = None
     uploaded_by_user_id: str = ""
     file_hash: str = ""
+    client_upload_id: str = ""
     metadata: dict[str, Any] = field(default_factory=dict)
     tag_ids: list[str] = field(default_factory=list)
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@@ -208,6 +209,7 @@ class Asset:
         quality_score: float | None = None,
         uploaded_by_user_id: str = "",
         file_hash: str = "",
+        client_upload_id: str = "",
     ) -> "Asset":
         clean_name = name.strip()
         if not clean_name:
@@ -235,6 +237,7 @@ class Asset:
             quality_score=quality_score,
             uploaded_by_user_id=uploaded_by_user_id.strip(),
             file_hash=file_hash.strip(),
+            client_upload_id=client_upload_id.strip(),
             metadata=metadata or {},
             tag_ids=[],
         )
@@ -266,6 +269,7 @@ class IngestJob:
     error_message: str = ""
     result_asset_id: str = ""
     file_hash: str = ""
+    asset_id: str = ""
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
     updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
 
@@ -276,6 +280,7 @@ class IngestJob:
         library_id: str,
         storage_key: str,
         file_hash: str = "",
+        asset_id: str = "",
     ) -> "IngestJob":
         if not project_id.strip():
             raise ValueError("project_id 不能为空")
@@ -289,4 +294,5 @@ class IngestJob:
             library_id=library_id.strip(),
             storage_key=storage_key.strip(),
             file_hash=file_hash.strip(),
+            asset_id=asset_id.strip(),
         )

packages/ports/asset_repository.py

@@ -125,3 +125,23 @@ class AssetRepository(ABC):
     ) -> Asset | None:
         """按素材库 + 文件哈希查找已有素材（去重检测）。"""
         pass
+
+    @abstractmethod
+    def find_by_library_and_client_upload_id(
+        self,
+        library_id: str,
+        client_upload_id: str,
+    ) -> Asset | None:
+        """按素材库 + 客户端幂等 token 查找已有素材（complete 幂等）。"""
+        pass
+
+    @abstractmethod
+    def find_recent_active_by_library_and_name(
+        self,
+        library_id: str,
+        name: str,
+        within_minutes: int = 30,
+        file_size: int = 0,
+    ) -> Asset | None:
+        """兜底去重：同库 + 同文件名（+同大小）且近期仍在 uploading/processing 的素材。"""
+        pass

tests/unit/test_ingest_hevc_orphan_1714.py

@@ -0,0 +1,317 @@
+"""Issue #1714：HEVC 转码后禁止兜底新建重复 READY 记录，必须回写占位 asset。
+
+覆盖：
+- 转码成功 + 占位 asset 存在（按原始 key 找到）→ 更新占位为 READY、
+  storage_key 改写为 *_h264，绝不 create 新记录（回归 P1 孤儿 PROCESSING bug）
+- job.asset_id 透传时优先按 id 关联占位（即使 key 对不上也能命中）
+- 无占位记录（旧链路）→ 兜底新建（保留兼容）
+- 非 HEVC：占位同样被更新为 READY，不新建
+- 无效媒体：占位标记为 ERROR，不新建 ERROR 记录
+- ingest 异常：占位（按还原后的原始 key）标记 ERROR
+"""
+
+from __future__ import annotations
+
+import sys
+import tempfile
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+# ── 与 test_ingest_hevc_transcode_task.py 相同的 worker 模块加载方式 ──
+_SAVED_MODULES_KEYS = set(sys.modules.keys())
+
+_mock_db_module = MagicMock()
+_mock_db_module.SessionLocal = MagicMock()
+sys.modules["worker_app.db"] = _mock_db_module
+sys.modules["worker_app.core.config"] = MagicMock()
+
+_mock_celery_module = MagicMock()
+
+
+def _passthrough_decorator(*args, **kwargs):
+    if len(args) == 1 and callable(args[0]):
+        return args[0]
+    return lambda f: f
+
+
+_mock_celery_module.celery_app.task = MagicMock(side_effect=_passthrough_decorator)
+sys.modules["worker_app.celery_app"] = _mock_celery_module
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
+
+import pytest  # noqa: E402
+from worker_app.tasks import ingest as ingest_mod  # noqa: E402
+
+from packages.domain import Asset, AssetStatus  # noqa: E402
+
+for _key in list(sys.modules.keys()):
+    if _key not in _SAVED_MODULES_KEYS and not _key.startswith("video_processing"):
+        del sys.modules[_key]
+del _SAVED_MODULES_KEYS
+
+
+# ── 假仓储 ──────────────────────────────────────────────────────────────
+class _FakeJobRepo:
+    def __init__(self, job):
+        self.job = job
+        self.updated = None
+
+    def get(self, job_id):
+        return self.job
+
+    def update(self, job):
+        self.updated = job
+        return job
+
+
+class _FakeAssetRepo:
+    """记录 create 调用；find_* 按内部 assets 列表查询。"""
+
+    def __init__(self, assets: list[Asset] | None = None):
+        self.assets = list(assets or [])
+        self.created: list[Asset] = []
+        self.updated: list[Asset] = []
+
+    def create(self, asset: Asset) -> Asset:
+        self.created.append(asset)
+        self.assets.append(asset)
+        return asset
+
+    def update(self, asset: Asset) -> Asset:
+        self.updated.append(asset)
+        return asset
+
+    def find_by_id(self, asset_id: str) -> Asset | None:
+        return next((a for a in self.assets if a.id == asset_id), None)
+
+    def find_by_storage_key(self, storage_key: str) -> Asset | None:
+        return next((a for a in self.assets if a.storage_key == storage_key), None)
+
+
+def _make_job(asset_id: str = "", storage_key: str = "uploads/proj/IMG_2282.MOV"):
+    return SimpleNamespace(
+        id="job-1",
+        project_id="proj-1",
+        library_id="lib-1",
+        storage_key=storage_key,
+        file_hash="hash-1",
+        asset_id=asset_id,
+        status=None,
+        error_message=None,
+        result_asset_id=None,
+        updated_at=None,
+    )
+
+
+def _make_placeholder(storage_key: str = "uploads/proj/IMG_2282.MOV", asset_id: str = "asset-ph"):
+    return Asset(
+        id=asset_id,
+        project_id="proj-1",
+        library_id="lib-1",
+        name="IMG_2282.MOV",
+        storage_key=storage_key,
+        mime_type="video/quicktime",
+        status=AssetStatus.PROCESSING,
+        file_hash="hash-1",
+    )
+
+
+def _video_metadata(codec="hevc"):
+    return {
+        "codec": codec,
+        "width": 1920,
+        "height": 1080,
+        "duration": 10.0,
+        "size_bytes": 5 * 1024 * 1024,
+    }
+
+
+@pytest.fixture
+def transcode_env(tmp_path):
+    """HEVC 转码成功的标准 mock 环境（同 test_ingest_hevc_transcode_task）。"""
+    local_file = tmp_path / "local_hevc.MOV"
+    local_file.write_bytes(b"fake-hevc-source")
+    tc_out = tmp_path / "transcode_out_h264.mp4"
+
+    control = {
+        "validate_ok": True,
+        "tc_out": tc_out,
+        "local_file": local_file,
+        "download_ok": True,
+        "extract_success": True,
+        "codec": "hevc",
+        "raise_in_flow": None,
+    }
+
+    def fake_ntf(*args, **kwargs):
+        mock_file = MagicMock()
+        mock_file.name = str(tc_out) if kwargs.get("suffix") == "_h264.mp4" else str(local_file)
+        mock_file.close = MagicMock()
+        mock_file.__enter__.return_value = mock_file
+        mock_file.__exit__.return_value = False
+        return mock_file
+
+    def fake_subprocess_run(cmd, **kwargs):
+        if cmd and cmd[0] == "ffmpeg" and "libx264" in cmd:
+            Path(cmd[-1]).write_bytes(b"fake-h264-output")
+            return SimpleNamespace(returncode=0, stderr="")
+        return SimpleNamespace(returncode=0, stdout="", stderr="")
+
+    control["patchers"] = {
+        "session": patch.object(ingest_mod, "SessionLocal", return_value=MagicMock()),
+        "download": patch.object(ingest_mod, "download_asset", side_effect=lambda *a, **kw: control["download_ok"]),
+        "upload": patch("video_processing.oss_helpers.upload_to_oss", return_value="https://oss/x"),
+        "metadata": patch.object(
+            ingest_mod,
+            "extract_media_metadata",
+            side_effect=lambda path, mt: (
+                (_video_metadata("h264"), control["extract_success"])
+                if Path(path).name == tc_out.name
+                else (_video_metadata(control["codec"]), control["extract_success"])
+            ),
+        ),
+        "validate": patch.object(
+            ingest_mod, "validate_transcode_output", side_effect=lambda p, portrait: control["validate_ok"]
+        ),
+        "subprocess": patch.object(ingest_mod.subprocess, "run", side_effect=fake_subprocess_run),
+        "ntf": patch.object(tempfile, "NamedTemporaryFile", side_effect=fake_ntf),
+        "thumb": patch(
+            "video_processing.thumbnail_generator.extract_first_frame",
+            side_effect=RuntimeError("skip thumb"),
+        ),
+    }
+    return control
+
+
+def _start(control, job, assets):
+    job_repo = _FakeJobRepo(job)
+    asset_repo = _FakeAssetRepo(assets)
+    patchers = dict(control["patchers"])
+    patchers["job_repo"] = patch.object(ingest_mod, "SQLAlchemyIngestJobRepository", return_value=job_repo)
+    patchers["asset_repo"] = patch.object(ingest_mod, "SQLAlchemyAssetRepository", return_value=asset_repo)
+    started = {name: p.start() for name, p in patchers.items()}
+    return started, job_repo, asset_repo
+
+
+def _stop(control):
+    for p in control["patchers"].values():
+        p.stop()
+
+
+class TestHEVCTranscodePlaceholderRewrite:
+    def test_transcode_success_updates_placeholder_no_duplicate_ready(self, transcode_env):
+        """转码成功 → 占位 asset 原地更新为 READY + storage_key 改写 _h264，禁止新建。"""
+        control = transcode_env
+        placeholder = _make_placeholder()
+        job = _make_job()  # 旧 job 无 asset_id，靠原始 key 关联
+        mocks, job_repo, asset_repo = _start(control, job, [placeholder])
+        try:
+            result = ingest_mod.ingest_asset("job-1")
+        finally:
+            _stop(control)
+
+        assert result["status"] == "completed"
+        # 核心断言 1：没有新建任何 READY 记录（旧 bug 会 create 一条 _h264 READY）
+        assert asset_repo.created == [], "转码回写不得新建 asset 记录"
+        # 核心断言 2：占位被更新为 READY，且 storage_key 已是 _h264
+        assert len(asset_repo.updated) == 1
+        updated = asset_repo.updated[0]
+        assert updated.id == placeholder.id
+        assert updated.status == AssetStatus.READY
+        assert updated.storage_key == "uploads/proj/IMG_2282_h264.MOV"
+        assert updated.metadata.get("hevc_transcoded") is True
+        assert updated.metadata.get("original_storage_key") == "uploads/proj/IMG_2282.MOV"
+        # job 关联到同一条 asset
+        assert job_repo.updated.result_asset_id == placeholder.id
+        assert job_repo.updated.storage_key == "uploads/proj/IMG_2282_h264.MOV"
+
+    def test_placeholder_resolved_by_job_asset_id(self, transcode_env):
+        """job.asset_id 透传时优先按 id 关联（即使 storage_key 对不上也命中）。"""
+        control = transcode_env
+        placeholder = _make_placeholder(storage_key="uploads/different/key.MOV", asset_id="asset-by-id")
+        job = _make_job(asset_id="asset-by-id")
+        _, _, asset_repo = _start(control, job, [placeholder])
+        try:
+            result = ingest_mod.ingest_asset("job-1")
+        finally:
+            _stop(control)
+
+        assert result["status"] == "completed"
+        assert asset_repo.created == []
+        assert len(asset_repo.updated) == 1
+        assert asset_repo.updated[0].id == "asset-by-id"
+        assert asset_repo.updated[0].status == AssetStatus.READY
+
+    def test_no_placeholder_fallback_creates_ready(self, transcode_env):
+        """旧链路无占位记录 → 兜底新建 READY（兼容保留，但必须是唯一一条）。"""
+        control = transcode_env
+        job = _make_job(asset_id="")
+        _, _, asset_repo = _start(control, job, [])
+        try:
+            result = ingest_mod.ingest_asset("job-1")
+        finally:
+            _stop(control)
+
+        assert result["status"] == "completed"
+        assert len(asset_repo.created) == 1
+        created = asset_repo.created[0]
+        assert created.status == AssetStatus.READY
+        assert created.storage_key == "uploads/proj/IMG_2282_h264.MOV"
+        assert asset_repo.updated == []
+
+    def test_non_hevc_placeholder_updated_no_create(self, transcode_env):
+        """非 HEVC（h264）不转码：占位按原始 key 找到并更新 READY，不新建。"""
+        control = transcode_env
+        control["codec"] = "h264"
+        placeholder = _make_placeholder()
+        job = _make_job()
+        mocks, _, asset_repo = _start(control, job, [placeholder])
+        try:
+            result = ingest_mod.ingest_asset("job-1")
+        finally:
+            _stop(control)
+
+        assert result["status"] == "completed"
+        assert asset_repo.created == []
+        assert len(asset_repo.updated) == 1
+        updated = asset_repo.updated[0]
+        assert updated.status == AssetStatus.READY
+        assert updated.storage_key == "uploads/proj/IMG_2282.MOV"  # 未转码，key 不变
+        mocks["upload"].assert_not_called()
+
+    def test_invalid_media_marks_placeholder_error_no_create(self, transcode_env):
+        """无效媒体：占位标记 ERROR 并 update，禁止再 create 一条 ERROR。"""
+        control = transcode_env
+        control["download_ok"] = False  # 下载失败 → extract_success=False → 无效媒体路径
+        placeholder = _make_placeholder()
+        job = _make_job()
+        _, job_repo, asset_repo = _start(control, job, [placeholder])
+        try:
+            result = ingest_mod.ingest_asset("job-1")
+        finally:
+            _stop(control)
+
+        assert result["status"] == "failed"
+        assert asset_repo.created == [], "无效媒体不得新建 ERROR 记录"
+        assert len(asset_repo.updated) == 1
+        assert asset_repo.updated[0].id == placeholder.id
+        assert asset_repo.updated[0].status == AssetStatus.ERROR
+        assert job_repo.updated.result_asset_id == placeholder.id
+
+    def test_exception_path_marks_placeholder_error(self, transcode_env):
+        """ingest 主流程抛异常（如元数据提取炸了）→ 占位按原始 key 找到并标 ERROR。"""
+        control = transcode_env
+        placeholder = _make_placeholder()
+        job = _make_job()
+        started, _, asset_repo = _start(control, job, [placeholder])
+        started["metadata"].side_effect = RuntimeError("boom in flow")
+        try:
+            result = ingest_mod.ingest_asset("job-1")
+        finally:
+            _stop(control)
+
+        assert result["status"] == "failed"
+        # 异常路径把占位标 ERROR（旧实现用被改写的 _h264 key 回查会落空）
+        error_marked = [a for a in asset_repo.assets if a.id == placeholder.id and a.status == AssetStatus.ERROR]
+        assert error_marked, "异常路径必须把占位 asset 标为 ERROR"

tests/unit/test_upload_complete_idempotency_1714.py

@@ -0,0 +1,338 @@
+"""Issue #1714：POST /upload/direct/complete 幂等 + multipart 幂等。
+
+覆盖：
+- 同 client_upload_id 重复 complete → 只建一条 asset、不重复派 ingest job
+- 同 file_hash 重复 complete → 返回已存在记录
+- 旧客户端不传 hash/token：近期同库同名 processing 占位 → 兜底幂等返回
+- 旧客户端不传 hash/token：READY 历史同名 → 不兜底（正常新建）
+- 兜底窗口外（>30 分钟）→ 不兜底
+- 旧仓储（无新方法）鸭子类型降级 → 不报错、正常新建
+- 重复 complete 时即使 OSS 已无文件（file_exists=False）也返回已存在记录
+  （模拟 complete 超时后 OSS 侧对象已过期/清理，重试仍不重复建库）
+- multipart 上传：同 client_upload_id 重复提交 → 第二次直接 duplicated，不再传 OSS
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest.mock import MagicMock
+
+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 fastapi import FastAPI  # noqa: E402
+from fastapi.testclient import TestClient  # noqa: E402
+
+from packages.domain import Asset, AssetLibrary, AssetLibraryKind, AssetStatus, IngestJob, Project  # noqa: E402
+
+
+class StubProjectRepository:
+    def __init__(self, projects: dict | None = None):
+        self._projects = projects or {}
+
+    def get(self, project_id: str):
+        return self._projects.get(project_id)
+
+    def find_by_id(self, project_id: str):
+        return self._projects.get(project_id)
+
+
+class StubAssetLibraryRepository:
+    def __init__(self, libraries: dict | None = None):
+        self._libraries = libraries or {}
+
+    def find_by_project(self, project_id: str, kind=None) -> list:
+        return list(self._libraries.values())
+
+
+class StubAssetRepository:
+    """支持三种幂等查询的内存仓储，并统计 create 次数。"""
+
+    def __init__(self, assets: list[Asset] | None = None):
+        self._assets = list(assets or [])
+        self.created: list[Asset] = []
+
+    def find_by_library_and_file_hash(self, library_id: str, file_hash: str) -> Asset | None:
+        if not file_hash:
+            return None
+        return next((a for a in self._assets if a.library_id == library_id and a.file_hash == file_hash), None)
+
+    def find_by_library_and_client_upload_id(self, library_id: str, client_upload_id: str) -> Asset | None:
+        if not client_upload_id:
+            return None
+        return next(
+            (a for a in self._assets if a.library_id == library_id and a.client_upload_id == client_upload_id),
+            None,
+        )
+
+    def find_recent_active_by_library_and_name(
+        self, library_id: str, name: str, within_minutes: int = 30, file_size: int = 0
+    ) -> Asset | None:
+        cutoff = datetime.now(timezone.utc) - timedelta(minutes=within_minutes)
+        candidates = [
+            a
+            for a in self._assets
+            if a.library_id == library_id
+            and a.name == name
+            and a.status in (AssetStatus.UPLOADING, AssetStatus.PROCESSING)
+            and a.created_at >= cutoff
+            and (not file_size or a.file_size == file_size)
+        ]
+        return max(candidates, key=lambda a: a.created_at) if candidates else None
+
+    def create(self, asset: Asset) -> Asset:
+        self._assets.append(asset)
+        self.created.append(asset)
+        return asset
+
+    def update(self, asset: Asset) -> Asset:
+        return asset
+
+
+class LegacyStubAssetRepository:
+    """旧仓储：只有 file_hash 去重，没有新方法（鸭子类型降级验证）。"""
+
+    def __init__(self, assets: list[Asset] | None = None):
+        self._assets = list(assets or [])
+        self.created: list[Asset] = []
+
+    def find_by_library_and_file_hash(self, library_id: str, file_hash: str) -> Asset | None:
+        if not file_hash:
+            return None
+        return next((a for a in self._assets if a.library_id == library_id and a.file_hash == file_hash), None)
+
+    def create(self, asset: Asset) -> Asset:
+        self._assets.append(asset)
+        self.created.append(asset)
+        return asset
+
+
+class StubIngestJobRepository:
+    def __init__(self):
+        self._jobs: dict[str, IngestJob] = {}
+        self.created_count = 0
+
+    def create(self, job: IngestJob) -> IngestJob:
+        self._jobs[job.id] = job
+        self.created_count += 1
+        return job
+
+    def get(self, job_id: str) -> IngestJob | None:
+        return self._jobs.get(job_id)
+
+    def update(self, job: IngestJob) -> IngestJob:
+        self._jobs[job.id] = job
+        return job
+
+
+def _make_project() -> Project:
+    return Project(id="proj-1", name="Test Project", owner_user_id="user-1")
+
+
+def _make_library() -> AssetLibrary:
+    return AssetLibrary(id="lib-1", name="Test Library", project_id="proj-1", kind=AssetLibraryKind.VIDEO)
+
+
+def _build_app(asset_repo=None, ingest_repo=None, storage=None):
+    from app.api.routes.upload import router
+    from app.auth import AuthenticatedUser, get_current_user
+    from app.core.storage import get_storage_service
+    from app.dependencies import (
+        get_asset_library_repository,
+        get_asset_repository,
+        get_ingest_job_repository,
+        get_project_repository,
+    )
+
+    app = FastAPI()
+    app.include_router(router, prefix="/api/v1")
+
+    project_repo = StubProjectRepository({"proj-1": _make_project()})
+    library_repo = StubAssetLibraryRepository({"lib-1": _make_library()})
+    asset_repo = asset_repo or StubAssetRepository()
+    ingest_repo = ingest_repo or StubIngestJobRepository()
+
+    storage = storage or MagicMock()
+    storage.is_configured = True
+    storage._normalize_storage_key = lambda key: key
+    storage.file_exists = MagicMock(return_value=True)
+    storage.upload_file = MagicMock(return_value="https://oss.example.com/file.mp4")
+    storage.get_url = MagicMock(return_value="https://oss.example.com/file.mp4")
+
+    mock_user = MagicMock(spec=AuthenticatedUser)
+    mock_user.id = "user-1"
+    mock_user.user = MagicMock(id="user-1")
+    mock_user.email = "test@example.com"
+
+    app.dependency_overrides[get_current_user] = lambda: mock_user
+    app.dependency_overrides[get_project_repository] = lambda: project_repo
+    app.dependency_overrides[get_asset_library_repository] = lambda: library_repo
+    app.dependency_overrides[get_asset_repository] = lambda: asset_repo
+    app.dependency_overrides[get_ingest_job_repository] = lambda: ingest_repo
+    app.dependency_overrides[get_storage_service] = lambda: storage
+    return app, asset_repo, ingest_repo, storage
+
+
+def _client(**kwargs):
+    app, asset_repo, ingest_repo, storage = _build_app(**kwargs)
+    return TestClient(app), asset_repo, ingest_repo, storage
+
+
+COMPLETE_BODY = {
+    "project_id": "proj-1",
+    "library_id": "lib-1",
+    "storage_key": "uploads/abc/IMG_2282.MOV",
+}
+
+
+class TestDirectCompleteIdempotency:
+    def test_same_client_upload_id_creates_single_asset_and_job(self):
+        """同一 client_upload_id 连发两次 complete：只建 1 条 asset、1 个 job。"""
+        client, asset_repo, ingest_repo, _ = _client()
+        body = {**COMPLETE_BODY, "client_upload_id": "up-token-1", "file_size": 12345}
+
+        r1 = client.post("/api/v1/direct/complete", json=body)
+        r2 = client.post("/api/v1/direct/complete", json={**body, "storage_key": "uploads/zzz/IMG_2282.MOV"})
+
+        assert r1.status_code == 200 and r2.status_code == 200
+        b1, b2 = r1.json(), r2.json()
+        assert b1["duplicated"] is False
+        assert b2["duplicated"] is True
+        assert b1["asset_id"] == b2["asset_id"]
+        assert len(asset_repo.created) == 1
+        assert ingest_repo.created_count == 1
+        # 第二次返回的是已存在记录（其 storage_key 为第一次的 key）
+        assert b2["storage_key"] == "uploads/abc/IMG_2282.MOV"
+
+    def test_same_file_hash_returns_existing(self):
+        """同 file_hash（不同 token）重复 complete → 返回已存在记录。"""
+        client, asset_repo, ingest_repo, _ = _client()
+        body1 = {**COMPLETE_BODY, "file_hash": "h" * 32, "client_upload_id": "tok-a"}
+        body2 = {
+            **COMPLETE_BODY,
+            "storage_key": "uploads/def/IMG_2282.MOV",
+            "file_hash": "h" * 32,
+            "client_upload_id": "tok-b",
+        }
+
+        client.post("/api/v1/direct/complete", json=body1)
+        r2 = client.post("/api/v1/direct/complete", json=body2)
+
+        assert r2.json()["duplicated"] is True
+        assert len(asset_repo.created) == 1
+        assert ingest_repo.created_count == 1
+
+    def test_fallback_dedup_when_no_hash_no_token(self):
+        """旧客户端不传 hash/token：近期同库同名 processing 占位 → 兜底幂等。
+
+        模拟 complete 超时重试：第一次已建好占位，第二次（OSS 重传拿到新 key）
+        不应再建第二条。
+        """
+        client, asset_repo, ingest_repo, _ = _client()
+        # 第一次 complete（旧客户端无 token/hash）
+        r1 = client.post("/api/v1/direct/complete", json=COMPLETE_BODY)
+        assert r1.json()["duplicated"] is False
+        # 重试：重新 prepare 产生新 storage_key（仅 uuid 目录不同，文件名一致——
+        # 前端重试传的是同一个 File），且近期
+        r2 = client.post(
+            "/api/v1/direct/complete",
+            json={**COMPLETE_BODY, "storage_key": "uploads/retry/IMG_2282.MOV", "file_size": 0},
+        )
+        assert r2.status_code == 200
+        assert r2.json()["duplicated"] is True
+        assert r2.json()["asset_id"] == r1.json()["asset_id"]
+        assert len(asset_repo.created) == 1
+        assert ingest_repo.created_count == 1
+
+    def test_fallback_dedup_ignores_ready_history(self):
+        """READY 历史同名素材不触发兜底（允许用户再次上传同名文件）。"""
+        ready = Asset(
+            id="ready-1",
+            project_id="proj-1",
+            library_id="lib-1",
+            name="IMG_2282.MOV",
+            storage_key="uploads/old/IMG_2282.MOV",
+            mime_type="video/quicktime",
+            status=AssetStatus.READY,
+        )
+        client, asset_repo, ingest_repo, _ = _client(asset_repo=StubAssetRepository([ready]))
+        r = client.post("/api/v1/direct/complete", json=COMPLETE_BODY)
+        assert r.status_code == 200
+        assert r.json()["duplicated"] is False
+        assert len(asset_repo.created) == 1
+
+    def test_fallback_dedup_window_expired(self):
+        """占位记录超过 30 分钟 → 不再兜底（视为孤儿，正常新建）。"""
+        stale = Asset(
+            id="stale-1",
+            project_id="proj-1",
+            library_id="lib-1",
+            name="IMG_2282.MOV",
+            storage_key="uploads/stale/IMG_2282.MOV",
+            mime_type="video/quicktime",
+            status=AssetStatus.PROCESSING,
+        )
+        stale.created_at = datetime.now(timezone.utc) - timedelta(minutes=45)
+        client, asset_repo, ingest_repo, _ = _client(asset_repo=StubAssetRepository([stale]))
+        r = client.post("/api/v1/direct/complete", json=COMPLETE_BODY)
+        assert r.status_code == 200
+        assert r.json()["duplicated"] is False
+        assert len(asset_repo.created) == 1
+
+    def test_legacy_repo_without_new_methods_still_works(self):
+        """旧仓储没有新幂等方法 → 鸭子类型降级，不报错、正常创建。"""
+        client, asset_repo, ingest_repo, _ = _client(asset_repo=LegacyStubAssetRepository())
+        r = client.post(
+            "/api/v1/direct/complete",
+            json={**COMPLETE_BODY, "client_upload_id": "tok-x", "file_hash": "f" * 32},
+        )
+        assert r.status_code == 200
+        assert r.json()["duplicated"] is False
+        assert len(asset_repo.created) == 1
+
+    def test_duplicate_complete_returns_existing_even_if_oss_missing(self):
+        """重复 complete 幂等检查先于 OSS file_exists：
+
+        第一次成功建占位后，重试时即使 OSS 对象已不存在（file_exists=False），
+        也必须返回已存在记录而不是 404/重复建库。"""
+        client, _, _, storage = _client()
+        body = {**COMPLETE_BODY, "client_upload_id": "tok-oss-gone"}
+        r1 = client.post("/api/v1/direct/complete", json=body)
+        assert r1.status_code == 200
+
+        storage.file_exists = MagicMock(return_value=False)
+        r2 = client.post(
+            "/api/v1/direct/complete",
+            json={**body, "storage_key": "uploads/retry2/IMG_2282.MOV"},
+        )
+        assert r2.status_code == 200
+        assert r2.json()["duplicated"] is True
+        assert r2.json()["asset_id"] == r1.json()["asset_id"]
+
+
+class TestMultipartUploadIdempotency:
+    def test_same_client_upload_id_second_submit_deduplicated(self):
+        """multipart 重复提交同 token：第二次直接 duplicated，不再上传 OSS。"""
+        client, asset_repo, ingest_repo, storage = _client()
+
+        def _post():
+            return client.post(
+                "/api/v1",
+                data={"project_id": "proj-1", "library_id": "lib-1", "client_upload_id": "mp-tok-1"},
+                files={"file": ("IMG_2282.MOV", b"fake-mov-data", "video/quicktime")},
+            )
+
+        r1 = _post()
+        r2 = _post()
+        assert r1.json()["duplicated"] is False
+        assert r2.json()["duplicated"] is True
+        assert r2.json()["asset_id"] == r1.json()["asset_id"]
+        assert len(asset_repo.created) == 1
+        assert ingest_repo.created_count == 1
+        # OSS 上传只发生一次（第二次在幂等检查处直接返回）
+        assert storage.upload_file.call_count == 1