style: auto-format with black + isort + prettier [skip ci-format-check]

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

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

@@ -108,9 +108,8 @@ def _infer_mime_type_from_storage_key(storage_key: str) -> str:
     return "video/mp4"  # default
 
 
-# 兜底去重：无 file_hash / client_upload_id 时，同库同名近期活动记录视为重复
+# 兜底去重：无 file_hash / client_upload_id 且大小已知时，同库同名同大小近期活动记录视为重复
 FALLBACK_DEDUP_WINDOW_MINUTES = 30
-ACTIVE_ASSET_STATUSES = (AssetStatus.UPLOADING, AssetStatus.PROCESSING)
 
 
 def _find_duplicate_asset(
@@ -126,8 +125,12 @@ def _find_duplicate_asset(
 
     1. client_upload_id（客户端幂等 token，同一次上传的重试保持一致）
     2. file_hash（内容哈希，不同上传只要内容相同即去重）
-    3. 兜底：同库 + 同文件名（+同大小）且 30 分钟内仍处 uploading/processing
-       的记录——旧客户端不传 hash/token 时，防止 complete 超时重试反复建占位。
+    3. 兜底（严格模式，宁可漏判不可误杀）：file_hash 与 client_upload_id
+       均缺失、且 file_size > 0 时，同库 + 同文件名 + **同大小** 且 30 分钟内
+       仍处 uploading/processing 的记录才判重。
+       - file_hash 非空时跳过兜底（hash 已代表内容；同名但内容全新的视频
+         如 iPhone 的 IMG_xxxx.MOV 绝不能被同名占位误杀）
+       - file_size=0（未知）时不允许仅凭同名 + processing 判重，直接放行
 
     全部为鸭子类型调用：旧仓储无对应方法时静默跳过，不破坏既有实现。
     """
@@ -156,53 +159,35 @@ def _find_duplicate_asset(
                 existing.id,
             )
             return existing
-    if filename:
+    # 同名兜底去重（最后防线，严格模式）：
+    # - 仅当 file_hash / client_upload_id 均缺失时启用（hash 能代表内容时不靠同名猜）
+    # - file_size 必须 > 0 且与记录大小严格一致；大小未知（0）直接放行
+    # - 只命中近期 UPLOADING/PROCESSING 活动记录（READY 历史素材不拦）
+    if filename and not file_hash and not client_upload_id and file_size and file_size > 0:
         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,
+                file_size=file_size,
             )
-            # 兜底去重：按状态区分处理
-            # - READY/ERROR：稳定素材，总命中（避免重复创建）
-            # - PROCESSING/UPLOADING：预建或 complete 占位，仅当 hash 一致才命中
-            #   - 占位无 hash（旧客户端 complete 建的）→ 命中
-            #   - 占位有 hash 且与当前请求 hash 一致 → 命中
-            #   - 占位有 hash 且与当前请求 hash 不同 → 跳过（内容不同）
             if existing is not None:
-                status = getattr(existing, "status", None)
-                existing_hash = getattr(existing, "file_hash", "") or ""
-                if status in (AssetStatus.READY, AssetStatus.ERROR):
-                    logger.info(
-                        "素材幂等兜底命中(近期同名稳定记录): library=%s name=%s asset=%s status=%s",
-                        library_id,
-                        filename,
-                        getattr(existing, "id", "?"),
-                        status,
-                    )
-                    return existing
-                elif status in ACTIVE_ASSET_STATUSES:
-                    if existing_hash and file_hash and existing_hash != file_hash:
-                        logger.debug(
-                            "素材兜底去重跳过(占位 hash 不同): library=%s name=%s asset=%s hash=%s req_hash=%s",
-                            library_id,
-                            filename,
-                            getattr(existing, "id", "?"),
-                            existing_hash,
-                            file_hash,
-                        )
-                        existing = None
-                    else:
-                        logger.info(
-                            "素材幂等兜底命中(近期同名活动记录): library=%s name=%s asset=%s status=%s",
-                            library_id,
-                            filename,
-                            getattr(existing, "id", "?"),
-                            status,
-                        )
-                        return existing
+                logger.info(
+                    "素材幂等兜底命中(近期同名同大小活动记录): library=%s name=%s asset=%s status=%s size=%s",
+                    library_id,
+                    filename,
+                    getattr(existing, "id", "?"),
+                    getattr(existing, "status", None),
+                    file_size,
+                )
+                return existing
+    elif filename and not file_hash and not client_upload_id and not file_size:
+        logger.debug(
+            "同名兜底去重跳过(file_size 未知，宁可放行不可误杀): library=%s name=%s",
+            library_id,
+            filename,
+        )
     return None
 
 
@@ -487,6 +472,7 @@ async def complete_direct_upload(
         user_id=authenticated_user.user.id,
         file_hash=request.file_hash,
         client_upload_id=request.client_upload_id,
+        file_size=request.file_size,
     )
 
     job = _submit_ingest_job(

apps/web/src/components/common/ChunkErrorBoundary.tsx

@@ -0,0 +1,85 @@
+/**
+ * 全局错误边界：专门兜底"发版后旧标签页懒加载 chunk 失效"导致的白屏，
+ * 同时兜住页面级渲染崩溃，避免任何未捕获错误导致整页白屏无反馈。
+ *
+ * 捕获到 ChunkLoadError / Failed to fetch dynamically imported module：
+ *  1. 首次：自动整页刷新一次（sessionStorage 标记，刷新后 index.html 重新拉取，
+ *     拿到新 chunk 引用，白屏自愈）
+ *  2. 刷新后仍失败（标记未过期）：不再自动刷新，显示"系统已更新，请点击刷新"
+ *     兜底界面，由用户手动点击
+ *
+ * 其他非 chunk 错误：显示通用错误页 + "返回首页"按钮（跳首页而非刷新当前 URL，
+ * 避免刷新后再次命中同一路由崩溃形成死循环）。
+ */
+import React from "react"
+import { Button, Result } from "antd"
+import {
+  getChunkReloadedAt,
+  goHomeRecover,
+  isChunkLoadError,
+  reloadForChunkError,
+} from "@/utils/chunkLoadError"
+
+interface Props {
+  children: React.ReactNode
+}
+
+interface State {
+  error: Error | null
+  isChunkError: boolean
+  /** 捕获错误时是否已经自动刷新过（决定显示自动刷新中还是手动兜底） */
+  alreadyReloaded: boolean
+}
+
+class ChunkErrorBoundary extends React.Component<Props, State> {
+  state: State = { error: null, isChunkError: false, alreadyReloaded: false }
+
+  static getDerivedStateFromError(error: Error): State {
+    const chunk = isChunkLoadError(error)
+    return {
+      error,
+      isChunkError: chunk,
+      alreadyReloaded: chunk ? getChunkReloadedAt() !== null : false,
+    }
+  }
+
+  componentDidCatch(error: Error): void {
+    // 仅 chunk 错误且本次会话没自动刷新过 → 打标记并整页刷新（自愈）
+    if (isChunkLoadError(error) && getChunkReloadedAt() === null) {
+      reloadForChunkError()
+    }
+  }
+
+  render(): React.ReactNode {
+    const { error, isChunkError, alreadyReloaded } = this.state
+    if (!error) return this.props.children
+
+    if (isChunkError && !alreadyReloaded) {
+      // 已打标记、componentDidCatch 里已触发 reload；极短瞬间展示加载中
+      return (
+        <Result status="info" title="系统正在更新" subTitle="检测到新版本，正在自动刷新页面…" />
+      )
+    }
+
+    // 手动兜底统一跳首页（整页导航）：chunk 失效时脱离旧 chunk 引用；
+    // 业务崩溃时绕开当前报错路由，避免刷新-再崩死循环
+    return (
+      <Result
+        status="warning"
+        title={isChunkError ? "系统已更新" : "页面出现异常"}
+        subTitle={
+          isChunkError
+            ? "检测到新版本，请点击下方按钮回到首页加载最新内容。"
+            : "页面加载遇到问题，点击返回首页通常可以恢复，未保存的内容可能丢失。"
+        }
+        extra={
+          <Button type="primary" onClick={goHomeRecover}>
+            {isChunkError ? "刷新并返回首页" : "返回首页"}
+          </Button>
+        }
+      />
+    )
+  }
+}
+
+export default ChunkErrorBoundary

apps/web/src/main.tsx

@@ -9,6 +9,7 @@ import { QueryClient, QueryClientProvider } from "@tanstack/react-query"
 import { ConfigProvider, App as AntApp } from "antd"
 import zhCN from "antd/locale/zh_CN"
 import router from "./router"
+import ChunkErrorBoundary from "./components/common/ChunkErrorBoundary"
 import { scheduleProactiveRefresh } from "./api/auth/tokenRefresh"
 
 // 应用启动时，如果用户已登录，立即调度主动 token 刷新
@@ -99,7 +100,9 @@ ReactDOM.createRoot(document.getElementById("root")!).render(
     <QueryClientProvider client={queryClient}>
       <ConfigProvider locale={zhCN} theme={theme}>
         <AntApp>
-          <RouterProvider router={router} />
+          <ChunkErrorBoundary>
+            <RouterProvider router={router} />
+          </ChunkErrorBoundary>
         </AntApp>
       </ConfigProvider>
     </QueryClientProvider>

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

@@ -9,12 +9,13 @@
  * - 标题样式（字体/颜色/位置/大小/粗斜描边/预设）：全局统一
  */
 import React, { useMemo, useState } from "react"
-import { AutoComplete, Input, message } from "antd"
+import { Input, message } from "antd"
 import { LoadingOutlined } from "@ant-design/icons"
 import type { TitleSettings } from "../types"
 import { POSITION_OPTIONS, FONT_OPTIONS } from "../constants"
 import { useStep4Title } from "../hooks/useStep4Title"
 import AiTitleGenerator from "./title/AiTitleGenerator"
+import TitleLibraryAutoComplete from "./title/TitleLibraryAutoComplete"
 import TitleStylePanel from "./title/TitleStylePanel"
 import { AI_TITLE_TEMPLATES } from "../constants"
 
@@ -201,21 +202,14 @@ const Step4TitleSettings: React.FC<Step4TitleSettingsProps> = (props) => {
               </div>
               <div className="xx-form-field">
                 <label>标题</label>
-                <AutoComplete
-                  placeholder="输入标题文字…"
-                  allowClear
-                  maxLength={50}
-                  style={{ width: "100%" }}
-                  value={(previewTitles?.[0] ?? t.titleSettings.title) || undefined}
+                <TitleLibraryAutoComplete
+                  placeholder="输入或从标题库选择"
+                  value={previewTitles?.[0] ?? t.titleSettings.title}
                   onChange={(val) => {
                     t.updateTitle(val || "")
                     onPreviewTitlesChange?.([val || ""])
                   }}
                   options={titleOptions}
-                  filterOption={(inputValue, option) => {
-                    const title = (option?.label || option?.value || "") as string
-                    return title.toLowerCase().includes((inputValue || "").toLowerCase())
-                  }}
                 />
               </div>
             </>
@@ -269,17 +263,11 @@ const Step4TitleSettings: React.FC<Step4TitleSettingsProps> = (props) => {
           {Array.from({ length: previewCount }, (_, i) => (
             <div className="xx-form-field" key={i}>
               <label>视频 {i + 1} 标题</label>
-              <AutoComplete
-                placeholder={`视频 ${i + 1} 的标题…`}
-                maxLength={50}
-                style={{ width: "100%" }}
-                value={previewTitles?.[i] || undefined}
-                onChange={(val) => updateVariantTitle(i, val || "")}
+              <TitleLibraryAutoComplete
+                placeholder={`输入或选择视频 ${i + 1} 的标题`}
+                value={previewTitles?.[i] || ""}
+                onChange={(val) => updateVariantTitle(i, val)}
                 options={titleOptions}
-                filterOption={(inputValue, option) => {
-                  const title = (option?.label || option?.value || "") as string
-                  return title.toLowerCase().includes((inputValue || "").toLowerCase())
-                }}
               />
             </div>
           ))}

apps/web/src/pages/generate/components/title/TitleLibraryAutoComplete.tsx

@@ -0,0 +1,75 @@
+/**
+ * 标题库 AutoComplete（Issue #1737）
+ *
+ * 原生 antd AutoComplete（combobox 模式）的两个行为不符合产品预期：
+ * 1. combobox 默认 showAction=[]，输入框聚焦时下拉不展开——用户必须先打字才能看到标题库，
+ *    且组件无下拉箭头，视觉上是"纯输入框"，不知道标题库里已有标题可选。
+ * 2. 空态聚焦不展示任何标题库内容。
+ *
+ * 本组件封装修复：
+ * - 受控 open：聚焦（且标题库非空）即展开，展示全部标题；失焦/选中/Esc 关闭
+ *   （rc-select 失焦会主动 onToggleOpen(false)，onOpenChange 同步状态即可，不会死循环）
+ * - suffixIcon 加下拉三角，视觉提示"可选择"；有值时 allowClear 的清除按钮照常出现
+ * - 输入文字时由 filterOption 过滤（空串展示全部）
+ * - 保留 combobox 自由输入能力：用户可输入标题库之外的自定义标题
+ */
+import React, { useState } from "react"
+import { AutoComplete } from "antd"
+import { DownOutlined } from "@ant-design/icons"
+import type { AutoCompleteProps } from "antd"
+
+export interface TitleOption {
+  label: string
+  value: string
+}
+
+interface TitleLibraryAutoCompleteProps {
+  value: string
+  onChange: (val: string) => void
+  options: TitleOption[]
+  placeholder?: string
+  allowClear?: boolean
+  maxLength?: number
+  style?: React.CSSProperties
+}
+
+const TitleLibraryAutoComplete: React.FC<TitleLibraryAutoCompleteProps> = ({
+  value,
+  onChange,
+  options,
+  placeholder = "输入或从标题库选择",
+  allowClear = true,
+  maxLength = 50,
+  style,
+}) => {
+  const [open, setOpen] = useState(false)
+  const hasTitles = options.length > 0
+
+  const filterOption: AutoCompleteProps["filterOption"] = (inputValue, option) => {
+    const title = (option?.label || option?.value || "") as string
+    return title.toLowerCase().includes((inputValue || "").toLowerCase())
+  }
+
+  return (
+    <AutoComplete
+      value={value || undefined}
+      onChange={(val) => onChange(val || "")}
+      options={options}
+      filterOption={filterOption}
+      open={open}
+      onOpenChange={setOpen}
+      onFocus={() => {
+        // 标题库为空时不展开（避免弹出"暂无数据"空壳）
+        if (hasTitles) setOpen(true)
+      }}
+      onSelect={() => setOpen(false)}
+      suffixIcon={<DownOutlined style={{ color: "var(--text-secondary, #bbb)", fontSize: 12 }} />}
+      placeholder={placeholder}
+      allowClear={allowClear}
+      maxLength={maxLength}
+      style={{ width: "100%", ...style }}
+    />
+  )
+}
+
+export default TitleLibraryAutoComplete

apps/web/src/router/appRoutes.tsx

@@ -1,6 +1,7 @@
 import { Navigate, type RouteObject } from "react-router-dom"
 import MainLayout from "@/components/layout/MainLayout"
 import { ProtectedRoute } from "./ProtectedRoute"
+import { lazyRoute } from "./lazyRoute"
 
 /**
  * 受保护的 /app 子路由
@@ -13,202 +14,118 @@ const appChildren: RouteObject[] = [
   },
   {
     path: "dashboard",
-    lazy: () =>
-      import("@/pages/dashboard/Dashboard").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/dashboard/Dashboard")),
   },
   {
     path: "assets",
-    lazy: () =>
-      import("@/pages/assets/AssetLibrary").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/assets/AssetLibrary")),
   },
   {
     path: "titles",
-    lazy: () =>
-      import("@/pages/titles/TitleLibrary").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/titles/TitleLibrary")),
   },
   {
     path: "voices",
-    lazy: () =>
-      import("@/pages/voices/VoiceLibrary").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/voices/VoiceLibrary")),
   },
   {
     path: "templates",
-    lazy: () =>
-      import("@/pages/templates/TemplateLibrary").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/templates/TemplateLibrary")),
   },
   {
     path: "generate",
-    lazy: () =>
-      import("@/pages/generate/GeneratePage").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/generate/GeneratePage")),
   },
   {
     path: "history",
-    lazy: () =>
-      import("@/pages/history/TaskHistory").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/history/TaskHistory")),
   },
   {
     path: "products",
-    lazy: () =>
-      import("@/pages/products/ProductLibrary").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/products/ProductLibrary")),
   },
   {
     path: "products/:id",
-    lazy: () =>
-      import("@/pages/products/ProductDetail").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/products/ProductDetail")),
   },
   {
     path: "tasks",
-    lazy: () =>
-      import("@/pages/tasks/TaskCenter").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/tasks/TaskCenter")),
   },
   {
     path: "editing-planner",
-    lazy: () =>
-      import("@/pages/editing-planner/EditingPlanner").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/editing-planner/EditingPlanner")),
   },
   {
     path: "my-templates",
-    lazy: () =>
-      import("@/pages/my-templates/MyTemplates").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/my-templates/MyTemplates")),
   },
   {
     path: "voice-clone",
-    lazy: () =>
-      import("@/pages/voice-clone/VoiceClone").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/voice-clone/VoiceClone")),
   },
   {
     path: "voice-materials",
-    lazy: () =>
-      import("@/pages/voice-materials/VoiceMaterialLibrary").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/voice-materials/VoiceMaterialLibrary")),
   },
   {
     path: "my-voices",
-    lazy: () =>
-      import("@/pages/my-voices/MyVoices").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/my-voices/MyVoices")),
   },
   {
     path: "accounts",
-    lazy: () =>
-      import("@/pages/accounts/Accounts").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/accounts/Accounts")),
   },
   {
     path: "duplication",
-    lazy: () =>
-      import("@/pages/duplication/DuplicationUpload").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/duplication/DuplicationUpload")),
   },
   {
     path: "duplication/results",
-    lazy: () =>
-      import("@/pages/duplication/DuplicationResults").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/duplication/DuplicationResults")),
   },
   {
     path: "duplication/:id",
-    lazy: () =>
-      import("@/pages/duplication/DuplicationDetail").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/duplication/DuplicationDetail")),
   },
   {
     path: "subscription",
-    lazy: () =>
-      import("@/pages/subscription/Plans").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/subscription/Plans")),
   },
   {
     path: "subscription/upgrade",
-    lazy: () =>
-      import("@/pages/subscription/UpgradeSubscription").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/subscription/UpgradeSubscription")),
   },
   {
     path: "subscription/billing",
-    lazy: () =>
-      import("@/pages/subscription/Billing").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/subscription/Billing")),
   },
   {
     path: "profile",
-    lazy: () =>
-      import("@/pages/profile/Settings").then((m) => ({
-        Component: m.default,
-      })),
+    lazy: lazyRoute(() => import("@/pages/profile/Settings")),
   },
   {
     path: "admin",
     children: [
       {
         index: true,
-        lazy: () =>
-          import("@/pages/admin/AdminComingSoon").then((m) => ({
-            Component: m.default,
-          })),
+        lazy: lazyRoute(() => import("@/pages/admin/AdminComingSoon")),
       },
       {
         path: "users",
-        lazy: () =>
-          import("@/pages/admin/AdminComingSoon").then((m) => ({
-            Component: m.default,
-          })),
+        lazy: lazyRoute(() => import("@/pages/admin/AdminComingSoon")),
       },
       {
         path: "analytics",
-        lazy: () =>
-          import("@/pages/admin/AdminComingSoon").then((m) => ({
-            Component: m.default,
-          })),
+        lazy: lazyRoute(() => import("@/pages/admin/AdminComingSoon")),
       },
       {
         path: "monitor",
-        lazy: () =>
-          import("@/pages/admin/AdminComingSoon").then((m) => ({
-            Component: m.default,
-          })),
+        lazy: lazyRoute(() => import("@/pages/admin/AdminComingSoon")),
       },
       {
         path: "logs",
-        lazy: () =>
-          import("@/pages/admin/AdminComingSoon").then((m) => ({
-            Component: m.default,
-          })),
+        lazy: lazyRoute(() => import("@/pages/admin/AdminComingSoon")),
       },
     ],
   },

apps/web/src/router/lazyRoute.ts

@@ -0,0 +1,40 @@
+import type { LazyRouteFunction, RouteObject } from "react-router-dom"
+import { isChunkLoadError } from "@/utils/chunkLoadError"
+
+/**
+ * 给 React Router data router 的路由懒加载包一层自动重试：
+ *
+ * - 网络抖动 / 瞬态失败：自动重试最多 2 次（间隔 300ms / 800ms），用户无感恢复
+ * - 发版后旧 chunk 404（chunk 文件名已不存在）：重试也拿不到旧文件名，
+ *   重试耗尽后抛出，由全局 ChunkErrorBoundary 捕获并引导整页刷新
+ *   （刷新后 index.html 是 no-cache 的，会拿到新 chunk 引用）
+ */
+const RETRY_DELAYS_MS = [300, 800]
+const RETRY_COUNT = RETRY_DELAYS_MS.length
+
+const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms))
+
+export const lazyRoute = (
+  factory: () => Promise<{ default: React.ComponentType }>,
+): LazyRouteFunction<RouteObject> => {
+  return async () => {
+    let lastError: unknown
+    for (let attempt = 0; attempt <= RETRY_COUNT; attempt++) {
+      try {
+        const mod = await factory()
+        if (!mod.default) {
+          throw new Error("lazyRoute: 目标模块缺少 default 导出")
+        }
+        return { Component: mod.default }
+      } catch (err) {
+        lastError = err
+        // 非 chunk 加载错误（代码 bug 等）立即抛出，不浪费重试
+        if (!isChunkLoadError(err)) throw err
+        if (attempt < RETRY_COUNT) {
+          await sleep(RETRY_DELAYS_MS[attempt])
+        }
+      }
+    }
+    throw lastError
+  }
+}

apps/web/src/test/components/ChunkErrorBoundary.test.tsx

@@ -0,0 +1,79 @@
+import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"
+import { render, screen, fireEvent } from "@testing-library/react"
+import { Button } from "antd"
+import { useState } from "react"
+import ChunkErrorBoundary from "@/components/common/ChunkErrorBoundary"
+import * as chunkUtils from "@/utils/chunkLoadError"
+
+// reload 函数 mock 掉（jsdom 不支持真实 window.location.reload）
+vi.mock("@/utils/chunkLoadError", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("@/utils/chunkLoadError")>()
+  return {
+    ...actual,
+    reloadForChunkError: vi.fn(),
+    goHomeRecover: vi.fn(),
+  }
+})
+const { reloadForChunkError, goHomeRecover } = vi.mocked(chunkUtils)
+
+/** 渲染时直接抛错的子组件 */
+const Boom: React.FC<{ error: Error }> = ({ error }) => {
+  throw error
+}
+
+/** 点击按钮后才抛 chunk 错误的子组件 */
+const ChunkBoomButton: React.FC = () => {
+  const [boom, setBoom] = useState(false)
+  if (boom) {
+    throw new TypeError("Failed to fetch dynamically imported module: /assets/x.js")
+  }
+  return <Button onClick={() => setBoom(true)}>boom</Button>
+}
+
+const renderBoundary = (ui: React.ReactNode) =>
+  render(<ChunkErrorBoundary>{ui}</ChunkErrorBoundary>)
+
+beforeEach(() => {
+  sessionStorage.clear()
+  vi.clearAllMocks()
+  // error boundary 捕获后 React 会打 error log，静默掉
+  vi.spyOn(console, "error").mockImplementation(() => {})
+})
+
+afterEach(() => {
+  vi.restoreAllMocks()
+  sessionStorage.clear()
+})
+
+describe("ChunkErrorBoundary", () => {
+  it("正常渲染 children", () => {
+    renderBoundary(<div>hello-child</div>)
+    expect(screen.getByText("hello-child")).toBeInTheDocument()
+  })
+
+  it("首次捕获 chunk 错误 → 自动刷新（reloadForChunkError）并显示自动刷新提示", () => {
+    renderBoundary(<ChunkBoomButton />)
+    fireEvent.click(screen.getByText("boom"))
+    expect(reloadForChunkError).toHaveBeenCalledTimes(1)
+    expect(screen.getByText(/正在自动刷新/)).toBeInTheDocument()
+  })
+
+  it("已刷新过仍失败 → 不再自动刷新，显示手动兜底按钮", () => {
+    // 模拟"本会话已经自动刷新过一次"
+    sessionStorage.setItem("chunk_error_reloaded_at", String(Date.now()))
+    renderBoundary(
+      <Boom error={new TypeError("Failed to fetch dynamically imported module: /assets/y.js")} />,
+    )
+    expect(reloadForChunkError).not.toHaveBeenCalled()
+    expect(screen.getByText("系统已更新")).toBeInTheDocument()
+    // 点击兜底按钮 → goHomeRecover（跳首页，不刷新当前 URL）
+    fireEvent.click(screen.getByText("刷新并返回首页"))
+    expect(goHomeRecover).toHaveBeenCalledTimes(1)
+  })
+
+  it("非 chunk 错误 → 显示通用错误页，不触发 chunk 自动刷新", () => {
+    renderBoundary(<Boom error={new Error("普通业务报错")} />)
+    expect(reloadForChunkError).not.toHaveBeenCalled()
+    expect(screen.getByText("页面出现异常")).toBeInTheDocument()
+  })
+})

apps/web/src/test/pages/generate/title-library-autocomplete.test.tsx

@@ -0,0 +1,126 @@
+/**
+ * TitleLibraryAutoComplete 单测（Issue #1737）
+ *
+ * 覆盖：
+ * - 聚焦空输入框 → 下拉立即展开，展示标题库全部标题（原生 AutoComplete 聚焦不展开，此为本工单核心修复）
+ * - 输入关键词 → 下拉只显示匹配项
+ * - 点击下拉项 → onChange 回填所选标题
+ * - 自由输入自定义标题 → onChange 正常透传，不被下拉干扰
+ * - 标题库为空 → 聚焦不展开（不出"暂无数据"空壳）
+ * - 选中后下拉关闭
+ */
+import { describe, it, expect, vi } from "vitest"
+import { render, screen, waitFor, fireEvent } from "@testing-library/react"
+import userEvent from "@testing-library/user-event"
+import TitleLibraryAutoComplete from "@/pages/generate/components/title/TitleLibraryAutoComplete"
+
+const OPTIONS = [
+  { label: "永康这家面馆绝了", value: "永康这家面馆绝了" },
+  { label: "永康美食探店vlog", value: "永康美食探店vlog" },
+  { label: "萌宠日常第一天", value: "萌宠日常第一天" },
+]
+
+function renderBox(initialValue = "", opts = OPTIONS) {
+  const onChange = vi.fn()
+  const result = render(
+    <TitleLibraryAutoComplete
+      value={initialValue}
+      onChange={onChange}
+      options={opts}
+      placeholder="输入或从标题库选择"
+    />,
+  )
+  return { onChange, ...result }
+}
+
+/** 聚焦输入框（combobox role） */
+function focusInput() {
+  const input = screen.getByRole("combobox") as HTMLInputElement
+  fireEvent.focus(input)
+  return input
+}
+
+/** 取下拉中实际可见的选项（rc-virtual-list 渲染为 .ant-select-item-option；role=option 的 listbox 是 a11y 哨兵） */
+function getVisibleOptions(): HTMLElement[] {
+  const dropdown = document.querySelector(".ant-select-dropdown:not(.ant-select-dropdown-hidden)")
+  if (!dropdown) return []
+  return Array.from(dropdown.querySelectorAll(".ant-select-item-option")) as HTMLElement[]
+}
+
+describe("TitleLibraryAutoComplete (#1737)", () => {
+  it("聚焦空输入框时下拉展开并展示标题库全部标题", async () => {
+    renderBox()
+    expect(screen.queryByRole("listbox")).not.toBeInTheDocument()
+
+    focusInput()
+
+    await screen.findByRole("listbox")
+    await waitFor(() => expect(getVisibleOptions()).toHaveLength(3))
+    const options = getVisibleOptions()
+    expect(options[0]).toHaveTextContent("永康这家面馆绝了")
+    expect(options[2]).toHaveTextContent("萌宠日常第一天")
+  })
+
+  it("输入关键词时下拉只显示匹配项", async () => {
+    const user = userEvent.setup()
+    renderBox()
+    const input = screen.getByRole("combobox")
+    await user.click(input)
+    await screen.findByRole("listbox")
+
+    await user.type(input, "永康")
+    await waitFor(() => expect(getVisibleOptions()).toHaveLength(2))
+    const options = getVisibleOptions()
+    expect(options.every((o) => o.textContent?.includes("永康"))).toBe(true)
+  })
+
+  it("点击下拉项后 onChange 回填标题且下拉关闭", async () => {
+    const user = userEvent.setup()
+    const { onChange } = renderBox()
+    const input = screen.getByRole("combobox") as HTMLInputElement
+    await user.click(input)
+    await screen.findByRole("listbox")
+
+    await user.click(screen.getByText("萌宠日常第一天"))
+
+    await waitFor(() => {
+      expect(onChange).toHaveBeenCalledWith("萌宠日常第一天")
+    })
+    await waitFor(() => {
+      expect(screen.queryByRole("listbox")).not.toBeInTheDocument()
+    })
+  })
+
+  it("自由输入自定义标题时 onChange 正常透传（不被下拉干扰）", async () => {
+    const user = userEvent.setup()
+    const { onChange } = renderBox()
+    const input = screen.getByRole("combobox")
+    await user.click(input)
+
+    await user.type(input, "我自己编的标题XYZ")
+    await waitFor(() => {
+      expect(onChange).toHaveBeenCalledWith("我自己编的标题XYZ")
+    })
+    // 输入无匹配关键词，下拉无 option 时不阻塞输入
+    expect(input).toHaveValue("我自己编的标题XYZ")
+  })
+
+  it("标题库为空时聚焦不展开下拉", async () => {
+    renderBox("", [])
+    focusInput()
+    // 等一帧确认没有 listbox
+    await new Promise((r) => setTimeout(r, 50))
+    expect(screen.queryByRole("listbox")).not.toBeInTheDocument()
+  })
+
+  it("渲染下拉箭头图标作为可选择提示", () => {
+    const { container } = renderBox()
+    // antd 后缀图标在 .ant-select-arrow 内
+    expect(container.querySelector(".ant-select-arrow")).toBeInTheDocument()
+  })
+
+  it("有初始值时输入框正常展示", () => {
+    renderBox("已有标题")
+    expect(screen.getByRole("combobox")).toHaveValue("已有标题")
+  })
+})

apps/web/src/test/router/lazyRoute.test.ts

@@ -0,0 +1,44 @@
+import { describe, it, expect, vi, afterEach } from "vitest"
+import { lazyRoute } from "@/router/lazyRoute"
+
+const chunkErr = () => new TypeError("Failed to fetch dynamically imported module: /assets/x.js")
+
+/** fake 模块 */
+const Comp = function Comp() {}
+const factoryOk = vi.fn(async () => ({ default: Comp }))
+
+afterEach(() => {
+  vi.clearAllMocks()
+})
+
+describe("lazyRoute", () => {
+  it("首次成功直接返回 Component", async () => {
+    const result = await lazyRoute(factoryOk)()
+    expect(result).toEqual({ Component: Comp })
+    expect(factoryOk).toHaveBeenCalledTimes(1)
+  })
+
+  it("chunk 失败重试：前两次失败、第三次成功 → 不抛出", async () => {
+    const f = vi
+      .fn()
+      .mockRejectedValueOnce(chunkErr())
+      .mockRejectedValueOnce(chunkErr())
+      .mockResolvedValueOnce({ default: Comp })
+
+    const result = await lazyRoute(f as never)()
+    expect(result).toEqual({ Component: Comp })
+    expect(f).toHaveBeenCalledTimes(3)
+  })
+
+  it("chunk 失败重试 2 次仍失败 → 抛出", async () => {
+    const f = vi.fn().mockRejectedValue(chunkErr())
+    await expect(lazyRoute(f as never)()).rejects.toThrow(/dynamically imported/)
+    expect(f).toHaveBeenCalledTimes(3)
+  })
+
+  it("非 chunk 错误立即抛出，不重试", async () => {
+    const f = vi.fn().mockRejectedValue(new Error("业务模块内部报错"))
+    await expect(lazyRoute(f as never)()).rejects.toThrow("业务模块内部报错")
+    expect(f).toHaveBeenCalledTimes(1)
+  })
+})

apps/web/src/test/utils/chunkLoadError.test.ts

@@ -0,0 +1,80 @@
+import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"
+import {
+  getChunkReloadedAt,
+  goHomeRecover,
+  isChunkLoadError,
+  reloadForChunkError,
+} from "@/utils/chunkLoadError"
+
+describe("isChunkLoadError", () => {
+  it("识别 Vite 动态 import 失败", () => {
+    const err = new TypeError(
+      "Failed to fetch dynamically imported module: https://x/assets/AssetLibrary-abc.js",
+    )
+    expect(isChunkLoadError(err)).toBe(true)
+  })
+
+  it("识别 Webpack 风格 ChunkLoadError", () => {
+    const err = new Error("Loading chunk 12 failed.")
+    err.name = "ChunkLoadError"
+    expect(isChunkLoadError(err)).toBe(true)
+  })
+
+  it("识别字符串形式错误", () => {
+    expect(isChunkLoadError("Error loading dynamically imported module")).toBe(true)
+  })
+
+  it("普通错误不命中", () => {
+    expect(isChunkLoadError(new Error("Cannot read properties of undefined"))).toBe(false)
+    expect(isChunkLoadError(null)).toBe(false)
+    expect(isChunkLoadError(undefined)).toBe(false)
+    expect(isChunkLoadError({ status: 500 })).toBe(false)
+  })
+})
+
+describe("reload 标记", () => {
+  beforeEach(() => {
+    sessionStorage.clear()
+    // jsdom 未实现真实导航，reload 仅打 "not implemented" 警告，静默掉
+    vi.spyOn(console, "error").mockImplementation(() => {})
+  })
+  afterEach(() => {
+    vi.restoreAllMocks()
+    sessionStorage.clear()
+  })
+
+  it("无标记返回 null", () => {
+    expect(getChunkReloadedAt()).toBeNull()
+  })
+
+  it("reloadForChunkError 写入刷新标记", () => {
+    expect(() => reloadForChunkError()).not.toThrow()
+    expect(getChunkReloadedAt()).not.toBeNull()
+  })
+
+  it("标记过期（>10min）返回 null", () => {
+    sessionStorage.setItem("chunk_error_reloaded_at", String(Date.now() - 11 * 60 * 1000))
+    expect(getChunkReloadedAt()).toBeNull()
+  })
+
+  it("goHomeRecover 清掉标记", () => {
+    reloadForChunkError()
+    expect(getChunkReloadedAt()).not.toBeNull()
+    expect(() => goHomeRecover()).not.toThrow()
+    expect(sessionStorage.getItem("chunk_error_reloaded_at")).toBeNull()
+  })
+
+  it("sessionStorage 抛异常（无痕模式）时降级不崩溃", () => {
+    const spy = vi.spyOn(Storage.prototype, "getItem").mockImplementation(() => {
+      throw new Error("Storage disabled")
+    })
+    const setSpy = vi.spyOn(Storage.prototype, "setItem").mockImplementation(() => {
+      throw new Error("Storage disabled")
+    })
+    expect(getChunkReloadedAt()).toBeNull()
+    expect(() => reloadForChunkError()).not.toThrow()
+    expect(() => goHomeRecover()).not.toThrow()
+    spy.mockRestore()
+    setSpy.mockRestore()
+  })
+})

apps/web/src/utils/chunkLoadError.ts

@@ -0,0 +1,84 @@
+/**
+ * 发版后旧标签页懒加载 chunk 失效（白屏）的识别与恢复工具。
+ *
+ * 背景：页面 React Router 的 lazy 动态 import，发版后旧 chunk 文件名被删除，
+ * 停留在旧标签页的用户点菜单时 import 404，抛出
+ * "Failed to fetch dynamically imported module"（Vite）/ ChunkLoadError，
+ * 不捕获就是整页白屏。
+ */
+
+/** sessionStorage 标记：最近已经为 chunk 失效自动刷新过一次（带时间戳，10min 有效） */
+const RELOAD_FLAG_KEY = "chunk_error_reloaded_at"
+/** 标记有效期：超过后允许再次自动刷新，避免用户手动正常刷新后标记永久残留 */
+const RELOAD_FLAG_TTL_MS = 10 * 60 * 1000
+
+/**
+ * Storage 在 Safari 无痕模式 / 禁用 Cookie 的浏览器 / 严格 iframe 策略下
+ * 访问可能抛异常；此处统一容错，拿不到存储就降级为"无标记"，绝不能让
+ * 错误边界本身因读存储而崩溃。
+ */
+const safeStorage = {
+  getItem: (key: string): string | null => {
+    try {
+      return sessionStorage.getItem(key)
+    } catch {
+      return null
+    }
+  },
+  setItem: (key: string, value: string): void => {
+    try {
+      sessionStorage.setItem(key, value)
+    } catch {
+      /* 存储不可用时静默降级：仅丢失"已刷新"标记，不影响恢复动作 */
+    }
+  },
+  removeItem: (key: string): void => {
+    try {
+      sessionStorage.removeItem(key)
+    } catch {
+      /* ignore */
+    }
+  },
+}
+
+/** 判断错误是否为懒加载 chunk 加载失败（发版 404 / 网络中断 / 动态 import 失败） */
+export const isChunkLoadError = (error: unknown): boolean => {
+  if (!error) return false
+  // Vite: Failed to fetch dynamically imported module: /assets/xxx-yyy.js
+  // Webpack: ChunkLoadError: Loading chunk xxx failed.
+  const needle =
+    error instanceof Error
+      ? `${error.name} ${error.message}`
+      : typeof error === "string"
+        ? error
+        : ""
+  return /failed to fetch dynamically imported module|chunkloaderror|loading chunk \d+ failed|error loading dynamically imported module|importing a module script failed/i.test(
+    needle,
+  )
+}
+
+/** 读取上次自动刷新时间戳；过期或不存在返回 null */
+export const getChunkReloadedAt = (): number | null => {
+  const raw = safeStorage.getItem(RELOAD_FLAG_KEY)
+  if (!raw) return null
+  const ts = Number(raw)
+  if (!Number.isFinite(ts)) return null
+  if (Date.now() - ts > RELOAD_FLAG_TTL_MS) return null
+  return ts
+}
+
+/** 标记"已为 chunk 失效自动刷新过"，然后刷新页面 */
+export const reloadForChunkError = (): void => {
+  safeStorage.setItem(RELOAD_FLAG_KEY, String(Date.now()))
+  window.location.reload()
+}
+
+/**
+ * 硬恢复：清掉标记后回到首页（整页导航，不是当前 URL 刷新）。
+ * - chunk 失效兜底：回到首页会拉取最新 index.html，彻底脱离旧 chunk 引用
+ * - 非 chunk 的页面级崩溃：跳首页能绕开当前报错路由，避免"刷新-再崩"死循环
+ */
+export const goHomeRecover = (): void => {
+  safeStorage.removeItem(RELOAD_FLAG_KEY)
+  window.location.href = "/"
+}

apps/worker/worker_app/celery_app.py

@@ -17,6 +17,12 @@ apply_queue_settings(celery_app)
 # 长渲染任务预取 1，避免任务被预取占住导致调度不均
 celery_app.conf.worker_prefetch_multiplier = GENERATION_WORKER_PREFETCH_MULTIPLIER
 celery_app.conf.task_acks_late = True  # worker 崩溃时未完成任务重回队列，由执行前守卫丢弃作废消息
+# worker 进程被 OOM/容器硬杀时拒绝 ack，消息留在队列由其他 worker 接手
+celery_app.conf.task_reject_on_worker_lost = True
+# Redis broker 消息可见性超时（#1714）：acks_late 下，消息被预取后 visibility_timeout
+# 内未 ack 才会重投。长任务（ingest HEVC 转码 20-30 分钟、生成硬超时 11 分钟）
+# 必须远大于最长执行时间，否则正常任务会在执行中被误重投；4 小时覆盖最长转码 + 余量。
+celery_app.conf.broker_transport_options = {"visibility_timeout": 4 * 60 * 60}
 
 celery_app.conf.imports = (
     "worker_app.tasks.health",
@@ -48,4 +54,11 @@ celery_app.conf.beat_schedule = {
         "schedule": 300.0,  # 每 5 分钟（秒）
         "options": {"expires": 240},
     },
+    # 上传/转码链路孤儿巡检：worker 重启丢 prefetch 消息后，卡 pending/processing
+    # 的 ingest_job + asset 占位超时标终态（#1714）。转码任务较长，10 分钟一轮
+    "cleanup-stale-ingest-jobs": {
+        "task": "worker.cleanup_stale_ingest_jobs",
+        "schedule": 600.0,  # 每 10 分钟（秒）
+        "options": {"expires": 540},
+    },
 }

apps/worker/worker_app/tasks/_startup.py

@@ -257,3 +257,32 @@ def _on_worker_ready(sender, **kwargs):  # pragma: no cover
     result = cleanup_all_stale_tasks()
     total = result["generation_tasks"] + result["jobs"]
     logger.info("Worker 启动清理完成，共清理 %d 个孤儿任务", total)
+
+
+@worker_ready.connect
+def _recover_stuck_ingest_jobs_on_ready(sender, **kwargs):  # pragma: no cover
+    """Worker 启动完成后恢复卡死在 processing 的 ingest_job（#1714）。
+
+    容器重启/进程 OOM 导致 transcode 队列 unacked 消息未重投时，processing
+    ingest_job 会永久卡死。启动时扫描 processing 超 10 分钟的 job，CAS 重置
+    pending 并重新派单；Redis 锁保证同容器 generation/transcode 双 worker
+    只有一个执行恢复。旧消息若后来重投，ingest_asset 执行前守卫会丢弃。
+    """
+    try:
+        from packages.application.ingest_orphan_cleanup import (
+            make_redis_recovery_lock,
+            recover_stuck_ingest_jobs_on_startup,
+        )
+
+        session = SessionLocal()
+        try:
+            recovered = recover_stuck_ingest_jobs_on_startup(
+                session,
+                lock_acquire=make_redis_recovery_lock(),
+                stuck_minutes=10,
+            )
+        finally:
+            session.close()
+        logger.info("Worker 启动 ingest 恢复完成，共重新派单 %d 个卡死任务", recovered)
+    except Exception as e:  # noqa: BLE001 — 启动恢复失败不能阻断 worker 起服
+        logger.error("启动 ingest 恢复扫描失败（beat 巡检仍会兜底标 failed）: %s", e, exc_info=True)

apps/worker/worker_app/tasks/cleanup.py

@@ -16,6 +16,12 @@ from worker_app.tasks._startup import (
     cleanup_stale_pending_tasks,
 )
 
+from packages.application.ingest_orphan_cleanup import (
+    ASSET_ORPHAN_TIMEOUT_MINUTES,
+    INGEST_PENDING_TIMEOUT_MINUTES,
+    INGEST_PROCESSING_TIMEOUT_MINUTES,
+)
+
 logger = logging.getLogger(__name__)
 
 
@@ -69,3 +75,53 @@ def scheduled_cleanup_stale_running(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_M
             timeout_minutes,
         )
     return {"generation_tasks": gen_count, "jobs": job_count}
+
+
+@shared_task(name="worker.cleanup_stale_ingest_jobs")
+def scheduled_cleanup_stale_ingest_jobs(
+    processing_timeout_minutes: int = INGEST_PROCESSING_TIMEOUT_MINUTES,
+    pending_timeout_minutes: int = INGEST_PENDING_TIMEOUT_MINUTES,
+    orphan_asset_timeout_minutes: int = ASSET_ORPHAN_TIMEOUT_MINUTES,
+) -> dict:
+    """Celery Beat 调度：清理上传/转码链路（IngestJob + Asset）孤儿记录。
+
+    每 10 分钟执行一次。worker 容器重启/进程 OOM 时，已 prefetch 的 transcode
+    celery 消息会丢失（队列里也不存在），ingest_job 永久卡 pending/processing、
+    asset 永久卡 processing/uploading，没有兜底永远不会恢复（#1714）。
+
+    - ingest_job processing > processing_timeout_minutes / pending > pending_timeout_minutes
+      → 标 failed；关联 asset 占位（processing/uploading）联动标 error
+    - 无 ingest_job 关联、created_at > orphan_asset_timeout_minutes 的占位 asset
+      → 标 error
+    - 作废 celery 消息 revoke + 物理清除（防重投，执行前守卫是第二道防线）
+    """
+    from worker_app.db import SessionLocal
+
+    from packages.application.ingest_orphan_cleanup import (
+        cleanup_orphan_processing_assets,
+        cleanup_stale_ingest_jobs,
+        revoke_stale_ingest_messages,
+    )
+
+    session = SessionLocal()
+    try:
+        job_items, asset_ids = cleanup_stale_ingest_jobs(
+            session,
+            processing_timeout_minutes=processing_timeout_minutes,
+            pending_timeout_minutes=pending_timeout_minutes,
+        )
+        orphan_asset_ids = cleanup_orphan_processing_assets(session, timeout_minutes=orphan_asset_timeout_minutes)
+    finally:
+        session.close()
+
+    purged = revoke_stale_ingest_messages(job_items) if job_items else 0
+    total_jobs = len(job_items)
+    total_assets = len(set(asset_ids) | set(orphan_asset_ids))
+    if total_jobs or total_assets:
+        logger.warning(
+            "[Beat] 清理 ingest 链路孤儿: stale_jobs=%d, assets→error=%d, 队列清除消息=%d",
+            total_jobs,
+            total_assets,
+            purged,
+        )
+    return {"stale_jobs": total_jobs, "assets_to_error": total_assets, "purged_messages": purged}

deploy/configs/nginx-production.conf

@@ -14,6 +14,10 @@ server {
     # SPA routing - index.html 禁止缓存，确保每次获取最新版本
     location / {
         try_files $uri /index.html;
+        # HTML 文档（含 try_files 回退的 SPA 路由，如 /login /app/dashboard）一律 no-cache，
+        # 每次校验 ETag/Last-Modified，保证发版后旧标签页重新加载拿到新 chunk 引用；
+        # 带 hash 的静态资源由下方 ~* \.(js|css...) location 优先匹配，不受影响、保持 immutable
+        add_header Cache-Control "no-cache" always;
     }
 
     # API proxy — Production 环境代理到 production API 容器

deploy/configs/nginx-staging.conf

@@ -21,6 +21,10 @@ server {
     # SPA fallback
     location / {
         try_files $uri /index.html;
+        # HTML 文档（含 try_files 回退的 SPA 路由，如 /login /app/dashboard）一律 no-cache，
+        # 每次校验 ETag/Last-Modified，保证发版后旧标签页重新加载拿到新 chunk 引用；
+        # 带 hash 的静态资源由下方 ~* \.(js|css...) location 优先匹配，不受影响、保持 immutable
+        add_header Cache-Control "no-cache" always;
     }
 
     # API proxy — Staging 环境代理到 staging API 容器

infra/docker/entrypoint-worker.sh

@@ -36,6 +36,7 @@ celery \
   worker \
   --loglevel=info \
   "-B" \
+  -s /tmp/celerybeat-schedule \
   -Q generation \
   "--concurrency=${GEN_CONCURRENCY}" \
   "--max-tasks-per-child=${MAX_TASKS}" \

infra/docker/nginx-production.conf

@@ -16,6 +16,10 @@ server {
     # 注意：不能加 $uri/，否则 /assets 等与构建产物目录同名的路由会被当成目录访问，返回 403
     location / {
         try_files $uri /index.html;
+        # HTML 文档（含 try_files 回退的 SPA 路由，如 /login /app/dashboard）一律 no-cache，
+        # 每次校验 ETag/Last-Modified，保证发版后旧标签页重新加载拿到新 chunk 引用；
+        # 带 hash 的静态资源由下方 ~* \.(js|css...) location 优先匹配，不受影响、保持 immutable
+        add_header Cache-Control "no-cache" always;
     }
 
     # API proxy

infra/docker/nginx-staging.conf

@@ -23,6 +23,10 @@ server {
 
     location / {
         try_files $uri /index.html;
+        # HTML 文档（含 try_files 回退的 SPA 路由，如 /login /app/dashboard）一律 no-cache，
+        # 每次校验 ETag/Last-Modified，保证发版后旧标签页重新加载拿到新 chunk 引用；
+        # 带 hash 的静态资源由下方 ~* \.(js|css...) location 优先匹配，不受影响、保持 immutable
+        add_header Cache-Control "no-cache" always;
     }
 
     # API proxy

infra/docker/nginx.conf

@@ -33,6 +33,10 @@ server {
 
     location / {
         try_files $uri /index.html;
+        # HTML 文档（含 try_files 回退的 SPA 路由，如 /login /app/dashboard）一律 no-cache，
+        # 每次校验 ETag/Last-Modified，保证发版后旧标签页重新加载拿到新 chunk 引用；
+        # 带 hash 的静态资源由下方 ~* \.(js|css...) location 优先匹配，不受影响、保持 immutable
+        add_header Cache-Control "no-cache" always;
     }
 
     # API proxy

packages/adapters/sqlalchemy_impl/asset_repository.py

@@ -488,20 +488,25 @@ class SQLAlchemyAssetRepository:
 
         用于旧客户端未传 file_hash/client_upload_id 时，防止 complete 超时重试
         反复创建 PROCESSING 占位记录。只命中"活动中"的近期记录，READY 历史素材不拦。
+
+        严格模式（#1714 误杀修复）：file_size 必须 > 0 且与记录大小严格一致；
+        file_size=0（大小未知）时直接返回 None——宁可漏判（极端情况下多建一条
+        占位）也不可仅凭同名 + processing 误杀内容全新的视频。
         """
         from datetime import datetime, timedelta, timezone
 
         if not name:
             return None
+        if not file_size or file_size <= 0:
+            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,
+            AssetModel.file_size == file_size,
         )
-        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

packages/application/auth/wechat_sync_use_case.py

@@ -2,8 +2,10 @@
 微信同步登录/注册 Use Case
 
 供 BFF 层调用的系统级接口：
-- 根据 openid 查找用户，找到则登录返回 token
-- 没找到则创建新用户并返回 token
+- 优先按 unionid 识别用户（跨应用/跨端识别同一微信用户）
+- 再按 openid 识别（同一应用内）
+- openid 命中老账号但 unionid 缺失时补写 unionid（开放平台绑定前的存量账号自动关联）
+- 都未命中则创建新用户
 - 支持 unionid 跨应用关联
 """
 
@@ -82,10 +84,10 @@ class WechatSyncResponse:
 
 
 class WechatSyncUseCase:
-    """微信同步登录/注册用例
+    """微信登录/注册同步用例
 
     系统级接口，由 BFF 通过 API Key 调用。
-    职责：根据 openid 查找或创建用户，返回 SaaS token。
+    职责：根据 unionid/openid 查找或创建用户，返回 SaaS token。
     """
 
     def __init__(self, user_repository, session_store=None, jwt_secret_key: str | None = None):
@@ -105,24 +107,54 @@ class WechatSyncUseCase:
                 return None, "openid is required"
 
             is_new_user = False
+            user = None
+            openid_user = None
+            unionid_user = None
 
-            # 1. 按 openid 查找用户
-            user = self.user_repository.find_by_wechat_openid(request.openid)
+            # 1. 先按 unionid 查找（跨应用识别同一微信用户，优先级最高）
+            if request.unionid:
+                unionid_user = self.user_repository.find_by_wechat_unionid(request.unionid)
 
-            # 2. 如果 openid 没找到，尝试 unionid
-            if not user and request.unionid:
-                user = self.user_repository.find_by_wechat_unionid(request.unionid)
-                if user:
-                    # 找到用户但 openid 为空，绑定一下当前 openid
-                    user.wechat_openid = request.openid
-                    self.user_repository.save(user)
+            # 2. 再按 openid 查找（同一应用内）
+            openid_user = self.user_repository.find_by_wechat_openid(request.openid)
 
-            # 3. 都没找到则创建新用户
-            if not user:
+            if unionid_user and openid_user:
+                # 3a. 两边都命中
+                if unionid_user.id == openid_user.id:
+                    # 同一个用户，直接登录
+                    user = unionid_user
+                else:
+                    # unionid 与 openid 分属两个不同账号：数据异常，拒绝写入，
+                    # 交由人工/数据修复合并，避免账号被错误串联
+                    return None, ("wechat account conflict: unionid and openid bound to " "different users")
+            elif unionid_user:
+                # 3b. unionid 命中（跨端老用户），当前 openid 未绑定过：
+                # 确认 openid 没有落在其他账号上后，把新 openid 绑到该用户
+                if openid_user is not None and openid_user.id != unionid_user.id:
+                    return None, ("wechat account conflict: openid bound to another user")
+                if unionid_user.wechat_openid != request.openid:
+                    unionid_user.wechat_openid = request.openid
+                    self.user_repository.save(unionid_user)
+                user = unionid_user
+            elif openid_user:
+                # 3c. 仅 openid 命中（开放平台绑定前创建的存量账号）：
+                # 本次请求带了 unionid 且该账号还没有 unionid 时补写
+                if request.unionid and not openid_user.wechat_unionid:
+                    # 去重：确认该 unionid 没有关联到其他用户
+                    conflict = self.user_repository.find_by_wechat_unionid(request.unionid)
+                    if conflict is not None and conflict.id != openid_user.id:
+                        return None, ("wechat account conflict: unionid already bound to " "another user")
+                    openid_user.wechat_unionid = request.unionid
+                    self.user_repository.save(openid_user)
+                user = openid_user
+            else:
+                # 4. 都没找到，创建新用户
+                # 额外兜底：若 unionid 已被其他账号占用（理论上上面已查过），
+                # 不创建带冲突 unionid 的新账号
                 user = self._create_wechat_user(request)
                 is_new_user = True
 
-            # 4. 创建 session 并生成 token
+            # 5. 创建 session 并生成 token
             session_id = secrets.token_urlsafe(16)
             refresh_token = secrets.token_urlsafe(32)
 

packages/application/ingest_orphan_cleanup.py

@@ -0,0 +1,310 @@
+"""上传/转码链路（IngestJob + Asset）孤儿清理核心逻辑。
+
+#1714：generation 链路有 cleanup_stale_running/pending 兜底，但上传链路
+（ingest_jobs + assets）没有。worker 容器重启/进程 OOM 时，已 prefetch 的
+celery 消息会丢失（transcode 队列 worker_prefetch_multiplier=1，消息预取后
+宕机即丢失，Redis 队列里也不再存在），导致：
+
+- ingest_jobs.status 永久卡 pending/processing
+- assets.status 永久卡 processing/uploading（complete 阶段预建的占位）
+
+本模块提供纯核心（session 注入，便于单测）：超时阈值内无更新的记录
+批量标终态（job→failed、asset→error），并返回 (job_id, celery_task_id)
+列表供调用方 revoke + purge 残留队列消息。
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timedelta, timezone
+from typing import Any, Callable
+
+logger = logging.getLogger(__name__)
+
+# ingest_job PROCESSING 超时阈值：ingest 任务包含下载 + ffprobe + HEVC 转码
+# （1GB 视频约 10-20 分钟）+ 回传 OSS，正常任务可能跑 20-30 分钟；
+# 60 分钟阈值覆盖大文件转码 + 抖动，绝不误杀正常任务。
+INGEST_PROCESSING_TIMEOUT_MINUTES = 60
+
+# ingest_job PENDING 超时阈值：transcode 队列 concurrency=1，队列积压时
+# 正常排队可能较久；90 分钟覆盖 worker 短暂停消费 + 排队。
+INGEST_PENDING_TIMEOUT_MINUTES = 90
+
+# Asset 占位超时阈值：无关联 ingest_job 的孤儿占位（complete 预建后派单失败等），
+# 阈值放宽到 120 分钟，避免与 ingest_job 生命周期错杀。
+ASSET_ORPHAN_TIMEOUT_MINUTES = 120
+
+_TERMINAL_JOB_STATUSES = ("failed", "completed")
+_TERMINAL_ASSET_STATUSES = ("ready", "error", "deleted")
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def cleanup_stale_ingest_jobs(
+    session: Any,
+    *,
+    processing_timeout_minutes: int = INGEST_PROCESSING_TIMEOUT_MINUTES,
+    pending_timeout_minutes: int = INGEST_PENDING_TIMEOUT_MINUTES,
+    commit: bool = True,
+) -> tuple[list[tuple[str, str]], list[str]]:
+    """清理超时卡 pending/processing 的 ingest_jobs，并联动关联 asset。
+
+    Args:
+        session: SQLAlchemy session（或提供 query/commit 的鸭子类型）
+        processing_timeout_minutes: processing 状态超时阈值
+        pending_timeout_minutes: pending 状态超时阈值
+        commit: 是否提交事务
+
+    Returns:
+        (job_items, asset_ids)
+        - job_items: [(job_id, celery_task_id), ...] 供 revoke/purge
+        - asset_ids: 被联动标记为 error 的 asset id 列表
+    """
+    from packages.adapters.sqlalchemy_impl.models import AssetModel, IngestJobModel
+
+    now = _now()
+    processing_cutoff = now - timedelta(minutes=processing_timeout_minutes)
+    pending_cutoff = now - timedelta(minutes=pending_timeout_minutes)
+
+    stale_jobs = (
+        session.query(IngestJobModel)
+        .filter(
+            IngestJobModel.status.in_(["pending", "processing"]),
+            (
+                (IngestJobModel.status == "processing") & (IngestJobModel.updated_at < processing_cutoff)
+                | (IngestJobModel.status == "pending") & (IngestJobModel.created_at < pending_cutoff)
+            ),
+        )
+        .all()
+    )
+
+    job_items: list[tuple[str, str]] = []
+    asset_ids: list[str] = []
+    stale_asset_models: list[Any] = []
+    for job_model in stale_jobs:
+        ref_time = job_model.updated_at or job_model.created_at
+        if ref_time.tzinfo is None:  # SQLite 读回 naive datetime 的防御
+            ref_time = ref_time.replace(tzinfo=timezone.utc)
+        stale_minutes = int((now - ref_time).total_seconds() // 60)
+        job_model.status = "failed"
+        job_model.error_message = (
+            f"转码任务执行中断（超过超时阈值未更新，疑似 worker 重启/进程退出，已卡死 {stale_minutes} 分钟）"
+        )
+        job_model.updated_at = now
+        job_items.append((job_model.id, getattr(job_model, "celery_task_id", "") or ""))
+        if job_model.asset_id:
+            asset_ids.append(job_model.asset_id)
+
+    if asset_ids:
+        stale_asset_models = (
+            session.query(AssetModel)
+            .filter(
+                AssetModel.id.in_(asset_ids),
+                AssetModel.status.in_(["processing", "uploading"]),
+            )
+            .all()
+        )
+        for asset_model in stale_asset_models:
+            asset_model.status = "error"
+            asset_model.updated_at = now
+
+    if commit and (job_items or stale_asset_models):
+        session.commit()
+
+    if job_items:
+        logger.warning(
+            "[ingest-cleanup] 清理 %d 个超时 ingest_job（processing>%dm / pending>%dm），联动 %d 个 asset 标 error",
+            len(job_items),
+            processing_timeout_minutes,
+            pending_timeout_minutes,
+            len(stale_asset_models),
+        )
+    return job_items, [a.id for a in stale_asset_models]
+
+
+def cleanup_orphan_processing_assets(
+    session: Any,
+    *,
+    timeout_minutes: int = ASSET_ORPHAN_TIMEOUT_MINUTES,
+    commit: bool = True,
+) -> list[str]:
+    """清理无 ingest_job 关联、超时卡 processing/uploading 的孤儿 asset 占位。
+
+    complete 阶段预建 asset 后若派单失败（或 direct 上传 complete 后
+    未触发 ingest），占位会永久卡住。这类 asset 没有对应 ingest_job，
+    只能按 created_at 超时兜底标 error。
+    """
+    from packages.adapters.sqlalchemy_impl.models import AssetModel, IngestJobModel
+
+    cutoff = _now() - timedelta(minutes=timeout_minutes)
+    orphan_assets = (
+        session.query(AssetModel)
+        .outerjoin(IngestJobModel, IngestJobModel.asset_id == AssetModel.id)
+        .filter(
+            AssetModel.status.in_(["processing", "uploading"]),
+            AssetModel.created_at < cutoff,
+            IngestJobModel.id.is_(None),
+        )
+        .all()
+    )
+    for asset_model in orphan_assets:
+        asset_model.status = "error"
+        asset_model.updated_at = _now()
+    if commit and orphan_assets:
+        session.commit()
+        logger.warning("[ingest-cleanup] 清理 %d 个无 job 关联的超时孤儿 asset 占位", len(orphan_assets))
+    return [a.id for a in orphan_assets]
+
+
+def revoke_stale_ingest_messages(
+    job_items: list[tuple[str, str]],
+    *,
+    celery_app_factory: Callable[[], Any] | None = None,
+    broker_url_factory: Callable[[], str] | None = None,
+) -> int:
+    """revoke + 物理清理 ingest 作废消息（transcode/celery 队列）。
+
+    消息可能已在 worker 宕机时丢失（队列里查不到），那也无害；
+    若消息还在（极端重复投递），物理清除防止重投执行。
+    失败不阻断清理（ingest_asset 的执行前状态守卫是第二道防线）。
+    """
+    biz_ids = [jid for jid, _ in job_items if jid]
+    celery_ids = [cid for _, cid in job_items if cid]
+    if not biz_ids and not celery_ids:
+        return 0
+    try:
+        from packages.shared.celery_orphan_guard import revoke_and_purge
+
+        app = celery_app_factory() if celery_app_factory else None
+        broker_url = broker_url_factory() if broker_url_factory else ""
+        if app is None or not broker_url:
+            from worker_app.celery_app import celery_app as _app
+            from worker_app.core.config import get_settings
+
+            app = _app
+            broker_url = get_settings().broker_url
+        return revoke_and_purge(
+            app,
+            broker_url,
+            business_task_ids=biz_ids,
+            celery_task_ids=celery_ids,
+            queue_names=("transcode", "celery"),
+        )
+    except Exception as e:  # noqa: BLE001
+        logger.error("撤销作废 ingest 队列消息失败（执行前守卫仍会兜底）: %s", e, exc_info=True)
+        return 0
+
+
+# ── worker 启动恢复（#1714）──────────────────────────────────────────────
+#
+# task_acks_late=True 下，worker 崩溃/容器重启时未 ack 的消息理论上会在
+# visibility_timeout 到期后重新投递；但 prefork 进程异常、部署窗口跨
+# visibility 配置边界等场景仍可能留下卡在 processing 的 ingest_job
+# （staging 实证：03:16 派单、03:45 置 processing 后 worker 重启，
+# unacked 消息未重投，任务永久卡死）。启动时做一次显式恢复扫描兜底。
+#
+# 恢复策略：processing 超过 stuck_minutes（默认 10 分钟，部署中跨进程
+# 交接的正常窗口 < 10 分钟，不会误抢别的 worker 正在执行的任务）的 job，
+# CAS 重置为 pending 并重新 send_task；旧消息若后来重投，ingest_asset
+# 的执行前守卫会把状态不匹配的旧 celery 消息丢弃。
+
+
+def recover_stuck_ingest_jobs_on_startup(
+    session: Any,
+    *,
+    send_task: Callable[..., Any] | None = None,
+    update_celery_task_id: Callable[[str, str], None] | None = None,
+    lock_acquire: Callable[[], bool] | None = None,
+    stuck_minutes: int = 10,
+    commit: bool = True,
+) -> int:
+    """worker 启动时把卡在 processing 超时的 ingest_job 重新派单。
+
+    Args:
+        session: SQLAlchemy session
+        send_task: celery send_task 可调用（注入便于测试）；不传则用 worker celery_app
+        update_celery_task_id: 回写新 celery task id 的回调（job_id, new_task_id）
+        lock_acquire: 分布式锁获取回调（多 worker 进程同时启动时只允许一个恢复）；
+            返回 False 表示未抢到锁，本次跳过
+        stuck_minutes: processing 超过该分钟数视为卡死
+
+    Returns:
+        重新派单的 job 数
+    """
+    if lock_acquire is not None and not lock_acquire():
+        logger.info("[ingest-recover] 未抢到恢复锁，跳过（另一进程正在恢复）")
+        return 0
+
+    from packages.adapters.sqlalchemy_impl.models import IngestJobModel
+
+    cutoff = _now() - timedelta(minutes=stuck_minutes)
+    stuck_jobs = (
+        session.query(IngestJobModel)
+        .filter(IngestJobModel.status == "processing", IngestJobModel.updated_at < cutoff)
+        .order_by(IngestJobModel.updated_at.asc())
+        .all()
+    )
+
+    if not stuck_jobs:
+        logger.info("[ingest-recover] 无卡死 processing ingest_job 需要恢复")
+        return 0
+
+    if send_task is None:
+        from worker_app.celery_app import celery_app as _app
+
+        send_task = _app.send_task
+
+    recovered = 0
+    for job_model in stuck_jobs:
+        # CAS：只有仍是 processing 才重置（并发/旧消息已回写终态时不碰）
+        updated = (
+            session.query(IngestJobModel)
+            .filter(IngestJobModel.id == job_model.id, IngestJobModel.status == "processing")
+            .update({"status": "pending", "error_message": "", "updated_at": _now()})
+        )
+        if not updated:
+            continue
+        try:
+            result = send_task("worker.ingest_asset", args=[job_model.id])
+            new_task_id = getattr(result, "id", "") or ""
+        except Exception as e:  # noqa: BLE001
+            logger.error("[ingest-recover] 重新派单失败 job_id=%s: %s", job_model.id, e)
+            continue
+        if new_task_id:
+            job_model.celery_task_id = new_task_id
+            if update_celery_task_id is not None:
+                update_celery_task_id(job_model.id, new_task_id)
+        logger.warning(
+            "[ingest-recover] 卡死 ingest_job %s 已重置 pending 并重新派单 (new celery task=%s)",
+            job_model.id,
+            new_task_id,
+        )
+        recovered += 1
+
+    if commit and recovered:
+        session.commit()
+    logger.warning("[ingest-recover] 启动恢复完成，共重新派单 %d 个卡死 ingest_job", recovered)
+    return recovered
+
+
+def make_redis_recovery_lock(lock_key: str = "ingest:recover:startup", ttl_seconds: int = 300):
+    """构造基于 Redis SET NX 的恢复锁工厂（多 worker 进程互斥）。
+
+    返回一个无参 callable，调用时尝试抢锁：抢到返回 True，未抢到返回 False。
+    Redis 不可用时不阻断启动恢复（返回 True，恢复逻辑自身有 CAS 幂等保护）。
+    """
+
+    def _acquire() -> bool:
+        try:
+            import redis as redis_lib
+            from worker_app.core.config import get_settings
+
+            client = redis_lib.Redis.from_url(get_settings().broker_url)
+            return bool(client.set(lock_key, "1", nx=True, ex=ttl_seconds))
+        except Exception as e:  # noqa: BLE001
+            logger.warning("[ingest-recover] Redis 锁不可用，降级为无锁执行（CAS 兜底）: %s", e)
+            return True
+
+    return _acquire

tests/unit/test_asset_repo_fallback_dedup_1714.py

@@ -0,0 +1,93 @@
+"""#1714 find_recent_active_by_library_and_name 严格模式测试。
+
+file_size=0（未知）时必须返回 None（宁可漏判不可误杀）；
+大小严格匹配；只命中近期 UPLOADING/PROCESSING 记录。
+"""
+
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+from sqlalchemy import create_engine  # noqa: E402
+from sqlalchemy.orm import sessionmaker  # noqa: E402
+
+from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository  # noqa: E402
+from packages.adapters.sqlalchemy_impl.models import Base  # noqa: E402
+from packages.domain import Asset, AssetStatus  # noqa: E402
+
+
+def _repository():
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    session = sessionmaker(bind=engine)()
+    return SQLAlchemyAssetRepository(session)
+
+
+def _mk_asset(name="IMG_2285.MOV", file_size=5_000_000, status=AssetStatus.PROCESSING, minutes_ago=5):
+    asset = Asset.create(
+        project_id="proj-1",
+        library_id="lib-1",
+        name=name,
+        storage_key=f"uploads/x/{name}",
+        mime_type="video/quicktime",
+        file_size=file_size,
+    )
+    asset.status = status
+    asset.created_at = datetime.now(timezone.utc) - timedelta(minutes=minutes_ago)
+    return asset
+
+
+def test_returns_none_when_file_size_zero():
+    """file_size=0（大小未知）直接返回 None——不许仅凭同名 + processing 判重。"""
+    repo = _repository()
+    repo.create(_mk_asset(file_size=0))
+
+    result = repo.find_recent_active_by_library_and_name(library_id="lib-1", name="IMG_2285.MOV", file_size=0)
+    assert result is None
+
+
+def test_matches_when_name_size_strict_equal():
+    """同名 + 同大小 + processing 近期记录 → 命中。"""
+    repo = _repository()
+    repo.create(_mk_asset(file_size=5_000_000))
+
+    result = repo.find_recent_active_by_library_and_name(library_id="lib-1", name="IMG_2285.MOV", file_size=5_000_000)
+    assert result is not None
+    assert result.name == "IMG_2285.MOV"
+
+
+def test_no_match_when_same_name_but_different_size():
+    """同名但大小不同 → 不命中（内容全新的视频不能误杀）。"""
+    repo = _repository()
+    repo.create(_mk_asset(file_size=5_000_000))
+
+    result = repo.find_recent_active_by_library_and_name(library_id="lib-1", name="IMG_2285.MOV", file_size=9_999_999)
+    assert result is None
+
+
+def test_no_match_ready_history_even_with_same_size():
+    """READY 历史同名素材不命中（允许再次上传同名文件）。"""
+    repo = _repository()
+    repo.create(_mk_asset(file_size=5_000_000, status=AssetStatus.READY))
+
+    result = repo.find_recent_active_by_library_and_name(library_id="lib-1", name="IMG_2285.MOV", file_size=5_000_000)
+    assert result is None
+
+
+def test_no_match_when_window_expired():
+    """超过 30 分钟窗口的活动记录不命中。"""
+    repo = _repository()
+    repo.create(_mk_asset(file_size=5_000_000, minutes_ago=45))
+
+    result = repo.find_recent_active_by_library_and_name(
+        library_id="lib-1", name="IMG_2285.MOV", within_minutes=30, file_size=5_000_000
+    )
+    assert result is None
+
+
+def test_returns_none_when_name_empty():
+    repo = _repository()
+    result = repo.find_recent_active_by_library_and_name(library_id="lib-1", name="", file_size=100)
+    assert result is None

tests/unit/test_cleanup_ingest_beat_1714.py

@@ -0,0 +1,75 @@
+"""#1714 beat 任务 scheduled_cleanup_stale_ingest_jobs 薄封装测试。
+
+mock SessionLocal 和清理核心，验证 beat 任务正确串联
+cleanup_stale_ingest_jobs → cleanup_orphan_processing_assets → revoke 消息。
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test_beat.db")
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+import worker_app.tasks.cleanup as cleanup  # noqa: E402
+
+
+def test_beat_cleanup_calls_core_and_revokes():
+    """beat 任务串联三个核心步骤，返回汇总计数。"""
+    fake_session = MagicMock()
+
+    with (
+        patch("worker_app.db.SessionLocal", return_value=fake_session) as m_db,
+        patch(
+            "packages.application.ingest_orphan_cleanup.cleanup_stale_ingest_jobs",
+            return_value=([("job-1", "cel-1"), ("job-2", "")], ["a-1"]),
+        ) as m_jobs,
+        patch(
+            "packages.application.ingest_orphan_cleanup.cleanup_orphan_processing_assets",
+            return_value=["a-2"],
+        ) as m_assets,
+        patch(
+            "packages.shared.celery_orphan_guard.revoke_and_purge",
+            return_value=1,
+        ) as m_revoke,
+    ):
+        result = cleanup.scheduled_cleanup_stale_ingest_jobs()
+
+    m_db.assert_called_once()
+    m_jobs.assert_called_once()
+    assert m_jobs.call_args.kwargs["processing_timeout_minutes"] == 60
+    m_assets.assert_called_once()
+    m_revoke.assert_called_once()
+    # 队列名只传 transcode/celery（不传 generation）
+    assert m_revoke.call_args.kwargs["queue_names"] == ("transcode", "celery")
+    fake_session.close.assert_called_once()
+    assert result == {"stale_jobs": 2, "assets_to_error": 2, "purged_messages": 1}
+
+
+def test_beat_cleanup_no_op_when_nothing_stale():
+    """无孤儿时不调 revoke，返回全 0。"""
+    fake_session = MagicMock()
+
+    with (
+        patch("worker_app.db.SessionLocal", return_value=fake_session),
+        patch(
+            "packages.application.ingest_orphan_cleanup.cleanup_stale_ingest_jobs",
+            return_value=([], []),
+        ),
+        patch(
+            "packages.application.ingest_orphan_cleanup.cleanup_orphan_processing_assets",
+            return_value=[],
+        ),
+        patch("packages.shared.celery_orphan_guard.revoke_and_purge") as m_revoke,
+    ):
+        result = cleanup.scheduled_cleanup_stale_ingest_jobs()
+
+    m_revoke.assert_not_called()
+    assert result == {"stale_jobs": 0, "assets_to_error": 0, "purged_messages": 0}

tests/unit/test_ingest_orphan_cleanup_1714.py

@@ -0,0 +1,262 @@
+"""#1714 上传/转码链路（IngestJob + Asset）孤儿清理测试。
+
+场景：worker 容器重启/进程 OOM 时，已 prefetch 的 transcode celery 消息丢失，
+ingest_job 永久卡 pending/processing、asset 永久卡 processing/uploading。
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test_ingest_orphan.db")
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from sqlalchemy import create_engine  # noqa: E402
+from sqlalchemy.orm import sessionmaker  # noqa: E402
+
+from packages.adapters.sqlalchemy_impl.models import AssetModel, Base, IngestJobModel  # noqa: E402
+from packages.application.ingest_orphan_cleanup import (  # noqa: E402
+    cleanup_orphan_processing_assets,
+    cleanup_stale_ingest_jobs,
+)
+
+
+@pytest.fixture()
+def session():
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(bind=engine)
+    Session = sessionmaker(bind=engine)
+    db = Session()
+    yield db
+    db.close()
+
+
+def _mk_job(session, *, status="processing", celery_task_id="cel-1", asset_id="a-1", minutes_ago=90):
+    now = datetime.now(timezone.utc)
+    job = IngestJobModel(
+        id=f"job-{minutes_ago}-{status}-{celery_task_id}",
+        project_id="p-1",
+        library_id="lib-1",
+        storage_key="uploads/x/IMG_2285.MOV",
+        status=status,
+        asset_id=asset_id,
+        celery_task_id=celery_task_id,
+        created_at=now - timedelta(minutes=minutes_ago),
+        updated_at=now - timedelta(minutes=minutes_ago),
+    )
+    session.add(job)
+    session.commit()
+    return job
+
+
+def _mk_asset(session, *, id="a-1", status="processing", minutes_ago=90, file_size=0):
+    now = datetime.now(timezone.utc)
+    asset = AssetModel(
+        id=id,
+        project_id="p-1",
+        asset_library_id="lib-1",
+        name="IMG_2285.MOV",
+        file_type="video",
+        file_size=file_size,
+        file_url="https://example.com/x.mov",
+        storage_key="uploads/x/IMG_2285.MOV",
+        status=status,
+        uploaded_by_user_id="u-1",
+        created_at=now - timedelta(minutes=minutes_ago),
+        updated_at=now - timedelta(minutes=minutes_ago),
+    )
+    session.add(asset)
+    session.commit()
+    return asset
+
+
+class TestCleanupStaleIngestJobs:
+    def test_stale_processing_job_marked_failed_and_asset_to_error(self, session):
+        """processing 超 60 分钟 → job failed，关联 processing asset → error。"""
+        _mk_asset(session, id="a-1", status="processing")
+        _mk_job(session, status="processing", celery_task_id="cel-dead", asset_id="a-1", minutes_ago=90)
+
+        items, asset_ids = cleanup_stale_ingest_jobs(session, processing_timeout_minutes=60, pending_timeout_minutes=90)
+
+        assert len(items) == 1
+        assert items[0] == ("job-90-processing-cel-dead", "cel-dead")
+        assert asset_ids == ["a-1"]
+        db_job = session.query(IngestJobModel).one()
+        assert db_job.status == "failed"
+        assert "中断" in db_job.error_message
+        db_asset = session.query(AssetModel).one()
+        assert db_asset.status == "error"
+
+    def test_stale_pending_job_marked_failed(self, session):
+        """pending 超 90 分钟（从未被消费）→ job failed。"""
+        _mk_asset(session, id="a-2", status="uploading")
+        _mk_job(session, status="pending", celery_task_id="", asset_id="a-2", minutes_ago=120)
+
+        items, asset_ids = cleanup_stale_ingest_jobs(session, processing_timeout_minutes=60, pending_timeout_minutes=90)
+
+        assert len(items) == 1
+        assert items[0][1] == ""  # 无 celery task id
+        assert session.query(IngestJobModel).one().status == "failed"
+        assert session.query(AssetModel).one().status == "error"
+
+    def test_recent_processing_job_not_touched(self, session):
+        """processing 仅 10 分钟（正常转码中）→ 不误杀。"""
+        _mk_asset(session, id="a-3", status="processing", minutes_ago=10)
+        _mk_job(session, status="processing", celery_task_id="cel-live", asset_id="a-3", minutes_ago=10)
+
+        items, asset_ids = cleanup_stale_ingest_jobs(session, processing_timeout_minutes=60, pending_timeout_minutes=90)
+
+        assert items == []
+        assert asset_ids == []
+        assert session.query(IngestJobModel).one().status == "processing"
+        assert session.query(AssetModel).one().status == "processing"
+
+    def test_recent_pending_job_not_touched(self, session):
+        """pending 仅 30 分钟（队列积压排队中）→ 不误杀。"""
+        _mk_job(session, status="pending", asset_id="", minutes_ago=30)
+
+        items, _ = cleanup_stale_ingest_jobs(session, processing_timeout_minutes=60, pending_timeout_minutes=90)
+
+        assert items == []
+        assert session.query(IngestJobModel).one().status == "pending"
+
+    def test_terminal_job_not_touched(self, session):
+        """已 completed/failed 的 job 不动。"""
+        _mk_job(session, status="completed", celery_task_id="", asset_id="", minutes_ago=999)
+        _mk_job(session, status="failed", celery_task_id="", asset_id="", minutes_ago=999)
+
+        items, _ = cleanup_stale_ingest_jobs(session)
+
+        assert items == []
+        statuses = sorted(j.status for j in session.query(IngestJobModel).all())
+        assert statuses == ["completed", "failed"]
+
+    def test_ready_asset_not_demoted(self, session):
+        """关联 asset 已是 ready（转码其实成功了，仅 job 回写失败）→ 不降级为 error。"""
+        _mk_asset(session, id="a-4", status="ready")
+        _mk_job(session, status="processing", celery_task_id="cel-x", asset_id="a-4", minutes_ago=90)
+
+        _, asset_ids = cleanup_stale_ingest_jobs(session)
+
+        assert asset_ids == []  # ready 不动
+        assert session.query(AssetModel).one().status == "ready"
+
+
+class TestCleanupOrphanProcessingAssets:
+    def test_orphan_asset_without_job_marked_error(self, session):
+        """无 ingest_job 关联、created 超 120 分钟的 processing 占位 → error。"""
+        _mk_asset(session, id="orphan-1", status="processing", minutes_ago=150)
+
+        ids = cleanup_orphan_processing_assets(session, timeout_minutes=120)
+
+        assert ids == ["orphan-1"]
+        assert session.query(AssetModel).one().status == "error"
+
+    def test_asset_with_active_job_not_touched(self, session):
+        """有 processing job 关联的 asset 不由本函数处理（归 cleanup_stale_ingest_jobs）。"""
+        _mk_asset(session, id="a-5", status="processing", minutes_ago=150)
+        _mk_job(session, status="processing", asset_id="a-5", minutes_ago=150)
+
+        ids = cleanup_orphan_processing_assets(session, timeout_minutes=120)
+
+        assert ids == []
+        assert session.query(AssetModel).one().status == "processing"
+
+    def test_recent_orphan_asset_not_touched(self, session):
+        """无 job 但才创建 30 分钟 → 可能 complete 刚建、job 派单中，不动。"""
+        _mk_asset(session, id="orphan-2", status="processing", minutes_ago=30)
+
+        ids = cleanup_orphan_processing_assets(session, timeout_minutes=120)
+
+        assert ids == []
+        assert session.query(AssetModel).one().status == "processing"
+
+
+class TestRecoverStuckIngestJobsOnStartup:
+    def test_stuck_processing_job_requeued(self, session):
+        """processing 超 10 分钟 → 重置 pending 并重新 send_task，回写新 celery id。"""
+        from types import SimpleNamespace
+
+        from packages.application.ingest_orphan_cleanup import recover_stuck_ingest_jobs_on_startup
+
+        job = _mk_job(session, status="processing", celery_task_id="old-cel-1", asset_id="a-1", minutes_ago=30)
+
+        sent = []
+
+        def fake_send_task(name, args=None, **kw):
+            sent.append((name, args))
+            return SimpleNamespace(id="new-cel-9")
+
+        updated_ids = []
+        recovered = recover_stuck_ingest_jobs_on_startup(
+            session,
+            send_task=fake_send_task,
+            update_celery_task_id=lambda jid, cid: updated_ids.append((jid, cid)),
+            stuck_minutes=10,
+        )
+
+        assert recovered == 1
+        assert sent == [("worker.ingest_asset", [job.id])]
+        refreshed = session.query(IngestJobModel).filter_by(id=job.id).one()
+        assert refreshed.status == "pending"
+        assert refreshed.celery_task_id == "new-cel-9"
+        assert updated_ids == [(job.id, "new-cel-9")]
+
+    def test_recent_processing_job_not_touched(self, session):
+        """processing 仅 5 分钟（正常转码中/部署交接窗口）→ 不抢。"""
+        from packages.application.ingest_orphan_cleanup import recover_stuck_ingest_jobs_on_startup
+
+        _mk_job(session, status="processing", celery_task_id="live", asset_id="", minutes_ago=5)
+
+        sent = []
+        recovered = recover_stuck_ingest_jobs_on_startup(
+            session,
+            send_task=lambda *a, **k: sent.append(a),
+            stuck_minutes=10,
+        )
+
+        assert recovered == 0
+        assert sent == []
+        assert session.query(IngestJobModel).one().status == "processing"
+
+    def test_lock_not_acquired_skips(self, session):
+        """未抢到分布式锁（另一 worker 正在恢复）→ 跳过。"""
+        from packages.application.ingest_orphan_cleanup import recover_stuck_ingest_jobs_on_startup
+
+        _mk_job(session, status="processing", celery_task_id="x", asset_id="", minutes_ago=30)
+
+        recovered = recover_stuck_ingest_jobs_on_startup(
+            session,
+            send_task=lambda *a, **k: None,
+            lock_acquire=lambda: False,
+            stuck_minutes=10,
+        )
+
+        assert recovered == 0
+        assert session.query(IngestJobModel).one().status == "processing"
+
+    def test_pending_and_terminal_not_requeued(self, session):
+        """pending/已终态 job 不在恢复范围。"""
+        from packages.application.ingest_orphan_cleanup import recover_stuck_ingest_jobs_on_startup
+
+        _mk_job(session, status="pending", celery_task_id="", asset_id="", minutes_ago=60)
+        _mk_job(session, status="failed", celery_task_id="", asset_id="", minutes_ago=60)
+
+        recovered = recover_stuck_ingest_jobs_on_startup(
+            session,
+            send_task=lambda *a, **k: None,
+            stuck_minutes=10,
+        )
+
+        assert recovered == 0
+        statuses = sorted(j.status for j in session.query(IngestJobModel).all())
+        assert statuses == ["failed", "pending"]

tests/unit/test_upload_complete_idempotency_1714.py

@@ -73,6 +73,9 @@ class StubAssetRepository:
     def find_recent_active_by_library_and_name(
         self, library_id: str, name: str, within_minutes: int = 30, file_size: int = 0
     ) -> Asset | None:
+        # 严格模式（#1714）：大小未知（0）直接不命中，宁可漏判不可误杀
+        if not file_size or file_size <= 0:
+            return None
         cutoff = datetime.now(timezone.utc) - timedelta(minutes=within_minutes)
         candidates = [
             a
@@ -81,7 +84,7 @@ class StubAssetRepository:
             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)
+            and a.file_size == file_size
         ]
         return max(candidates, key=lambda a: a.created_at) if candidates else None
 
@@ -234,14 +237,21 @@ class TestDirectCompleteIdempotency:
         不应再建第二条。
         """
         client, asset_repo, ingest_repo, _ = _client()
-        # 第一次 complete（旧客户端无 token/hash）
-        r1 = client.post("/api/v1/direct/complete", json=COMPLETE_BODY)
+        # 第一次 complete（旧客户端无 token/hash，但 file_size 可知）
+        r1 = client.post(
+            "/api/v1/direct/complete",
+            json={**COMPLETE_BODY, "file_size": 5_000_000},
+        )
         assert r1.json()["duplicated"] is False
         # 重试：重新 prepare 产生新 storage_key（仅 uuid 目录不同，文件名一致——
-        # 前端重试传的是同一个 File），且近期
+        # 前端重试传的是同一个 File），且近期；同大小才允许兜底命中
         r2 = client.post(
             "/api/v1/direct/complete",
-            json={**COMPLETE_BODY, "storage_key": "uploads/retry/IMG_2282.MOV", "file_size": 0},
+            json={
+                **COMPLETE_BODY,
+                "storage_key": "uploads/retry/IMG_2282.MOV",
+                "file_size": 5_000_000,
+            },
         )
         assert r2.status_code == 200
         assert r2.json()["duplicated"] is True
@@ -249,6 +259,80 @@ class TestDirectCompleteIdempotency:
         assert len(asset_repo.created) == 1
         assert ingest_repo.created_count == 1
 
+    def test_fallback_dedup_skipped_when_file_size_unknown(self):
+        """file_size=0（未知）时不允许仅凭同名 + processing 判重，直接放行（#1714）。
+
+        根因场景：complete 没传 file_size，30 分钟内同名占位（如 iPhone 的
+        IMG_2285.MOV）会把内容/大小全新的视频误判为重复跳过。
+        """
+        client, asset_repo, _ingest_repo, _ = _client()
+        r1 = client.post(
+            "/api/v1/direct/complete",
+            json={**COMPLETE_BODY, "file_size": 0},
+        )
+        assert r1.json()["duplicated"] is False
+        # 第二个全新视频：同名（IMG_2285.MOV）、无 hash/token、file_size 仍未知
+        r2 = client.post(
+            "/api/v1/direct/complete",
+            json={**COMPLETE_BODY, "storage_key": "uploads/retry2/IMG_2282.MOV", "file_size": 0},
+        )
+        assert r2.status_code == 200
+        assert r2.json()["duplicated"] is False  # 不能误杀
+        assert len(asset_repo.created) == 2  # 两条记录，放行新上传
+
+    def test_fallback_dedup_skipped_when_same_name_but_different_size(self):
+        """同名但 file_size 不同 → 不判重，正常建记录（#1714）。"""
+        client, asset_repo, _ingest_repo, _ = _client()
+        r1 = client.post(
+            "/api/v1/direct/complete",
+            json={**COMPLETE_BODY, "file_size": 5_000_000},
+        )
+        assert r1.json()["duplicated"] is False
+        r2 = client.post(
+            "/api/v1/direct/complete",
+            json={
+                **COMPLETE_BODY,
+                "storage_key": "uploads/retry3/IMG_2282.MOV",
+                "file_size": 9_999_999,  # 同名但大小完全不同的新视频
+            },
+        )
+        assert r2.status_code == 200
+        assert r2.json()["duplicated"] is False
+        assert len(asset_repo.created) == 2
+
+    def test_fallback_dedup_skipped_when_hash_present_even_if_name_size_match(self):
+        """file_hash 非空且 hash 未命中时，不允许退回同名兜底（#1714）。
+
+        hash 已能代表内容：同名同大小但 hash 不同是真实的新内容，必须放行。
+        """
+        client, asset_repo, _ingest_repo, _ = _client()
+        # 第一次：某 hash 的视频
+        r1 = client.post(
+            "/api/v1/direct/complete",
+            json={
+                **COMPLETE_BODY,
+                "file_hash": "a" * 64,
+                "client_upload_id": "tok-1",
+                "file_size": 5_000_000,
+            },
+        )
+        assert r1.json()["duplicated"] is False
+        # 第二次：同名同大小但 hash 不同（新视频内容不同）；
+        # 注意 client_upload_id 也必须不同，否则会先被 token 命中
+        r2 = client.post(
+            "/api/v1/direct/complete",
+            json={
+                **COMPLETE_BODY,
+                "storage_key": "uploads/retry4/IMG_2282.MOV",
+                "file_hash": "b" * 64,
+                "client_upload_id": "tok-2",
+                "file_size": 5_000_000,
+            },
+        )
+        assert r2.status_code == 200
+        assert r2.json()["duplicated"] is False
+        assert len(asset_repo.created) == 2
+
     def test_fallback_dedup_ignores_ready_history(self):
         """READY 历史同名素材不触发兜底（允许用户再次上传同名文件）。"""
         ready = Asset(

tests/unit/test_wechat_sync_use_case.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock
 
 import pytest
 
@@ -13,10 +13,17 @@ from packages.application.auth.wechat_sync_use_case import (
 )
 from packages.domain.entities import User
 
+JWT_KEY = "test-secret-key-for-jwt-12345"
+
 
 @pytest.fixture
 def mock_user_repo():
-    return MagicMock()
+    repo = MagicMock()
+    # 默认全部查不到，具体用例再覆盖
+    repo.find_by_wechat_openid.return_value = None
+    repo.find_by_wechat_unionid.return_value = None
+    repo.find_by_username.return_value = None
+    return repo
 
 
 @pytest.fixture
@@ -41,40 +48,29 @@ def sample_user():
     return user
 
 
-class TestWechatSyncRequest:
-    """WechatSyncRequest 测试"""
+def make_use_case(repo, store):
+    return WechatSyncUseCase(repo, session_store=store, jwt_secret_key=JWT_KEY)
 
+
+class TestWechatSyncRequest:
     def test_openid_stripped(self):
-        """openid 被 strip"""
-        req = WechatSyncRequest(openid="  openid_123  ")
-        assert req.openid == "openid_123"
+        assert WechatSyncRequest(openid="  openid_123  ").openid == "openid_123"
 
     def test_unionid_stripped(self):
-        """unionid 被 strip"""
-        req = WechatSyncRequest(openid="o1", unionid="  unionid_456  ")
-        assert req.unionid == "unionid_456"
+        assert WechatSyncRequest(openid="o1", unionid="  unionid_456  ").unionid == "unionid_456"
 
     def test_default_nickname(self):
-        """默认昵称"""
-        req = WechatSyncRequest(openid="o1")
-        assert req.nickname == "微信用户"
+        assert WechatSyncRequest(openid="o1").nickname == "微信用户"
 
     def test_default_source(self):
-        """默认来源"""
-        req = WechatSyncRequest(openid="o1")
-        assert req.source == "miniapp"
+        assert WechatSyncRequest(openid="o1").source == "miniapp"
 
     def test_empty_unionid(self):
-        """不传 unionid 默认为空字符串"""
-        req = WechatSyncRequest(openid="o1")
-        assert req.unionid == ""
+        assert WechatSyncRequest(openid="o1").unionid == ""
 
 
 class TestWechatSyncResponse:
-    """WechatSyncResponse 测试"""
-
     def test_to_dict_contains_fields(self):
-        """to_dict 包含所有必要字段"""
         resp = WechatSyncResponse(
             access_token="access_123",
             refresh_token="refresh_456",
@@ -85,276 +81,241 @@ class TestWechatSyncResponse:
             expires_in=1800,
         )
         data = resp.to_dict()
-
         assert data["access_token"] == "access_123"
-        assert data["token"] == "access_123"  # 兼容字段
+        assert data["token"] == "access_123"
         assert data["refresh_token"] == "refresh_456"
         assert data["user_id"] == "user_001"
         assert data["is_new_user"] is False
         assert data["expires_in"] == 1800
-        assert "user" in data
-        assert "user_info" in data
         assert data["user"]["id"] == "user_001"
-        assert data["user"]["nickname"] == "测试用户"
         assert data["user"]["display_name"] == "测试用户"
 
 
-class TestWechatSyncUseCaseLoginExisting:
-    """已有用户登录测试"""
-
+class TestWechatSyncLoginExisting:
     def test_login_by_openid(self, mock_user_repo, mock_session_store, sample_user):
-        """通过 openid 登录已有用户"""
+        """openid 命中、unionid 一致，正常登录"""
         mock_user_repo.find_by_wechat_openid.return_value = sample_user
-        mock_user_repo.find_by_wechat_unionid.return_value = None
-        mock_user_repo.save.return_value = sample_user
-
-        use_case = WechatSyncUseCase(
-            mock_user_repo,
-            session_store=mock_session_store,
-            jwt_secret_key="test-secret-key-for-jwt-12345",
-        )
-        request = WechatSyncRequest(openid="openid_123", nickname="测试")
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-        assert response.user_id == "user_001"
-        assert response.is_new_user is False
-        mock_user_repo.find_by_wechat_openid.assert_called_once_with("openid_123")
-        mock_session_store.save_session.assert_called_once()
-
-    def test_login_by_unionid(self, mock_user_repo, mock_session_store, sample_user):
-        """openid 没找到，通过 unionid 找到并绑定 openid"""
-        sample_user.wechat_openid = None  # 没有当前 openid
-        mock_user_repo.find_by_wechat_openid.return_value = None
         mock_user_repo.find_by_wechat_unionid.return_value = sample_user
-        mock_user_repo.save.return_value = sample_user
-
-        use_case = WechatSyncUseCase(
-            mock_user_repo,
-            session_store=mock_session_store,
-            jwt_secret_key="test-secret-key-for-jwt-12345",
+        resp, err = make_use_case(mock_user_repo, mock_session_store).execute(
+            WechatSyncRequest(openid="openid_123", unionid="unionid_456")
         )
-        request = WechatSyncRequest(
-            openid="new_openid",
-            unionid="unionid_456",
-            nickname="测试",
-        )
-        response, error = use_case.execute(request)
+        assert err is None
+        assert resp.user_id == "user_001"
+        assert resp.is_new_user is False
 
-        assert error is None
-        assert response is not None
-        assert response.is_new_user is False
-        # 应该保存了新的 openid
-        assert sample_user.wechat_openid == "new_openid"
+    def test_backfill_unionid_for_legacy_openid_user(self, mock_user_repo, mock_session_store):
+        """核心修复：openid 命中的老账号没有 unionid，请求带 unionid 时补写"""
+        legacy = User(
+            id="legacy_001",
+            email="legacy@wechat.local",
+            username="wx_legacy",
+            display_name="微信用户",
+            password_hash="h",
+            email_verified=True,
+            wechat_openid="oGjxK3_old",
+            wechat_unionid=None,
+        )
+        mock_user_repo.find_by_wechat_openid.return_value = legacy
+        # unionid 查找：补写前确认无其他账号占用
+        mock_user_repo.find_by_wechat_unionid.return_value = None
+
+        resp, err = make_use_case(mock_user_repo, mock_session_store).execute(
+            WechatSyncRequest(openid="oGjxK3_old", unionid="o5nVk_union")
+        )
+        assert err is None
+        assert resp.user_id == "legacy_001"
+        assert resp.is_new_user is False
+        assert legacy.wechat_unionid == "o5nVk_union"
+        # 至少保存过一次（补写 + 最后登录更新）
         mock_user_repo.save.assert_called()
 
-    def test_updates_last_login(self, mock_user_repo, mock_session_store, sample_user):
-        """登录时更新最后登录信息"""
-        mock_user_repo.find_by_wechat_openid.return_value = sample_user
-        mock_user_repo.save.return_value = sample_user
+    def test_login_by_unionid_binds_new_openid(self, mock_user_repo, mock_session_store, sample_user):
+        """unionid 命中（跨端老用户），openid 未绑定过 → 绑定新 openid"""
+        sample_user.wechat_openid = None
+        mock_user_repo.find_by_wechat_openid.return_value = None
+        mock_user_repo.find_by_wechat_unionid.return_value = sample_user
 
-        use_case = WechatSyncUseCase(
-            mock_user_repo,
-            session_store=mock_session_store,
-            jwt_secret_key="test-secret-key-for-jwt-12345",
+        resp, err = make_use_case(mock_user_repo, mock_session_store).execute(
+            WechatSyncRequest(openid="new_openid", unionid="unionid_456", nickname="测试")
         )
-        request = WechatSyncRequest(openid="openid_123")
-        use_case.execute(request)
+        assert err is None
+        assert resp.is_new_user is False
+        assert sample_user.wechat_openid == "new_openid"
 
+    def test_unionid_user_already_has_same_openid_no_extra_write(self, mock_user_repo, mock_session_store, sample_user):
+        """unionid 命中且 openid 已经是当前 openid，不额外改写"""
+        mock_user_repo.find_by_wechat_openid.return_value = sample_user
+        mock_user_repo.find_by_wechat_unionid.return_value = sample_user
+        saved = []
+        mock_user_repo.save.side_effect = lambda u: saved.append(u)
+        make_use_case(mock_user_repo, mock_session_store).execute(
+            WechatSyncRequest(openid="openid_123", unionid="unionid_456")
+        )
+        # 只有最后登录信息那一次 save，没有绑定/补写导致的额外 save
+        assert len(saved) == 1
+
+    def test_updates_last_login(self, mock_user_repo, mock_session_store, sample_user):
+        mock_user_repo.find_by_wechat_openid.return_value = sample_user
+        mock_user_repo.find_by_wechat_unionid.return_value = sample_user
+        make_use_case(mock_user_repo, mock_session_store).execute(WechatSyncRequest(openid="openid_123"))
         assert sample_user.last_login_at is not None
         assert sample_user.last_login_ip == "bff_gateway"
 
     def test_returns_tokens(self, mock_user_repo, mock_session_store, sample_user):
-        """返回 access_token 和 refresh_token"""
         mock_user_repo.find_by_wechat_openid.return_value = sample_user
-        mock_user_repo.save.return_value = sample_user
+        mock_user_repo.find_by_wechat_unionid.return_value = sample_user
+        resp, _ = make_use_case(mock_user_repo, mock_session_store).execute(WechatSyncRequest(openid="openid_123"))
+        assert resp.access_token and resp.refresh_token and resp.expires_in > 0
 
-        use_case = WechatSyncUseCase(
-            mock_user_repo,
-            session_store=mock_session_store,
-            jwt_secret_key="test-secret-key-for-jwt-12345",
+
+class TestWechatSyncConflicts:
+    def test_unionid_and_openid_bound_to_different_users(self, mock_user_repo, mock_session_store):
+        """unionid 与 openid 分属两个账号 → 冲突报错，不写库"""
+        ua = User(
+            id="ua",
+            email="a@wechat.local",
+            username="wxa",
+            display_name="A",
+            password_hash="h",
+            wechat_openid="o1",
+            wechat_unionid=None,
         )
-        request = WechatSyncRequest(openid="openid_123")
-        response, _ = use_case.execute(request)
+        ub = User(
+            id="ub",
+            email="b@wechat.local",
+            username="wxb",
+            display_name="B",
+            password_hash="h",
+            wechat_openid="oX",
+            wechat_unionid="un1",
+        )
+        mock_user_repo.find_by_wechat_openid.return_value = ua
+        mock_user_repo.find_by_wechat_unionid.return_value = ub
 
-        assert response.access_token is not None
-        assert len(response.access_token) > 0
-        assert response.refresh_token is not None
-        assert len(response.refresh_token) > 0
-        assert response.expires_in > 0
+        resp, err = make_use_case(mock_user_repo, mock_session_store).execute(
+            WechatSyncRequest(openid="o1", unionid="un1")
+        )
+        assert resp is None
+        assert "conflict" in err
+        # 补写不得发生
+        assert ua.wechat_unionid is None
+
+    def test_backfill_unionid_already_used_by_other(self, mock_user_repo, mock_session_store):
+        """给 openid 老账号补 unionid 时发现 unionid 已被他人占用 → 冲突"""
+        ua = User(
+            id="ua",
+            email="a@wechat.local",
+            username="wxa",
+            display_name="A",
+            password_hash="h",
+            wechat_openid="o1",
+            wechat_unionid=None,
+        )
+        ub = User(
+            id="ub",
+            email="b@wechat.local",
+            username="wxb",
+            display_name="B",
+            password_hash="h",
+            wechat_openid="o2",
+            wechat_unionid="un1",
+        )
+        # openid 命中 ua；unionid 首次查找（优先级查询）命中 ub
+        mock_user_repo.find_by_wechat_openid.return_value = ua
+        mock_user_repo.find_by_wechat_unionid.return_value = ub
+
+        resp, err = make_use_case(mock_user_repo, mock_session_store).execute(
+            WechatSyncRequest(openid="o1", unionid="un1")
+        )
+        assert resp is None
+        assert "conflict" in err
+        assert ua.wechat_unionid is None
+
+    def test_unionid_user_openid_belongs_to_other(self, mock_user_repo, mock_session_store):
+        """unionid 命中 ua，但请求的 openid 属于另一个账号 ub → 冲突，不抢占 openid"""
+        ua = User(
+            id="ua",
+            email="a@wechat.local",
+            username="wxa",
+            display_name="A",
+            password_hash="h",
+            wechat_openid="oA",
+            wechat_unionid="un1",
+        )
+        ub = User(
+            id="ub",
+            email="b@wechat.local",
+            username="wxb",
+            display_name="B",
+            password_hash="h",
+            wechat_openid="oB",
+            wechat_unionid=None,
+        )
+        mock_user_repo.find_by_wechat_openid.return_value = ub
+        mock_user_repo.find_by_wechat_unionid.return_value = ua
+
+        resp, err = make_use_case(mock_user_repo, mock_session_store).execute(
+            WechatSyncRequest(openid="oB", unionid="un1")
+        )
+        assert resp is None
+        assert "conflict" in err
+        assert ua.wechat_openid == "oA"  # 未被改写
 
 
-class TestWechatSyncUseCaseNewUser:
-    """新用户注册测试"""
-
+class TestWechatSyncNewUser:
     def test_create_new_user(self, mock_user_repo, mock_session_store):
-        """openid 和 unionid 都没找到，创建新用户"""
-        mock_user_repo.find_by_wechat_openid.return_value = None
-        mock_user_repo.find_by_wechat_unionid.return_value = None
-        mock_user_repo.find_by_username.return_value = None  # username 不重复
-
-        saved_user = None
-
-        def capture_save(user):
-            nonlocal saved_user
-            saved_user = user
-
-        mock_user_repo.save.side_effect = capture_save
-
-        use_case = WechatSyncUseCase(
-            mock_user_repo,
-            session_store=mock_session_store,
-            jwt_secret_key="test-secret-key-for-jwt-12345",
+        saved = {}
+        mock_user_repo.save.side_effect = lambda u: saved.update({u.id: u})
+        resp, err = make_use_case(mock_user_repo, mock_session_store).execute(
+            WechatSyncRequest(openid="new_openid_789", unionid="new_union_789", nickname="新用户")
         )
-        request = WechatSyncRequest(
-            openid="new_openid_789",
-            unionid="new_union_789",
-            nickname="新用户",
-            avatar_url="https://example.com/avatar.jpg",
-        )
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-        assert response.is_new_user is True
-        assert saved_user is not None
-        assert saved_user.wechat_openid == "new_openid_789"
-        assert saved_user.wechat_unionid == "new_union_789"
-        assert saved_user.email.endswith("@wechat.local")
-        assert saved_user.username.startswith("wx_")
-        assert saved_user.email_verified is True
-
-    def test_new_user_email_based_on_openid(self, mock_user_repo, mock_session_store):
-        """新用户邮箱基于 openid 生成"""
-        mock_user_repo.find_by_wechat_openid.return_value = None
-        mock_user_repo.find_by_wechat_unionid.return_value = None
-        mock_user_repo.find_by_username.return_value = None
-
-        saved_user = None
-
-        def capture_save(user):
-            nonlocal saved_user
-            saved_user = user
-
-        mock_user_repo.save.side_effect = capture_save
-
-        use_case = WechatSyncUseCase(
-            mock_user_repo,
-            session_store=mock_session_store,
-            jwt_secret_key="test-secret-key-for-jwt-12345",
-        )
-        request = WechatSyncRequest(openid="abcdef1234567890")
-        use_case.execute(request)
-
-        assert "abcdef1234567890" in saved_user.email or "abcdef1234567890"[:20] in saved_user.email
-        assert saved_user.email.endswith("@wechat.local")
+        assert err is None
+        assert resp.is_new_user is True
+        u = saved[resp.user_id]
+        assert u.wechat_openid == "new_openid_789"
+        assert u.wechat_unionid == "new_union_789"
+        assert u.email.endswith("@wechat.local")
+        assert u.username.startswith("wx_")
+        assert u.email_verified is True
+        assert u.password_hash
 
     def test_username_conflict_adds_suffix(self, mock_user_repo, mock_session_store):
-        """用户名冲突时加后缀"""
         call_count = [0]
 
-        def mock_find_by_username(username):
-            # 前两次返回存在（模拟冲突），第三次返回 None（可用）
+        def find_by_username(username):
             call_count[0] += 1
-            if call_count[0] <= 2:
-                return MagicMock()
-            return None
+            return MagicMock() if call_count[0] <= 2 else None
 
-        mock_user_repo.find_by_wechat_openid.return_value = None
-        mock_user_repo.find_by_wechat_unionid.return_value = None
-        mock_user_repo.find_by_username.side_effect = mock_find_by_username
-
-        use_case = WechatSyncUseCase(
-            mock_user_repo,
-            session_store=mock_session_store,
-            jwt_secret_key="test-secret-key-for-jwt-12345",
-        )
-        request = WechatSyncRequest(openid="test_openid")
-        response, error = use_case.execute(request)
-
-        assert error is None
-        assert response is not None
-        assert response.is_new_user is True
-        # find_by_username 被调用了多次（找不冲突的用户名）
-        assert mock_user_repo.find_by_username.call_count >= 2
-
-    def test_new_user_has_password_hash(self, mock_user_repo, mock_session_store):
-        """新用户有随机密码哈希（不能是空的）"""
-        mock_user_repo.find_by_wechat_openid.return_value = None
-        mock_user_repo.find_by_wechat_unionid.return_value = None
-        mock_user_repo.find_by_username.return_value = None
-
-        saved_user = None
-
-        def capture_save(user):
-            nonlocal saved_user
-            saved_user = user
-
-        mock_user_repo.save.side_effect = capture_save
-
-        use_case = WechatSyncUseCase(
-            mock_user_repo,
-            session_store=mock_session_store,
-            jwt_secret_key="test-secret-key-for-jwt-12345",
-        )
-        request = WechatSyncRequest(openid="new_openid")
-        use_case.execute(request)
-
-        assert saved_user.password_hash is not None
-        assert len(saved_user.password_hash) > 0
+        mock_user_repo.find_by_username.side_effect = find_by_username
+        resp, err = make_use_case(mock_user_repo, mock_session_store).execute(WechatSyncRequest(openid="test_openid"))
+        assert err is None
+        assert resp.is_new_user is True
+        assert call_count[0] >= 2
 
 
-class TestWechatSyncUseCaseErrors:
-    """错误场景测试"""
-
+class TestWechatSyncErrors:
     def test_empty_openid(self, mock_user_repo, mock_session_store):
-        """空 openid 返回错误"""
-        use_case = WechatSyncUseCase(
-            mock_user_repo,
-            session_store=mock_session_store,
-            jwt_secret_key="test-secret-key-for-jwt-12345",
-        )
-        request = WechatSyncRequest(openid="")
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert "openid is required" in error
+        resp, err = make_use_case(mock_user_repo, mock_session_store).execute(WechatSyncRequest(openid=""))
+        assert resp is None
+        assert "openid is required" in err
 
     def test_exception_returns_error(self, mock_user_repo, mock_session_store):
-        """异常时返回友好错误"""
         mock_user_repo.find_by_wechat_openid.side_effect = Exception("DB error")
-
-        use_case = WechatSyncUseCase(
-            mock_user_repo,
-            session_store=mock_session_store,
-            jwt_secret_key="test-secret-key-for-jwt-12345",
-        )
-        request = WechatSyncRequest(openid="openid_123")
-        response, error = use_case.execute(request)
-
-        assert response is None
-        assert "Internal error" in error
+        mock_user_repo.find_by_wechat_unionid.side_effect = Exception("DB error")
+        resp, err = make_use_case(mock_user_repo, mock_session_store).execute(WechatSyncRequest(openid="openid_123"))
+        assert resp is None
+        assert "Internal error" in err
 
 
 class TestWechatSyncSession:
-    """Session 相关测试"""
-
     def test_session_saved(self, mock_user_repo, mock_session_store, sample_user):
-        """登录时保存 session"""
         mock_user_repo.find_by_wechat_openid.return_value = sample_user
-        mock_user_repo.save.return_value = sample_user
-
-        use_case = WechatSyncUseCase(
-            mock_user_repo,
-            session_store=mock_session_store,
-            jwt_secret_key="test-secret-key-for-jwt-12345",
+        mock_user_repo.find_by_wechat_unionid.return_value = sample_user
+        make_use_case(mock_user_repo, mock_session_store).execute(
+            WechatSyncRequest(openid="openid_123", source="miniapp")
         )
-        request = WechatSyncRequest(openid="openid_123", source="miniapp")
-        use_case.execute(request)
-
         mock_session_store.save_session.assert_called_once()
-        call_kwargs = mock_session_store.save_session.call_args[1]
-        assert call_kwargs["user_id"] == "user_001"
-        assert "wechat_miniapp" in call_kwargs["device_info"]
-        assert call_kwargs["expires_in_seconds"] == 30 * 24 * 3600
+        kw = mock_session_store.save_session.call_args[1]
+        assert kw["user_id"] == "user_001"
+        assert "wechat_miniapp" in kw["device_info"]
+        assert kw["expires_in_seconds"] == 30 * 24 * 3600