refactor: 统一字体常量，消除 generate/constants.ts 与 components/title/constants.ts 的重复定义

- 删除 generate/constants.ts 中旧的 FONT_OPTIONS(string[])、FONT_FAMILY_MAP 和 getFontFamily
- 改为 re-export getFontFamily from @/components/title/constants（单一数据源）
- 更新 TitleMiniPreview/TitlePresetsGrid/FrontendPreviewPlayer 的 import 路径
- 修复 font-family CSS 值不一致问题（优设标题黑顺序、抖音美好体Bold/Sans、思源黑体Heavy Noto变体）

apps/api/app/api/routes/gpu_lipsync.py

@@ -93,7 +93,7 @@ def register_worker(
     svc: GpuLipsyncService = Depends(_get_svc),
     _token: str = Depends(_verify_gpu_token),
 ):
-    svc.register_worker(
+    worker, cancel_task = svc.register_worker(
         worker_id=body.worker_id,
         hostname=body.hostname,
         gpu_name=body.gpu_name,
@@ -101,7 +101,7 @@ def register_worker(
         capabilities=body.capabilities,
         task_id=body.task_id,
     )
-    return GpuWorkerRegisterResponse(ok=True, server_time=datetime.now(UTC), message="ok")
+    return GpuWorkerRegisterResponse(ok=True, server_time=datetime.now(UTC), message="ok", cancel_task=cancel_task)
 
 
 # ── GET /lipsync/poll — Worker 轮询拉任务 ─────────────────────────

apps/api/app/api/routes/lipsync.py

@@ -346,13 +346,13 @@ def cancel_lipsync_job(
     current_user: AuthenticatedUser = Depends(get_current_user),
     svc: LipsyncService = Depends(_get_service),
 ):
-    """取消对口型任务（仅 pending/tts_processing/submitted 状态可取消）."""
+    """取消对口型任务（仅 pending/tts_processing/submitted/processing 状态可取消）."""
     job = svc.cancel_job(job_id, current_user.user.id)
     if job is None:
         raise HTTPException(status_code=404, detail="任务不存在")
     if job.status != "cancelled":
         raise HTTPException(
             status_code=400,
-            detail=f"任务状态 {job.status} 不可取消，仅 pending/tts_processing/submitted 可取消",
+            detail=f"任务状态 {job.status} 不可取消，仅 pending/tts_processing/submitted/processing 可取消",
         )
     return job

apps/api/app/schemas/gpu_lipsync.py

@@ -36,6 +36,7 @@ class GpuWorkerRegisterResponse(BaseModel):
     ok: bool = True
     server_time: datetime
     message: str = "ok"
+    cancel_task: bool = Field(False, description="当前心跳任务是否已被用户取消；为 true 时 Worker 应终止推理")
 
 
 # ── 轮询任务 ────────────────────────────────────────────────────

apps/api/app/services/gpu_lipsync_service.py

@@ -50,13 +50,16 @@ class GpuLipsyncService:
         free_vram_mb: int = 0,
         capabilities: str = "musetalk",
         task_id: Optional[str] = None,
-    ) -> GpuWorkerModel:
+    ) -> tuple[GpuWorkerModel, bool]:
         """Worker 注册/心跳。
 
         task_id 非空时（Worker 推理期间的任务级心跳），同步把对应 processing
         任务的 last_heartbeat_at 续到当前时间，使长推理不会被
         ``_recover_timed_out_tasks`` 误回退。任务已结束 / 不属于该 worker
         （如已被超时回收重新派发）时忽略，不报错。
+
+        返回 ``(worker, cancel_task)``：当心跳任务已被用户取消时
+        ``cancel_task=True``，Worker 应尽快终止推理并释放 GPU。
         """
         now = datetime.now(UTC)
         worker = self.db.query(GpuWorkerModel).filter(GpuWorkerModel.worker_id == worker_id).one_or_none()
@@ -77,10 +80,11 @@ class GpuLipsyncService:
             worker.free_vram_mb = free_vram_mb
             worker.capabilities = capabilities or worker.capabilities
             worker.last_heartbeat_at = now
+        cancel_task = False
         if task_id:
-            self._touch_task_heartbeat(task_id, worker_id, now)
+            cancel_task = self._touch_task_heartbeat(task_id, worker_id, now)
         self.db.commit()
-        return worker
+        return worker, cancel_task
 
     # ── 轮询拉任务（Worker 调用） ──────────────────────────────────
 
@@ -166,6 +170,11 @@ class GpuLipsyncService:
             task.result_duration = duration_seconds or 0.0
             task.error_msg = ""
             task.finished_at = now
+        elif task.status == "cancelled":
+            # 用户已取消的任务，Worker 终止后上报失败，保持 cancelled 状态不回退
+            task.finished_at = now
+            task.error_msg = (error_msg or "用户取消")[:2000]
+            logger.info("GPU 任务 %s 已被用户取消，保持 cancelled 状态", task_id)
         else:
             # 失败：若仍可重试（已尝试次数 < MAX_ATTEMPTS）→ 回退 pending；否则 → failed
             if task.attempt < MAX_ATTEMPTS:
@@ -250,15 +259,21 @@ class GpuLipsyncService:
     def _result_key(self, task_id: str) -> str:
         return f"{self.RESULT_PREFIX}{task_id}.mp4"
 
-    def _touch_task_heartbeat(self, task_id: str, worker_id: str, now: datetime) -> None:
+    def _touch_task_heartbeat(self, task_id: str, worker_id: str, now: datetime) -> bool:
         """Worker 推理期间的任务级心跳：只刷新属于该 worker 且仍在 processing 的任务。
 
         任务不存在 / 已被超时回收重新派发 / 已完成 → 静默忽略（此时旧 worker 的
         结果上报会被结果接口按最终态处理）。
+
+        返回 ``cancel_task``：任务已被用户取消时为 True，Worker 应终止推理。
         """
         task = self.db.get(GpuLipsyncTaskModel, task_id)
         if task is None:
-            return
+            return False
+        # 任务已被用户取消 → 通知 Worker 终止推理
+        if task.status == "cancelled":
+            logger.info("任务心跳检测到已取消 task=%s worker=%s，通知 Worker 终止", task_id, worker_id)
+            return True
         if task.status != "processing" or task.worker_id != worker_id:
             logger.info(
                 "忽略过期任务心跳 task=%s worker=%s（status=%s owner=%s）",
@@ -267,10 +282,11 @@ class GpuLipsyncService:
                 task.status,
                 task.worker_id,
             )
-            return
+            return False
         task.last_heartbeat_at = now
         task.updated_at = now
         self.db.flush()
+        return False
 
     def _touch_worker(self, worker_id: str, now: datetime) -> None:
         if not worker_id:
@@ -371,9 +387,7 @@ class GpuLipsyncService:
             task = self.db.get(GpuLipsyncTaskModel, task_id)
             if task is None:
                 return None
-            if task.status == "done":
-                return task
-            if task.status == "failed":
+            if task.status in ("done", "failed", "cancelled"):
                 return task
             # pending/processing 继续等
             if time.monotonic() >= deadline:

apps/api/app/services/lipsync_service.py

@@ -783,12 +783,32 @@ class LipsyncService:
     # ── 取消任务 ──────────────────────────────────────────────────────────
 
     def cancel_job(self, job_id: str, user_id: str) -> Optional[LipsyncJobModel]:
-        """取消任务（仅 pending/tts_processing/submitted 状态可取消）."""
+        """取消任务（pending/tts_processing/submitted/processing 状态可取消）.
+
+        当 job 走 GPU 路径（mediakit_task_id 以 "gpu:" 开头）且状态为 processing 时，
+        同步将关联的 GpuLipsyncTask 标记为 cancelled，以便 Worker 心跳时检测到取消信号。
+        """
         job = self.get_job(job_id, user_id)
         if job is None:
             return None
 
-        if job.status in ("pending", "tts_processing", "submitted"):
+        if job.status in ("pending", "tts_processing", "submitted", "processing"):
+            # GPU 路径：同步标记关联的 GPU 任务为 cancelled
+            if job.status == "processing" and job.mediakit_task_id and job.mediakit_task_id.startswith("gpu:"):
+                gpu_task_id = job.mediakit_task_id[4:]  # 去掉 "gpu:" 前缀
+                try:
+                    from packages.adapters.sqlalchemy_impl.models import GpuLipsyncTaskModel
+
+                    gpu_task = self.db.get(GpuLipsyncTaskModel, gpu_task_id)
+                    if gpu_task and gpu_task.status == "processing":
+                        gpu_task.status = "cancelled"
+                        gpu_task.error_msg = "用户取消"
+                        gpu_task.updated_at = datetime.now(UTC)
+                        gpu_task.finished_at = datetime.now(UTC)
+                        logger.info("GPU 任务 %s 已被用户取消（通过 job_id=%s）", gpu_task_id, job_id)
+                except Exception as exc:
+                    logger.warning("标记 GPU 任务取消失败（不影响 job 取消）: %s", exc)
+
             job.status = "cancelled"
             job.updated_at = datetime.now(UTC)
             self.db.commit()

apps/api/app/tasks/lipsync_gpu.py

@@ -98,6 +98,14 @@ def lipsync_gpu_process_async(self, job_id: str, user_id: str, gpu_task_id: str)
             _fallback_to_mediakit(db, job)
             return
 
+        if final_task.status == "cancelled":
+            # 用户已取消任务，不回退 MediaKit，直接标记 job 为 cancelled
+            job.status = "cancelled"
+            job.updated_at = datetime.now(UTC)
+            db.commit()
+            logger.info("[lipsync_gpu_async] GPU 任务已被用户取消: job_id=%s", job_id)
+            return
+
         if final_task.status != "done":
             logger.warning(
                 "[lipsync_gpu_async] GPU 失败，回退 MediaKit: job_id=%s gpu_task=%s status=%s",

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

@@ -17,7 +17,7 @@ import type { EditPlanClip } from "@/api/template-editor"
 import { useSegmentScheduler, type PlaybackSegment } from "../hooks/useSegmentScheduler"
 import { usePreviewAudio } from "../hooks/usePreviewAudio"
 import { PreviewControls } from "./PreviewControls"
-import { getFontFamily } from "../constants"
+import { getFontFamily } from "@/components/title/constants"
 
 interface FrontendPreviewPlayerProps {
   assets: AssetItem[]

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

@@ -12,7 +12,7 @@
  */
 import React, { useEffect, useRef } from "react"
 import type { TitleSettings } from "../../types"
-import { getFontFamily } from "../../constants"
+import { getFontFamily } from "@/components/title/constants"
 
 interface Props {
   settings: TitleSettings

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

@@ -4,7 +4,7 @@
  * 避免 -webkit-text-stroke 在 Chromium 中吞掉填充色的问题
  */
 import React from "react"
-import { getFontFamily } from "../../constants"
+import { getFontFamily } from "@/components/title/constants"
 
 interface TitlePresetItem {
   key: string

apps/web/src/pages/generate/constants.ts

@@ -54,39 +54,8 @@ export const POSITION_OPTIONS = [
   { value: "custom", label: "自定义" },
 ]
 
-/* ── 标题字体选项（#2001：新增 4 款爆款字体） ── */
-export const FONT_OPTIONS = [
-  "优设标题黑",
-  "阿里普惠体Bold",
-  "抖音美好体",
-  "思源黑体Heavy",
-  "思源黑体",
-  "思源宋体",
-  "苹方",
-  "微软雅黑",
-  "楷体",
-]
-
-/* ── 标题字体 CSS font-family 映射（中文显示名 → 浏览器可识别的字体栈） ── */
-export const FONT_FAMILY_MAP: Record<string, string> = {
-  优设标题黑:
-    '"YouSheBiaoTiHei","YouShe Title Black","Source Han Sans SC Heavy","Noto Sans SC","PingFang SC",sans-serif',
-  阿里普惠体Bold:
-    '"Alibaba PuHuiTi Bold","Alibaba PuHuiTi","Source Han Sans SC","PingFang SC",sans-serif',
-  抖音美好体: '"Douyin Sans","DouyinSansBold","Source Han Sans SC Heavy","PingFang SC",sans-serif',
-  思源黑体Heavy:
-    '"Source Han Sans SC Heavy","Noto Sans SC Heavy","Source Han Sans CN Heavy","PingFang SC",sans-serif',
-  思源黑体: '"Source Han Sans SC", "Noto Sans SC", "PingFang SC", "Microsoft YaHei", sans-serif',
-  思源宋体: '"Source Han Serif SC", "Noto Serif SC", "Songti SC", "SimSun", serif',
-  苹方: '"PingFang SC", -apple-system, "Helvetica Neue", sans-serif',
-  PingFang: '"PingFang SC", -apple-system, "Helvetica Neue", sans-serif',
-  微软雅黑: '"Microsoft YaHei", "PingFang SC", sans-serif',
-  楷体: '"KaiTi", "STKaiti", "DFKai-SB", serif',
-}
-
-export function getFontFamily(font: string): string {
-  return FONT_FAMILY_MAP[font] || FONT_FAMILY_MAP["思源黑体"]
-}
+/* ── 标题字体：统一使用公共层定义（#2001） ── */
+export { getFontFamily } from "@/components/title/constants"
 
 /* ── 标题样式预设 ── */
 export const TITLE_PRESETS = [

deploy/gpu_worker/gpu_worker.py

@@ -108,11 +108,14 @@ def _check_musetalk_health() -> tuple[bool, dict]:
         return False, {"error": str(exc)}
 
 
-def _register(task_id: Optional[str] = None) -> bool:
+def _register(task_id: Optional[str] = None) -> tuple[bool, bool]:
     """向服务端注册 / 心跳，附带 GPU 信息。
 
     推理期间的心跳线程传 task_id：服务端会同步刷新该 processing 任务的
-    last_heartbeat_at，防止长推理被误判超时回收。
+    last_heartbeat_at，防止长推理被误判超时回收。同时服务端会检查该任务
+    是否已被用户取消，若是则返回 cancel_task=True。
+
+    返回 (ok, cancel_task)。
     """
     ok, info = _check_musetalk_health()
     if isinstance(info, dict):
@@ -142,12 +145,14 @@ def _register(task_id: Optional[str] = None) -> bool:
             timeout=15,
         )
         if r.status_code == 200:
-            return True
+            resp_body = r.json()
+            cancel_task = resp_body.get("cancel_task", False)
+            return True, cancel_task
         logger.error("注册/心跳失败: HTTP %d body=%s", r.status_code, r.text[:300])
-        return False
+        return False, False
     except Exception as exc:
         logger.error("注册/心跳异常: %s", exc)
-        return False
+        return False, False
 
 
 def _probe_gpu_name() -> str:
@@ -329,6 +334,9 @@ class TaskHeartbeat(threading.Thread):
     期间无法发送，服务端会因任务 last_heartbeat_at 停滞而误判超时回退 pending。
     本线程每 task_heartbeat_interval 秒（默认 30s）POST /gpu/register 并
     携带当前 task_id，让服务端持续续期任务心跳；任务处理结束 stop()。
+
+    同时检测服务端返回的 cancel_task 信号：若为 True，说明用户已取消任务，
+    立即调用 _cancel_musetalk() 终止本地推理，并设置 cancelled 标志供主流程检查。
     """
 
     def __init__(self, task_id: str, interval: float):
@@ -336,13 +344,21 @@ class TaskHeartbeat(threading.Thread):
         self.task_id = task_id
         self.interval = max(5.0, interval)
         self._stop_event = threading.Event()
+        self.cancelled = False  # 外部可读的取消标志
 
     def run(self) -> None:
         # 先立即发一次，再按间隔循环（首次心跳失败不影响主流程）
         while not self._stop_event.is_set():
             try:
-                if _register(self.task_id):
+                ok, cancel_task = _register(self.task_id)
+                if ok:
                     logger.debug("任务 %s 心跳已发送", self.task_id)
+                if cancel_task:
+                    logger.warning("任务 %s 已被用户取消，正在终止本地推理...", self.task_id)
+                    self.cancelled = True
+                    _cancel_musetalk()
+                    self._stop_event.set()
+                    return
             except Exception as exc:  # noqa: BLE001
                 logger.warning("任务 %s 心跳异常（忽略）: %s", self.task_id, exc)
             self._stop_event.wait(self.interval)
@@ -369,9 +385,17 @@ def _handle_task(task: dict) -> None:
             if not _download(task["video_url"], video_path):
                 _report_result(task_id, False, 0.0, "下载人物视频失败")
                 return
+            if hb.cancelled:
+                logger.info("任务 %s 在下载阶段被用户取消", task_id)
+                _report_result(task_id, False, 0.0, "用户取消任务")
+                return
             if not _download(task["audio_url"], audio_path):
                 _report_result(task_id, False, 0.0, "下载驱动音频失败")
                 return
+            if hb.cancelled:
+                logger.info("任务 %s 在下载阶段被用户取消", task_id)
+                _report_result(task_id, False, 0.0, "用户取消任务")
+                return
 
             # 2. 输入时长前置校验：短视频 MuseTalk 会 division by zero，
             #    直接上报 failed，不浪费 GPU 时间。ffprobe 不可用/读失败（0.0）
@@ -392,12 +416,20 @@ def _handle_task(task: dict) -> None:
             err = ""
             retryable = False
             for attempt in range(Config.task_max_retry + 1):
+                if hb.cancelled:
+                    logger.info("任务 %s 在推理前被用户取消", task_id)
+                    _report_result(task_id, False, 0.0, "用户取消任务")
+                    return
                 if attempt > 0:
                     logger.info("任务 %s 第 %d 次重试（瞬时错误）...", task_id, attempt + 1)
                     time.sleep(2)
                 success, duration, err, retryable = _call_musetalk(video_path, audio_path, out_path)
                 if success or not retryable:
                     break
+            if hb.cancelled:
+                logger.info("任务 %s 被用户取消（推理已终止）", task_id)
+                _report_result(task_id, False, 0.0, "用户取消任务")
+                return
             if not success:
                 logger.error("任务 %s 推理失败: %s", task_id, err)
                 _report_result(task_id, False, 0.0, err)
@@ -467,7 +499,8 @@ def main() -> int:
             # 心跳
             now = time.time()
             if now - last_heartbeat >= Config.heartbeat_interval:
-                if _register():
+                ok, _ = _register()
+                if ok:
                     last_heartbeat = now
 
             # 轮询任务

deploy/gpu_worker/musetalk_server.py

@@ -526,13 +526,12 @@ def _load_musetalk_models():
                         gfpgan_key = "params_ema" if "params_ema" in gfpgan_ckpt else "params"
                         gfpgan_model.load_state_dict(gfpgan_ckpt[gfpgan_key], strict=True)
                         gfpgan_model.eval()
-                        if Config.use_float16:
-                            gfpgan_model = gfpgan_model.half()
+                        # GFPGAN 始终使用 FP32 推理，避免 FP16 色偏导致紫/灰色块
                         gfpgan_model = gfpgan_model.to(device)
                         del gfpgan_ckpt
                         _gfpgan_loaded = True
                         _gfpgan_load_error = None
-                        logger.info("GFPGAN 加载完成 (FP16=%s)", Config.use_float16)
+                        logger.info("GFPGAN 加载完成 (FP32，避免色偏)")
                     else:
                         _gfpgan_loaded = False
                         _gfpgan_load_error = f"模型文件不存在: {gfpgan_path}"
@@ -920,25 +919,30 @@ def _run_inference(
                 _ff_proc.stdin.write(ori_frame.tobytes())
                 continue
 
-            # GFPGAN 人脸超分增强
+            # GFPGAN 人脸超分增强（FP32 推理，避免 FP16 色偏）
+            # 色彩通道约定：ori_frame / res_frame / _face_up 均为 BGR（OpenCV 默认）；
+            # GFPGAN 输出用 return_rgb=True 拿到 RGB，再转 BGR，与后续 face_parsing 融合保持一致。
             if gfpgan_enhancer is not None:
                 try:
                     _fh, _fw = res_frame_resized.shape[:2]
                     _face_up = cv2.resize(res_frame_resized, (512, 512),
                                           interpolation=cv2.INTER_LANCZOS4)
                     _face_rgb = cv2.cvtColor(_face_up, cv2.COLOR_BGR2RGB).astype(np.float32) / 255.0
-                    _face_t = torch.from_numpy(_face_rgb.transpose(2,0,1)).unsqueeze(0)
+                    _face_t = torch.from_numpy(_face_rgb.transpose(2, 0, 1)).unsqueeze(0)
+                    # GFPGAN 始终 FP32，避免 FP16 精度导致色偏；归一化到 [-1, 1]
                     _face_t = ((_face_t - 0.5) / 0.5).to(device)
-                    if Config.use_float16:
-                        _face_t = _face_t.half()
                     with torch.no_grad():
-                        _out = gfpgan_enhancer(_face_t, return_rgb=False, weight=0.5)[0]
-                    _out = _out.squeeze(0).float().cpu().clamp_(-1,1)
-                    _out = ((_out + 1)/2*255).numpy().transpose(1,2,0)
-                    _out_bgr = cv2.cvtColor(_out.astype(np.uint8), cv2.COLOR_RGB2BGR)
+                        _out = gfpgan_enhancer(_face_t, return_rgb=True, weight=0.35)[0]
+                    # 输出 tensor: RGB, [-1, 1] 范围 → clamp → 映射到 [0, 255] uint8
+                    _out = _out.squeeze(0).float().cpu().clamp_(-1.0, 1.0)
+                    _out = ((_out + 1.0) / 2.0 * 255.0).numpy().transpose(1, 2, 0)
+                    _out_rgb = _out.astype(np.uint8)
+                    # RGB → BGR，与 ori_frame 保持一致，确保 face_parsing 融合时通道正确
+                    _out_bgr = cv2.cvtColor(_out_rgb, cv2.COLOR_RGB2BGR)
+
                     res_frame_resized = cv2.resize(_out_bgr, (_fw, _fh),
                                                     interpolation=cv2.INTER_LANCZOS4)
-                    del _face_t, _out, _out_bgr
+                    del _face_t, _out, _out_rgb, _out_bgr
                 except Exception as _gfpgan_err:
                     logger.warning("GFPGAN 增强失败（帧 %d），使用原图: %s", i, _gfpgan_err)
 
@@ -985,7 +989,7 @@ def _run_inference(
     if video_downsampled:
         logger.info("输入视频 %.1f fps，降帧至 %.1f fps 推理后直接输出（不做插帧还原）", original_fps, inference_fps)
 
-    shutil.copy2(str(final_video), str(output_path))
+    _mux_video_with_audio(final_video, audio_path, output_path)
 
     try:
         torch.cuda.empty_cache()

packages/adapters/sqlalchemy_impl/session.py

@@ -18,6 +18,10 @@ def build_engine(
     pool_timeout: int = 30,
     pool_recycle: int = 3600,
 ):
+    # SQLite 不支持 QueuePool 的 pool_size/max_overflow/pool_timeout，
+    # 传了会在 create_engine 阶段直接 TypeError，这里只对非 SQLite 传连接池参数。
+    if _is_sqlite(database_url):
+        return create_engine(database_url, pool_recycle=pool_recycle)
     return create_engine(
         database_url,
         pool_size=pool_size,

tests/unit/test_1970_gpu_worker_heartbeat.py

@@ -68,7 +68,9 @@ def test_register_payload_includes_task_id_only_when_provided(worker, monkeypatc
 
     class _Resp:
         status_code = 200
-        text = ""
+
+        def json(self):
+            return {"worker_id": captured[-1]["worker_id"], "cancel_task": False}
 
     def _fake_post(url, json=None, headers=None, timeout=None):
         captured.append(json)
@@ -77,7 +79,9 @@ def test_register_payload_includes_task_id_only_when_provided(worker, monkeypatc
     monkeypatch.setattr(worker.requests, "post", _fake_post)
     monkeypatch.setattr(worker, "_check_musetalk_health", lambda: (True, {}))
 
-    assert worker._register("task-abc") is True
+    _ok, _cancel = worker._register("task-abc")
+    assert _ok is True
+    assert _cancel is False
     assert captured[-1]["task_id"] == "task-abc"
     assert captured[-1]["worker_id"]
 
@@ -93,7 +97,7 @@ def test_task_heartbeat_thread_sends_and_stops(worker, monkeypatch):
 
     def _fake_register(task_id=None):
         calls.append(task_id)
-        return True
+        return True, False
 
     monkeypatch.setattr(worker, "_register", _fake_register)
     hb = worker.TaskHeartbeat("task-hb1", interval=5)
@@ -105,6 +109,50 @@ def test_task_heartbeat_thread_sends_and_stops(worker, monkeypatch):
     assert calls and all(c == "task-hb1" for c in calls)
 
 
+def test_task_heartbeat_cancel_calls_musetalk_cancel(worker, monkeypatch):
+    """心跳响应 cancel_task=True → 调 _cancel_musetalk 并设置 cancelled 标志。"""
+    cancel_calls = []
+
+    monkeypatch.setattr(worker, "_register", lambda *a, **k: (True, True))
+    monkeypatch.setattr(worker, "_cancel_musetalk", lambda: cancel_calls.append(1))
+
+    hb = worker.TaskHeartbeat("task-cancel-1", interval=5)
+    hb.start()
+    hb.join(timeout=2)  # 检测到取消后线程自行 return
+    assert not hb.is_alive()
+    assert hb.cancelled is True
+    assert cancel_calls == [1]
+
+
+def test_handle_task_reports_cancelled_after_musetalk_abort(worker, monkeypatch):
+    """推理被 /cancel 终止后，hb.cancelled=True → 上报失败而非重试。"""
+    reports = []
+
+    monkeypatch.setattr(worker, "_register", lambda *a, **k: (True, False))
+    monkeypatch.setattr(worker, "_download", lambda url, path: True)
+    monkeypatch.setattr(worker, "_probe_duration", lambda path: 12.0)
+    # 模拟推理被终止（/inference 返回错误）
+    monkeypatch.setattr(worker, "_call_musetalk", lambda v, a, o: (False, 0.0, "推理被取消", False))
+    monkeypatch.setattr(
+        worker,
+        "_report_result",
+        lambda task_id, success, duration=0.0, error_msg="": reports.append(error_msg) or True,
+    )
+
+    # 让 TaskHeartbeat 在主线程检查时报告已取消
+    orig_hb_init = worker.TaskHeartbeat
+
+    def _hb(task_id, interval):
+        h = orig_hb_init(task_id, interval)
+        h.cancelled = True
+        return h
+
+    monkeypatch.setattr(worker, "TaskHeartbeat", _hb)
+
+    worker._handle_task({"task_id": "t-canceled", "video_url": "u", "audio_url": "u"})
+    assert reports == ["用户取消任务"]
+
+
 # ── 短视频前置拦截 ─────────────────────────────────────────────────
 
 
@@ -115,7 +163,7 @@ def test_handle_task_short_video_reports_failed_without_inference(worker, monkey
     audio.write_bytes(b"fake-audio")
     reports = []
 
-    monkeypatch.setattr(worker, "_register", lambda *a, **k: True)
+    monkeypatch.setattr(worker, "_register", lambda *a, **k: (True, False))
     monkeypatch.setattr(worker, "_download", lambda url, path: True)
     # ffprobe 读出 1.2s → 低于 3s 阈值
     monkeypatch.setattr(worker, "_probe_duration", lambda path: 1.2)
@@ -148,7 +196,7 @@ def test_handle_task_short_video_reports_failed_without_inference(worker, monkey
 def test_handle_task_probe_failure_does_not_block(worker, monkeypatch):
     """ffprobe 不可用（duration=0.0）时不能误杀，应继续推理."""
     reports = []
-    monkeypatch.setattr(worker, "_register", lambda *a, **k: True)
+    monkeypatch.setattr(worker, "_register", lambda *a, **k: (True, False))
     monkeypatch.setattr(worker, "_download", lambda url, path: True)
     monkeypatch.setattr(worker, "_probe_duration", lambda path: 0.0)
     monkeypatch.setattr(
@@ -214,7 +262,7 @@ def test_handle_task_retries_once_for_transient_then_succeeds(worker, monkeypatc
             return False, 0.0, "MuseTalk HTTP 503: busy", True
         return True, 6.5, "", False
 
-    monkeypatch.setattr(worker, "_register", lambda *a, **k: True)
+    monkeypatch.setattr(worker, "_register", lambda *a, **k: (True, False))
     monkeypatch.setattr(worker, "_download", lambda url, path: True)
     monkeypatch.setattr(worker, "_probe_duration", lambda path: 12.0)
     monkeypatch.setattr(worker, "_call_musetalk", _fake_call)
@@ -239,7 +287,7 @@ def test_handle_task_no_retry_for_deterministic_failure(worker, monkeypatch):
         return False, 0.0, "MuseTalk HTTP 400: bad input", False
 
     reports = []
-    monkeypatch.setattr(worker, "_register", lambda *a, **k: True)
+    monkeypatch.setattr(worker, "_register", lambda *a, **k: (True, False))
     monkeypatch.setattr(worker, "_download", lambda url, path: True)
     monkeypatch.setattr(worker, "_probe_duration", lambda path: 12.0)
     monkeypatch.setattr(worker, "_call_musetalk", _fake_call)

tests/unit/test_gpu_lipsync_routes.py

@@ -0,0 +1,89 @@
+"""GPU Worker 路由单测 — #2009 取消链路.
+
+直接调用路由函数（不经 HTTP 栈），显式注入 svc / _token 以跳过 Depends。
+CI 增量映射: gpu_lipsync.py (route) → test_gpu_lipsync_routes.py
+"""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+def _payload(**overrides):
+    from app.schemas.gpu_lipsync import GpuWorkerRegisterRequest
+
+    data = {
+        "worker_id": "w-1",
+        "hostname": "gpu-host",
+        "gpu_name": "RTX3060",
+        "free_vram_mb": 10000,
+        "capabilities": json.dumps({"musetalk": True}),
+    }
+    data.update(overrides)
+    return GpuWorkerRegisterRequest(**data)
+
+
+def test_register_returns_cancel_task_true_when_cancelled():
+    """心跳接口在任务已取消时必须把 cancel_task=True 透传给 Worker."""
+    fake_worker = MagicMock()
+    fake_worker.worker_id = "w-1"
+    fake_worker.hostname = "gpu-host"
+    fake_worker.gpu_name = "RTX3060"
+    fake_worker.free_vram_mb = 10000
+    fake_worker.capabilities = "musetalk"
+    fake_svc = MagicMock()
+    fake_svc.register_worker.return_value = (fake_worker, True)
+
+    from app.api.routes.gpu_lipsync import register_worker as route
+
+    resp = route(_payload(task_id="task-cancelled"), svc=fake_svc, _token="t")
+
+    assert resp.cancel_task is True
+    assert resp.ok is True
+    fake_svc.register_worker.assert_called_once()
+    kwargs = fake_svc.register_worker.call_args.kwargs
+    assert kwargs["task_id"] == "task-cancelled"
+
+
+def test_register_returns_cancel_task_false_normal():
+    """正常心跳 cancel_task=False."""
+    fake_worker = MagicMock()
+    fake_worker.worker_id = "w-1"
+    fake_worker.hostname = "gpu-host"
+    fake_worker.gpu_name = "RTX3060"
+    fake_worker.free_vram_mb = 10000
+    fake_worker.capabilities = "musetalk"
+    fake_svc = MagicMock()
+    fake_svc.register_worker.return_value = (fake_worker, False)
+
+    from app.api.routes.gpu_lipsync import register_worker as route
+
+    resp = route(_payload(), svc=fake_svc, _token="t")
+
+    assert resp.cancel_task is False
+
+
+def test_cancel_route_accepts_processing_status():
+    """cancel 路由允许 processing 状态（GPU 推理中），不再 400。"""
+    fake_job = MagicMock()
+    fake_job.status = "cancelled"
+
+    svc = MagicMock()
+    svc.cancel_job.return_value = fake_job
+
+    current_user = MagicMock()
+    current_user.user.id = "u1"
+
+    from app.api.routes.lipsync import cancel_lipsync_job as route
+
+    result = route("job-1", current_user, svc)
+
+    svc.cancel_job.assert_called_once_with("job-1", "u1")
+    assert result.status == "cancelled"
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-q"])

tests/unit/test_gpu_lipsync_service.py

@@ -173,10 +173,10 @@ def test_timed_out_task_is_redispatched(svc):
 
 
 def test_register_worker_creates_then_updates(svc):
-    w = svc.register_worker("w-1", hostname="pc1", gpu_name="RTX2060", free_vram_mb=3500)
+    w, _cancel = svc.register_worker("w-1", hostname="pc1", gpu_name="RTX2060", free_vram_mb=3500)
     assert w.worker_id == "w-1"
     assert w.gpu_name == "RTX2060"
-    w2 = svc.register_worker("w-1", hostname="pc1", gpu_name="RTX2060", free_vram_mb=2000)
+    w2, _cancel2 = svc.register_worker("w-1", hostname="pc1", gpu_name="RTX2060", free_vram_mb=2000)
     assert w2.free_vram_mb == 2000  # 更新
     assert w2.created_at == w.created_at  # 没新建
 
@@ -195,7 +195,7 @@ def test_register_with_task_id_refreshes_task_heartbeat(svc):
         {"last_heartbeat_at": old_hb - timedelta(seconds=300)}
     )
     svc.db.commit()
-    svc.register_worker("w-1", task_id=t.id)
+    _w, _c = svc.register_worker("w-1", task_id=t.id)
     svc.db.refresh(t)
     assert t.last_heartbeat_at > old_hb
     assert t.status == "processing"  # 心跳不改变状态
@@ -213,7 +213,7 @@ def test_register_task_heartbeat_ignores_finished_or_foreign_task(svc):
     svc.poll_task("w-1")
     done = svc.report_result(t.id, "w-1", success=True, duration_seconds=10.0)
     hb_when_done = done.last_heartbeat_at
-    svc.register_worker("w-1", task_id=t.id)
+    _w, _c = svc.register_worker("w-1", task_id=t.id)
     svc.db.refresh(t)
     assert t.status == "done"
     assert t.last_heartbeat_at == hb_when_done  # 没被改写
@@ -232,17 +232,61 @@ def test_register_task_heartbeat_ignores_finished_or_foreign_task(svc):
         {"last_heartbeat_at": owner_hb - timedelta(seconds=600)}
     )
     svc.db.commit()
-    svc.register_worker("w-1", task_id=t2.id)  # 旧 worker 迟到心跳
+    _w2, _c2 = svc.register_worker("w-1", task_id=t2.id)  # 旧 worker 迟到心跳
     svc.db.refresh(t2)
     assert t2.worker_id == "w-2"
     assert t2.status == "processing"
     assert t2.last_heartbeat_at == owner_hb
 
     # 场景 3：不存在的 task_id 不报错
-    svc.register_worker("w-1", task_id="nonexistent-id")
+    _wn, _cn = svc.register_worker("w-1", task_id="nonexistent-id")
     assert svc.db.get(GpuLipsyncTaskModel, "nonexistent-id") is None
 
 
+def test_register_task_heartbeat_detects_cancelled(svc):
+    """取消链路：任务已 cancelled 时，register 心跳必须返回 cancel_task=True."""
+    t = svc.create_task(video_url="v", audio_url="a")
+    svc.poll_task("w-1")
+    svc.db.refresh(t)
+    # 用户取消：直接把任务置为 cancelled
+    t.status = "cancelled"
+    t.finished_at = datetime.now(UTC)
+    svc.db.commit()
+
+    _w, cancel_task = svc.register_worker("w-1", task_id=t.id)
+    assert cancel_task is True
+    svc.db.refresh(t)
+    assert t.status == "cancelled"  # 心跳不改写已取消状态
+
+
+def test_report_result_cancelled_stays_cancelled(svc):
+    """Worker 终止取消任务后上报失败，report_result 必须保持 cancelled 不回退 pending."""
+    t = svc.create_task(video_url="v", audio_url="a")
+    svc.poll_task("w-1")
+    svc.db.refresh(t)
+    t.status = "cancelled"
+    svc.db.commit()
+
+    result = svc.report_result(t.id, "w-1", success=False, error_msg="推理被终止")
+    assert result.status == "cancelled"
+    assert result.finished_at is not None
+    assert "推理被终止" in (result.error_msg or "")
+
+
+def test_wait_for_result_returns_when_cancelled(svc):
+    """wait_for_result 将 cancelled 视为终态，立即返回，Celery 不回退 MediaKit."""
+    t = svc.create_task(video_url="v", audio_url="a")
+    svc.poll_task("w-1")
+    svc.db.refresh(t)
+    t.status = "cancelled"
+    t.finished_at = datetime.now(UTC)
+    svc.db.commit()
+
+    result = svc.wait_for_result(t.id, timeout_seconds=5, poll_interval=0.1)
+    assert result is not None
+    assert result.status == "cancelled"
+
+
 def test_default_gpu_task_timeout_is_900(svc):
     """#1970 默认超时 300→900，覆盖 RTX2060 长视频推理."""
     assert svc.settings.gpu_task_timeout_seconds == 900

tests/unit/test_lipsync_gpu_async_task.py

@@ -248,3 +248,41 @@ class TestSignMediaUrl:
         with patch.object(task_mod, "get_shared_storage_service", side_effect=RuntimeError("x")):
             url = "https://own-bucket.oss-cn-beijing.aliyuncs.com/a.wav"
             assert task_mod._sign_media_url(url) == url
+
+
+def test_cancelled_gpu_task_does_not_fallback_mediakit(monkeypatch):
+    """GPU 任务被用户取消 → Celery 任务直接标记 cancelled，不回退 MediaKit。"""
+    job = MagicMock()
+    job.id = "job-1"
+    job.status = "processing"
+    job.mediakit_task_id = "gpu:gpu-task-1"
+
+    gpu_task = MagicMock()
+    gpu_task.status = "cancelled"
+    gpu_task.error_msg = "用户取消"
+
+    fake_gpu_svc = MagicMock()
+    fake_gpu_svc.wait_for_result.return_value = gpu_task
+    gpu_service_cls = MagicMock(return_value=fake_gpu_svc)
+
+    fake_db = MagicMock()
+    fake_db.query.return_value.filter_by.return_value.first.return_value = job
+
+    # 直接替换 sys.modules 里的 gpu_lipsync_service 模块（全量跑时它可能已被
+    # 其他测试换成 MagicMock），保证任务函数内 from...import 一定拿到我们的类；
+    # 并替换 _get_db_session 绕开 worker_app / app.db 两条 import 分支。
+    import sys
+    from types import SimpleNamespace
+
+    fake_mod = SimpleNamespace(GpuLipsyncService=gpu_service_cls)
+    monkeypatch.setitem(sys.modules, "app.services.gpu_lipsync_service", fake_mod)
+    monkeypatch.setattr(task_mod, "_get_db_session", lambda: fake_db)
+    monkeypatch.setattr(task_mod, "logger", MagicMock())
+
+    task_mod.lipsync_gpu_process_async.run("job-1", "u1", "gpu-task-1")
+
+    assert job.status == "cancelled"
+    assert not str(job.mediakit_task_id).startswith("mk-")
+    fake_db.commit.assert_called()
+    fake_gpu_svc.wait_for_result.assert_called_once()
+    gpu_service_cls.assert_called_once_with(fake_db)

tests/unit/test_lipsync_gpu_integration.py

@@ -323,3 +323,126 @@ class TestGpuServiceHelpers:
         svc = GpuLipsyncService(db=fake_db)
         fake_db.query.return_value.filter.return_value.first.return_value = None
         assert svc.has_available_worker() is False
+
+
+# ── cancel_job 取消链路 (#2009) ─────────────────────────────────────
+
+
+def _build_sqlite_session():
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
+
+    from packages.adapters.sqlalchemy_impl import models as _  # noqa: F401
+    from packages.adapters.sqlalchemy_impl.models import Base
+
+    engine = create_engine("sqlite:///:memory:", future=True)
+    Base.metadata.create_all(engine)
+    Session = sessionmaker(bind=engine, future=True)
+    return Session()
+
+
+def _make_real_job(db, *, status="processing", mediakit_task_id="gpu:gpu-task-1"):
+    import uuid
+
+    from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
+
+    job = LipsyncJobModel(
+        id=str(uuid.uuid4()),
+        user_id="u1",
+        project_id="p1",
+        video_url="videos/v.mp4",
+        audio_url="audios/a.wav",
+        enable_video_loop=True,
+        mediakit_task_id=mediakit_task_id,
+        status=status,
+    )
+    db.add(job)
+    db.commit()
+    return job
+
+
+def test_cancel_processing_gpu_job_marks_gpu_task_cancelled():
+    """processing 的 GPU job 取消时，关联 GpuLipsyncTask 必须同步置 cancelled."""
+    import uuid
+
+    from packages.adapters.sqlalchemy_impl.models import GpuLipsyncTaskModel
+
+    db = _build_sqlite_session()
+    gpu_task_id = str(uuid.uuid4())
+    gpu_task = GpuLipsyncTaskModel(
+        id=gpu_task_id,
+        video_url="v",
+        audio_url="a",
+        status="processing",
+        worker_id="w-1",
+        attempt=1,
+    )
+    db.add(gpu_task)
+    db.commit()
+
+    job = _make_real_job(db, mediakit_task_id=f"gpu:{gpu_task_id}")
+    from app.services.lipsync_service import LipsyncService
+
+    svc = LipsyncService(db=db, client=MagicMock())
+    result = svc.cancel_job(job.id, "u1")
+
+    assert result.status == "cancelled"
+    db.refresh(gpu_task)
+    assert gpu_task.status == "cancelled"
+    assert gpu_task.error_msg == "用户取消"
+    assert gpu_task.finished_at is not None
+
+
+def test_cancel_processing_gpu_job_skips_non_processing_gpu_task():
+    """GPU task 已不在 processing（如已 done）时，取消 job 不应改它，也不报错."""
+    import uuid
+
+    from packages.adapters.sqlalchemy_impl.models import GpuLipsyncTaskModel
+
+    db = _build_sqlite_session()
+    gpu_task_id = str(uuid.uuid4())
+    gpu_task = GpuLipsyncTaskModel(
+        id=gpu_task_id, video_url="v", audio_url="a", status="done", worker_id="w-1", attempt=1
+    )
+    db.add(gpu_task)
+    db.commit()
+
+    job = _make_real_job(db, mediakit_task_id=f"gpu:{gpu_task_id}")
+    from app.services.lipsync_service import LipsyncService
+
+    svc = LipsyncService(db=db, client=MagicMock())
+    result = svc.cancel_job(job.id, "u1")
+
+    assert result.status == "cancelled"
+    db.refresh(gpu_task)
+    assert gpu_task.status == "done"  # 没被动
+
+
+def test_cancel_processing_non_gpu_job_does_not_touch_gpu_table():
+    """mediakit_task_id 不是 gpu: 前缀（普通 MediaKit 任务）时，不查 GPU task."""
+    db = _build_sqlite_session()
+    job = _make_real_job(db, mediakit_task_id="mk-task-99")
+    from app.services.lipsync_service import LipsyncService
+
+    svc = LipsyncService(db=db, client=MagicMock())
+    result = svc.cancel_job(job.id, "u1")
+    assert result.status == "cancelled"
+
+
+def test_cancel_completed_job_unchanged():
+    """completed 状态不可取消，cancel_job 原样返回."""
+    db = _build_sqlite_session()
+    job = _make_real_job(db, status="completed", mediakit_task_id="gpu:x")
+    from app.services.lipsync_service import LipsyncService
+
+    svc = LipsyncService(db=db, client=MagicMock())
+    result = svc.cancel_job(job.id, "u1")
+    assert result.status == "completed"
+
+
+def test_cancel_job_not_found_returns_none():
+    db = _build_sqlite_session()
+    from app.services.lipsync_service import LipsyncService
+
+    svc = LipsyncService(db=db, client=MagicMock())
+    assert svc.cancel_job("nonexistent", "u1") is None