拆镜复刻、爆款开头复刻容灾优化

video-gen-api/app/api/v1/hot_opening_replicate.py

@@ -8,7 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.dependencies import get_current_user, get_db
 from app.models.user import User
-from app.enums.hot_opening_replicate import ModuleCodeEnum
+from app.enums.hot_opening_replicate import HotOpeningStepCodeEnum, ModuleCodeEnum
 from app.schemas.hot_opening_replicate import (
     HotOpeningActionOut,
     HotOpeningDeleteOut,
@@ -40,6 +40,11 @@ from app.services.hot_opening_replicate_service import (
     update_hot_opening_video_prompt_schema,
 )
 from app.services.module_generation_log_service import log_module_error
+from app.services.module_async_recovery_service import (
+    TASK_HOT_IMAGE_PROMPT,
+    TASK_HOT_VIDEO_PROMPT,
+    register_module_step_task,
+)
 from app.tasks.celery_app import celery_app
 
 MODULE = ModuleCodeEnum.HOT_OPENING_REPLICATE.value
@@ -431,8 +436,15 @@ async def generate_image_prompt(
 
     from app.tasks.hot_opening_replicate_tasks import start_image_prompt_optimize
 
+    await register_module_step_task(
+        module=MODULE,
+        project_id=project_id_value,
+        step_id=step_id_value,
+        step_code=HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
+        task_name=TASK_HOT_IMAGE_PROMPT,
+    )
     try:
-        start_image_prompt_optimize.delay(project_id_value, step_id_value)
+        start_image_prompt_optimize.apply_async(args=[project_id_value, step_id_value], queue="gen_chatapi_create", countdown=0)
     except Exception as exc:
         await _mark_dispatch_failed_and_raise(
             db,
@@ -560,8 +572,15 @@ async def generate_video_prompt(
 
     from app.tasks.hot_opening_replicate_tasks import start_video_prompt_optimize
 
+    await register_module_step_task(
+        module=MODULE,
+        project_id=project_id_value,
+        step_id=step_id_value,
+        step_code=HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
+        task_name=TASK_HOT_VIDEO_PROMPT,
+    )
     try:
-        start_video_prompt_optimize.delay(project_id_value, step_id_value)
+        start_video_prompt_optimize.apply_async(args=[project_id_value, step_id_value], queue="gen_chatapi_create", countdown=0)
     except Exception as exc:
         await _mark_dispatch_failed_and_raise(
             db,

video-gen-api/app/api/v1/shot_replicate.py

@@ -8,7 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.dependencies import get_current_user, get_db
 from app.models.user import User
-from app.enums.shot_replicate import ModuleCodeEnum
+from app.enums.shot_replicate import ModuleCodeEnum, ShotReplicateStepCodeEnum
 from app.schemas.shot_replicate import (
     ShotReplicateActionOut,
     ShotReplicateDeleteOut,
@@ -57,6 +57,14 @@ from app.services.shot_replicate_taskset_service import (
     task_set_detail,
 )
 from app.services.module_generation_log_service import log_module_error, log_module_event_file
+from app.services.module_async_recovery_service import (
+    TASK_SHOT_IMAGE_PROMPT,
+    TASK_SHOT_VIDEO_PROMPT,
+    register_module_step_task,
+    register_shot_segment_analysis_task,
+    register_shot_split_task,
+    register_shot_task_set_analysis_task,
+)
 from app.tasks.celery_app import celery_app
 
 MODULE = ModuleCodeEnum.SHOT_REPLICATE.value
@@ -141,6 +149,22 @@ def _log_api_exception_from_locals(exc: BaseException, local_values: dict, messa
         exc=exc,
     )
 
+
+
+def _ensure_celery_enabled(*, current_user: User | None = None, project_id: str | None = None, step_id: str | None = None) -> None:
+    if celery_app is not None:
+        return
+    message = "Celery未启用：请配置 REDIS_URL 或 CELERY_BROKER_URL 后启动 worker"
+    _log_api_error(
+        event_type="CELERY_DISABLED",
+        current_user=current_user,
+        project_id=project_id,
+        step_id=step_id,
+        message=message,
+        detail={"api": "shot_replicate"},
+    )
+    raise HTTPException(status_code=503, detail=message)
+
 async def _reload_project_detail(db: AsyncSession, current_user: User, project_id: str) -> ShotReplicateTaskDetailOut:
     project = await _get_project_for_user(
         db,
@@ -213,6 +237,7 @@ async def create_shot_task_set(
     current_user: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
+    _ensure_celery_enabled(current_user=current_user, project_id=locals().get("project_id") or locals().get("task_set_id"))
     try:
         task_set = await create_task_set(db, current_user=current_user, req=req)
         task_set_id = task_set.id
@@ -225,13 +250,12 @@ async def create_shot_task_set(
         _log_api_exception_from_locals(exc, locals(), f"创建拆镜总任务集失败: {exc}")
         raise HTTPException(status_code=500, detail=f"创建拆镜总任务集失败: {exc}")
 
-    if celery_app:
     try:
         from app.tasks.shot_replicate_tasks import analyze_original_video
 
+        await register_shot_task_set_analysis_task(task_set_id)
         analyze_original_video.apply_async(args=[task_set_id], queue="gen_chatapi_create", countdown=0)
     except Exception as exc:
-            # 分析任务投递失败时保留总任务，前端可稍后通过恢复/重试处理。
         _log_api_error(
             event_type="CELERY_DISPATCH_FAILED",
             current_user=current_user,
@@ -296,6 +320,7 @@ async def split_by_ai(
     current_user: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
+    _ensure_celery_enabled(current_user=current_user, project_id=locals().get("project_id") or locals().get("task_set_id"))
     try:
         out = await create_segments_by_ai(db, current_user=current_user, task_set_id=task_set_id, req=req)
         segment_ids = [item.id for item in out.segments]
@@ -308,10 +333,10 @@ async def split_by_ai(
         _log_api_exception_from_locals(exc, locals(), f"按 AI 建议拆镜失败: {exc}")
         raise HTTPException(status_code=500, detail=f"按 AI 建议拆镜失败: {exc}")
 
-    if celery_app:
     from app.tasks.shot_replicate_tasks import split_one_segment
 
     for segment_id in segment_ids:
+        await register_shot_split_task(segment_id, task_set_id=task_set_id)
         split_one_segment.apply_async(args=[segment_id], queue="gen_result_download", countdown=0)
     return out
 
@@ -327,6 +352,7 @@ async def split_custom(
     current_user: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
+    _ensure_celery_enabled(current_user=current_user, project_id=locals().get("project_id") or locals().get("task_set_id"))
     try:
         out = await create_custom_segment(db, current_user=current_user, task_set_id=task_set_id, req=req)
         segment_id = out.segment.id
@@ -339,9 +365,9 @@ async def split_custom(
         _log_api_exception_from_locals(exc, locals(), f"自定义拆镜失败: {exc}")
         raise HTTPException(status_code=500, detail=f"自定义拆镜失败: {exc}")
 
-    if celery_app:
     from app.tasks.shot_replicate_tasks import split_one_segment
 
+    await register_shot_split_task(segment_id, task_set_id=task_set_id)
     split_one_segment.apply_async(args=[segment_id], queue="gen_result_download", countdown=0)
     return out
 
@@ -520,6 +546,7 @@ async def generate_image_prompt(
     current_user: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
+    _ensure_celery_enabled(current_user=current_user, project_id=locals().get("project_id") or locals().get("task_set_id"))
     try:
         project, step = await submit_image_prompt_optimize(db, current_user=current_user, project_id=project_id, req=req)
         project_id_value, step_id_value = project.id, step.id
@@ -533,9 +560,15 @@ async def generate_image_prompt(
         raise HTTPException(status_code=500, detail=f"提交图片 AI 提词失败: {exc}")
 
     try:
-        if celery_app:
         from app.tasks.shot_replicate_flow_tasks import start_image_prompt_optimize
 
+        await register_module_step_task(
+            module=MODULE,
+            project_id=project_id_value,
+            step_id=step_id_value,
+            step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
+            task_name=TASK_SHOT_IMAGE_PROMPT,
+        )
         start_image_prompt_optimize.apply_async(args=[project_id_value, step_id_value], queue="gen_chatapi_create", countdown=0)
     except Exception as exc:
         await _mark_dispatch_failed_and_raise(db, current_user=current_user, project_id=project_id_value, step_id=step_id_value, message=f"图片 AI 提词任务投递失败: {exc}")
@@ -554,6 +587,7 @@ async def generate_image(
     current_user: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
+    _ensure_celery_enabled(current_user=current_user, project_id=locals().get("project_id") or locals().get("task_set_id"))
     try:
         project, step, chat_task = await generate_image_from_prompt(db, current_user=current_user, project_id=project_id, req=req)
         project_id_value, step_id_value, chat_task_id_value = project.id, step.id, chat_task.id
@@ -567,7 +601,6 @@ async def generate_image(
         raise HTTPException(status_code=500, detail=f"提交图片生成失败: {exc}")
 
     try:
-        if celery_app:
         from app.tasks.generation_create_tasks import chatapi_create_generation_task
 
         chatapi_create_generation_task.apply_async(args=[chat_task_id_value], queue="gen_chatapi_create", countdown=0)
@@ -588,6 +621,7 @@ async def generate_video_prompt(
     current_user: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
+    _ensure_celery_enabled(current_user=current_user, project_id=locals().get("project_id") or locals().get("task_set_id"))
     try:
         project, step = await submit_video_prompt_optimize(db, current_user=current_user, project_id=project_id, req=req)
         project_id_value, step_id_value = project.id, step.id
@@ -601,9 +635,15 @@ async def generate_video_prompt(
         raise HTTPException(status_code=500, detail=f"提交视频 AI 提词失败: {exc}")
 
     try:
-        if celery_app:
         from app.tasks.shot_replicate_flow_tasks import start_video_prompt_optimize
 
+        await register_module_step_task(
+            module=MODULE,
+            project_id=project_id_value,
+            step_id=step_id_value,
+            step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
+            task_name=TASK_SHOT_VIDEO_PROMPT,
+        )
         start_video_prompt_optimize.apply_async(args=[project_id_value, step_id_value], queue="gen_chatapi_create", countdown=0)
     except Exception as exc:
         await _mark_dispatch_failed_and_raise(db, current_user=current_user, project_id=project_id_value, step_id=step_id_value, message=f"视频 AI 提词任务投递失败: {exc}")
@@ -622,6 +662,7 @@ async def generate_video(
     current_user: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
+    _ensure_celery_enabled(current_user=current_user, project_id=locals().get("project_id") or locals().get("task_set_id"))
     try:
         project, step, chat_task = await generate_video_from_prompt(db, current_user=current_user, project_id=project_id, req=req)
         project_id_value, step_id_value, chat_task_id_value = project.id, step.id, chat_task.id
@@ -635,7 +676,6 @@ async def generate_video(
         raise HTTPException(status_code=500, detail=f"提交视频生成失败: {exc}")
 
     try:
-        if celery_app:
         from app.tasks.generation_create_tasks import chatapi_create_generation_task
 
         chatapi_create_generation_task.apply_async(args=[chat_task_id_value], queue="gen_chatapi_create", countdown=0)

video-gen-api/app/config.py

@@ -145,6 +145,17 @@ class Settings(BaseSettings):
     CELERY_STARTUP_RECOVERY_LOCK_KEY: str = "vg:celery:startup_recovery_lock"
     CELERY_STARTUP_RECOVERY_LOCK_TTL_SECONDS: int = 120
 
+    # 模块异步任务容灾配置。
+    # 覆盖 ModuleGenerationStep 提词任务、shot 原视频/片段分析、shot ffmpeg 切割 active 注册。
+    # 不新增 worker 队列：恢复扫描仍走 gen_result_download，真实业务任务回到原始队列。
+    MODULE_ASYNC_RECOVERY_BATCH_SIZE: int = 100
+    MODULE_ASYNC_QUEUE_TIMEOUT_SECONDS: int = 5 * 60
+    MODULE_ASYNC_LEASE_SECONDS: int = 10 * 60
+    MODULE_ASYNC_LOCK_TTL_SECONDS: int = 10 * 60
+    MODULE_ASYNC_REQUEUE_DELAY_SECONDS: int = 10
+    MODULE_ASYNC_ACTIVE_REDIS_HASH_KEY: str = "vg:celery:module_async:active"
+    MODULE_ASYNC_ACTIVE_REDIS_ZSET_KEY: str = "vg:celery:module_async:active_index"
+    MODULE_ASYNC_LOCK_KEY_PREFIX: str = "vg:celery:module_async:lock"
 
     RESOURCE_SIGN_SECRET: str = "resource-signature-secret-key-for-API-authentication"
     RESOURCE_SIGN_EXPIRE_SECONDS: int = 60

video-gen-api/app/services/module_async_recovery_service.py

@@ -0,0 +1,615 @@
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timedelta, timezone
+from typing import Any, Iterable
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import settings
+from app.enums.common import ModuleStepStatusEnum
+from app.enums.hot_opening_replicate import HotOpeningStepCodeEnum, ModuleCodeEnum as HotModuleCodeEnum
+from app.enums.shot_replicate import (
+    ModuleCodeEnum as ShotModuleCodeEnum,
+    ShotAnalysisStatusEnum,
+    ShotReplicateStepCodeEnum,
+    ShotSegmentAnalysisStatusEnum,
+    ShotSplitStatusEnum,
+)
+from app.models.module_generation_step import ModuleGenerationStep
+from app.models.shot_replicate_segment import ShotReplicateSegment
+from app.models.shot_replicate_task_set import ShotReplicateTaskSet
+from app.services.redis_registry_service import (
+    datetime_to_epoch,
+    redis_acquire_lock,
+    redis_get_due_registry_ids,
+    redis_get_registry_payloads,
+    redis_postpone_registry_item,
+    redis_release_lock,
+    redis_remove_registry_item,
+    redis_upsert_registry_item,
+    utc_now,
+)
+from app.tasks.celery_app import celery_app
+
+logger = logging.getLogger("video_gen")
+
+QUEUE_CREATE = "gen_chatapi_create"
+QUEUE_DOWNLOAD = "gen_result_download"
+
+OBJECT_MODULE_STEP = "module_step"
+OBJECT_SHOT_TASK_SET_ANALYSIS = "shot_task_set_analysis"
+OBJECT_SHOT_SEGMENT_ANALYSIS = "shot_segment_analysis"
+OBJECT_SHOT_SPLIT_SEGMENT = "shot_split_segment"
+
+TASK_HOT_IMAGE_PROMPT = "hot_opening.start_image_prompt_optimize"
+TASK_HOT_VIDEO_PROMPT = "hot_opening.start_video_prompt_optimize"
+TASK_SHOT_IMAGE_PROMPT = "shot_replicate.start_image_prompt_optimize"
+TASK_SHOT_VIDEO_PROMPT = "shot_replicate.start_video_prompt_optimize"
+TASK_SHOT_ANALYZE_ORIGINAL = "shot_replicate.analyze_original_video"
+TASK_SHOT_ANALYZE_CUSTOM_SEGMENT = "shot_replicate.analyze_custom_segment_video"
+TASK_SHOT_SPLIT_ONE = "shot_replicate.split_one_segment"
+
+HOT_MODULE = HotModuleCodeEnum.HOT_OPENING_REPLICATE.value
+SHOT_MODULE = ShotModuleCodeEnum.SHOT_REPLICATE.value
+
+TERMINAL_STEP_STATUSES = {
+    ModuleStepStatusEnum.COMPLETED.value,
+    ModuleStepStatusEnum.FAILED.value,
+    ModuleStepStatusEnum.CANCELLED.value,
+}
+TERMINAL_ANALYSIS_STATUSES = {
+    ShotAnalysisStatusEnum.COMPLETED.value,
+    ShotAnalysisStatusEnum.FAILED.value,
+}
+TERMINAL_SEGMENT_ANALYSIS_STATUSES = {
+    ShotSegmentAnalysisStatusEnum.COMPLETED.value,
+    ShotSegmentAnalysisStatusEnum.FAILED.value,
+    ShotSegmentAnalysisStatusEnum.NOT_REQUIRED.value,
+}
+TERMINAL_SPLIT_STATUSES = {
+    ShotSplitStatusEnum.COMPLETED.value,
+    ShotSplitStatusEnum.FAILED.value,
+}
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _ensure_aware(value: datetime | None) -> datetime | None:
+    if value is None:
+        return None
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)
+
+
+def _hash_key() -> str:
+    return settings.MODULE_ASYNC_ACTIVE_REDIS_HASH_KEY
+
+
+def _zset_key() -> str:
+    return settings.MODULE_ASYNC_ACTIVE_REDIS_ZSET_KEY
+
+
+def _lease_seconds() -> int:
+    return max(1, int(settings.MODULE_ASYNC_LEASE_SECONDS or 600))
+
+
+def _queue_timeout_seconds() -> int:
+    return max(1, int(settings.MODULE_ASYNC_QUEUE_TIMEOUT_SECONDS or 300))
+
+
+def _requeue_delay_seconds() -> int:
+    return max(1, int(settings.MODULE_ASYNC_REQUEUE_DELAY_SECONDS or 10))
+
+
+def _item_id(object_type: str, object_id: str) -> str:
+    return f"{object_type}:{object_id}"
+
+
+def _lock_key(object_type: str, object_id: str) -> str:
+    return f"{settings.MODULE_ASYNC_LOCK_KEY_PREFIX}:{object_type}:{object_id}"
+
+
+def _check_at_after(seconds: int | None = None) -> datetime:
+    return _now() + timedelta(seconds=int(seconds or _lease_seconds()))
+
+
+def _base_payload(
+    *,
+    object_type: str,
+    object_id: str,
+    task_name: str,
+    queue: str,
+    args: list[Any],
+    module: str | None = None,
+    project_id: str | None = None,
+    step_id: str | None = None,
+    step_code: str | None = None,
+    task_set_id: str | None = None,
+    segment_id: str | None = None,
+    reason: str = "submit",
+) -> dict[str, Any]:
+    now_epoch = datetime_to_epoch(utc_now())
+    return {
+        "object_type": object_type,
+        "object_id": object_id,
+        "task_name": task_name,
+        "queue": queue,
+        "args": args,
+        "module": module,
+        "project_id": project_id,
+        "step_id": step_id,
+        "step_code": step_code,
+        "task_set_id": task_set_id,
+        "segment_id": segment_id,
+        "reason": reason,
+        "created_at": now_epoch,
+        "updated_at": now_epoch,
+    }
+
+
+async def register_active_task(
+    *,
+    object_type: str,
+    object_id: str,
+    task_name: str,
+    queue: str,
+    args: list[Any],
+    module: str | None = None,
+    project_id: str | None = None,
+    step_id: str | None = None,
+    step_code: str | None = None,
+    task_set_id: str | None = None,
+    segment_id: str | None = None,
+    check_after_seconds: int | None = None,
+    reason: str = "submit",
+) -> str:
+    item_id = _item_id(object_type, object_id)
+    payload = _base_payload(
+        object_type=object_type,
+        object_id=object_id,
+        task_name=task_name,
+        queue=queue,
+        args=args,
+        module=module,
+        project_id=project_id,
+        step_id=step_id,
+        step_code=step_code,
+        task_set_id=task_set_id,
+        segment_id=segment_id,
+        reason=reason,
+    )
+    await redis_upsert_registry_item(
+        hash_key=_hash_key(),
+        zset_key=_zset_key(),
+        item_id=item_id,
+        payload=payload,
+        check_at=_check_at_after(check_after_seconds),
+        log_context="module_async_active",
+    )
+    return item_id
+
+
+async def register_module_step_task(
+    *,
+    module: str,
+    project_id: str,
+    step_id: str,
+    step_code: str,
+    task_name: str,
+    queue: str = QUEUE_CREATE,
+) -> str:
+    return await register_active_task(
+        object_type=OBJECT_MODULE_STEP,
+        object_id=step_id,
+        task_name=task_name,
+        queue=queue,
+        args=[project_id, step_id],
+        module=module,
+        project_id=project_id,
+        step_id=step_id,
+        step_code=step_code,
+        check_after_seconds=_lease_seconds(),
+    )
+
+
+async def register_shot_task_set_analysis_task(task_set_id: str) -> str:
+    return await register_active_task(
+        object_type=OBJECT_SHOT_TASK_SET_ANALYSIS,
+        object_id=task_set_id,
+        task_name=TASK_SHOT_ANALYZE_ORIGINAL,
+        queue=QUEUE_CREATE,
+        args=[task_set_id],
+        module=SHOT_MODULE,
+        project_id=task_set_id,
+        task_set_id=task_set_id,
+        check_after_seconds=_lease_seconds(),
+    )
+
+
+async def register_shot_segment_analysis_task(segment_id: str, *, task_set_id: str | None = None) -> str:
+    return await register_active_task(
+        object_type=OBJECT_SHOT_SEGMENT_ANALYSIS,
+        object_id=segment_id,
+        task_name=TASK_SHOT_ANALYZE_CUSTOM_SEGMENT,
+        queue=QUEUE_CREATE,
+        args=[segment_id],
+        module=SHOT_MODULE,
+        project_id=task_set_id,
+        task_set_id=task_set_id,
+        segment_id=segment_id,
+        check_after_seconds=_lease_seconds(),
+    )
+
+
+async def register_shot_split_task(segment_id: str, *, task_set_id: str | None = None) -> str:
+    return await register_active_task(
+        object_type=OBJECT_SHOT_SPLIT_SEGMENT,
+        object_id=segment_id,
+        task_name=TASK_SHOT_SPLIT_ONE,
+        queue=QUEUE_DOWNLOAD,
+        args=[segment_id],
+        module=SHOT_MODULE,
+        project_id=task_set_id,
+        task_set_id=task_set_id,
+        segment_id=segment_id,
+        check_after_seconds=max(_lease_seconds(), int(settings.SHOT_SPLIT_LEASE_SECONDS or 600)),
+    )
+
+
+async def remove_active_task(*, object_type: str, object_id: str) -> None:
+    await redis_remove_registry_item(
+        hash_key=_hash_key(),
+        zset_key=_zset_key(),
+        item_id=_item_id(object_type, object_id),
+        log_context="module_async_active",
+    )
+
+
+async def postpone_active_task(
+    *,
+    object_type: str,
+    object_id: str,
+    delay_seconds: int | None = None,
+    reason: str | None = None,
+) -> None:
+    item_id = _item_id(object_type, object_id)
+    payloads = await redis_get_registry_payloads(
+        hash_key=_hash_key(),
+        item_ids=[item_id],
+        log_context="module_async_active",
+    )
+    payload = payloads.get(item_id)
+    if payload is not None and reason:
+        payload["reason"] = reason
+    await redis_postpone_registry_item(
+        hash_key=_hash_key(),
+        zset_key=_zset_key(),
+        item_id=item_id,
+        payload=payload,
+        check_at=_check_at_after(delay_seconds or _requeue_delay_seconds()),
+        log_context="module_async_active",
+    )
+
+
+async def mark_active_started(*, object_type: str, object_id: str, reason: str = "started") -> None:
+    await postpone_active_task(
+        object_type=object_type,
+        object_id=object_id,
+        delay_seconds=_lease_seconds(),
+        reason=reason,
+    )
+
+
+async def acquire_object_lock(*, object_type: str, object_id: str) -> str | None:
+    return await redis_acquire_lock(
+        lock_key=_lock_key(object_type, object_id),
+        ttl_seconds=int(settings.MODULE_ASYNC_LOCK_TTL_SECONDS or _lease_seconds()),
+        log_context="module_async_object_lock",
+    )
+
+
+async def release_object_lock(*, object_type: str, object_id: str, token: str | None) -> None:
+    if token:
+        await redis_release_lock(
+            lock_key=_lock_key(object_type, object_id),
+            token=token,
+            log_context="module_async_object_lock",
+        )
+
+
+async def cleanup_active_if_terminal(db: AsyncSession, *, object_type: str, object_id: str) -> bool:
+    if object_type == OBJECT_MODULE_STEP:
+        result = await db.execute(
+            select(ModuleGenerationStep)
+            .where(ModuleGenerationStep.id == object_id)
+            .limit(1)
+        )
+        step = result.scalar_one_or_none()
+        if not step or step.deleted_at is not None or step.status in TERMINAL_STEP_STATUSES:
+            await remove_active_task(object_type=object_type, object_id=object_id)
+            return True
+        return False
+
+    if object_type == OBJECT_SHOT_TASK_SET_ANALYSIS:
+        result = await db.execute(
+            select(ShotReplicateTaskSet)
+            .where(ShotReplicateTaskSet.id == object_id)
+            .limit(1)
+        )
+        task_set = result.scalar_one_or_none()
+        if not task_set or task_set.deleted_at is not None or task_set.analysis_status in TERMINAL_ANALYSIS_STATUSES:
+            await remove_active_task(object_type=object_type, object_id=object_id)
+            return True
+        return False
+
+    if object_type == OBJECT_SHOT_SEGMENT_ANALYSIS:
+        result = await db.execute(
+            select(ShotReplicateSegment)
+            .where(ShotReplicateSegment.id == object_id)
+            .limit(1)
+        )
+        segment = result.scalar_one_or_none()
+        if not segment or segment.deleted_at is not None or segment.analysis_status in TERMINAL_SEGMENT_ANALYSIS_STATUSES:
+            await remove_active_task(object_type=object_type, object_id=object_id)
+            return True
+        return False
+
+    if object_type == OBJECT_SHOT_SPLIT_SEGMENT:
+        result = await db.execute(
+            select(ShotReplicateSegment)
+            .where(ShotReplicateSegment.id == object_id)
+            .limit(1)
+        )
+        segment = result.scalar_one_or_none()
+        if not segment or segment.deleted_at is not None or segment.split_status in TERMINAL_SPLIT_STATUSES:
+            await remove_active_task(object_type=object_type, object_id=object_id)
+            return True
+        return False
+
+    return False
+
+
+def _payload_args(payload: dict[str, Any]) -> list[Any]:
+    args = payload.get("args")
+    if isinstance(args, list):
+        return args
+    object_type = str(payload.get("object_type") or "")
+    if object_type == OBJECT_MODULE_STEP:
+        return [payload.get("project_id"), payload.get("step_id")]
+    if object_type == OBJECT_SHOT_TASK_SET_ANALYSIS:
+        return [payload.get("task_set_id") or payload.get("object_id")]
+    if object_type in {OBJECT_SHOT_SEGMENT_ANALYSIS, OBJECT_SHOT_SPLIT_SEGMENT}:
+        return [payload.get("segment_id") or payload.get("object_id")]
+    return []
+
+
+def _send_task(task_name: str, *, args: list[Any], queue: str, countdown: int = 0, priority: int | None = None) -> bool:
+    if celery_app is None:
+        return False
+    if not task_name or not queue:
+        return False
+    celery_app.send_task(
+        task_name,
+        args=args,
+        queue=queue,
+        countdown=max(0, int(countdown or 0)),
+        priority=priority if priority is not None else settings.DOWNLOAD_TASK_PRIORITY_RECOVER,
+    )
+    return True
+
+
+async def _recover_payload_from_redis(db: AsyncSession, item_id: str, payload: dict[str, Any]) -> str:
+    object_type = str(payload.get("object_type") or "")
+    object_id = str(payload.get("object_id") or "")
+    task_name = str(payload.get("task_name") or "")
+    queue = str(payload.get("queue") or QUEUE_CREATE)
+    args = _payload_args(payload)
+
+    if not object_type or not object_id or not task_name or not args:
+        await redis_remove_registry_item(hash_key=_hash_key(), zset_key=_zset_key(), item_id=item_id, log_context="module_async_active")
+        return "remove_invalid_payload"
+
+    if await cleanup_active_if_terminal(db, object_type=object_type, object_id=object_id):
+        return "remove_terminal"
+
+    if object_type == OBJECT_SHOT_SPLIT_SEGMENT:
+        from app.services.shot_replicate_recovery_service import recover_one_split_segment
+
+        result = await db.execute(
+            select(ShotReplicateSegment)
+            .where(ShotReplicateSegment.id == object_id, ShotReplicateSegment.deleted_at.is_(None))
+            .with_for_update(skip_locked=True)
+            .limit(1)
+        )
+        segment = result.scalar_one_or_none()
+        if not segment:
+            await remove_active_task(object_type=object_type, object_id=object_id)
+            return "remove_missing_split_segment"
+        action = await recover_one_split_segment(db, segment, source="redis_active")
+        if action.startswith("recover_"):
+            await postpone_active_task(object_type=object_type, object_id=object_id, delay_seconds=int(settings.SHOT_SPLIT_LEASE_SECONDS or _lease_seconds()), reason="redis_recovered")
+        elif action.startswith("skip_completed") or action.startswith("skip_failed") or action.startswith("mark_failed"):
+            await remove_active_task(object_type=object_type, object_id=object_id)
+        return f"split_{action}"
+
+    try:
+        _send_task(task_name, args=args, queue=queue, countdown=0, priority=settings.DOWNLOAD_TASK_PRIORITY_RECOVER)
+        await postpone_active_task(object_type=object_type, object_id=object_id, delay_seconds=_lease_seconds(), reason="redis_requeued")
+        return "redis_requeued"
+    except Exception as exc:
+        logger.exception("Redis active 恢复投递失败。item_id=%s, task=%s", item_id, task_name)
+        await postpone_active_task(object_type=object_type, object_id=object_id, delay_seconds=_requeue_delay_seconds(), reason=f"redis_requeue_failed:{exc}")
+        return "redis_requeue_failed"
+
+
+async def _recover_due_redis_items(db: AsyncSession, *, limit: int) -> dict[str, int]:
+    item_ids = await redis_get_due_registry_ids(
+        zset_key=_zset_key(),
+        limit=limit,
+        now=_now(),
+        log_context="module_async_active",
+    )
+    if not item_ids:
+        return {}
+    payloads = await redis_get_registry_payloads(hash_key=_hash_key(), item_ids=item_ids, log_context="module_async_active")
+    results: dict[str, int] = {}
+    for item_id in item_ids:
+        payload = payloads.get(item_id)
+        if not payload:
+            await redis_remove_registry_item(hash_key=_hash_key(), zset_key=_zset_key(), item_id=item_id, log_context="module_async_active")
+            action = "remove_missing_payload"
+        else:
+            action = await _recover_payload_from_redis(db, item_id, payload)
+        results[action] = results.get(action, 0) + 1
+    return results
+
+
+def _is_stale_datetime(value: datetime | None, *, seconds: int, now: datetime) -> bool:
+    checked = _ensure_aware(value)
+    if checked is None:
+        return True
+    return checked + timedelta(seconds=max(1, int(seconds))) <= now
+
+
+async def _recover_stale_module_steps(db: AsyncSession, *, limit: int) -> dict[str, int]:
+    now = _now()
+    stale_cutoff = now - timedelta(seconds=_lease_seconds())
+    result = await db.execute(
+        select(ModuleGenerationStep)
+        .where(
+            ModuleGenerationStep.deleted_at.is_(None),
+            ModuleGenerationStep.is_current == True,
+            ModuleGenerationStep.status == ModuleStepStatusEnum.PROCESSING.value,
+            ModuleGenerationStep.module.in_([HOT_MODULE, SHOT_MODULE]),
+            ModuleGenerationStep.step_code.in_(
+                [
+                    HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
+                    HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
+                    ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
+                    ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
+                ]
+            ),
+            ModuleGenerationStep.updated_at <= stale_cutoff,
+        )
+        .order_by(ModuleGenerationStep.updated_at.asc())
+        .limit(limit)
+        .with_for_update(skip_locked=True)
+    )
+    steps = list(result.scalars().all())
+    results: dict[str, int] = {}
+    for step in steps:
+        if step.module == HOT_MODULE and step.step_code == HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value:
+            task_name = TASK_HOT_IMAGE_PROMPT
+        elif step.module == HOT_MODULE and step.step_code == HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value:
+            task_name = TASK_HOT_VIDEO_PROMPT
+        elif step.module == SHOT_MODULE and step.step_code == ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value:
+            task_name = TASK_SHOT_IMAGE_PROMPT
+        elif step.module == SHOT_MODULE and step.step_code == ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value:
+            task_name = TASK_SHOT_VIDEO_PROMPT
+        else:
+            results["skip_unknown_step"] = results.get("skip_unknown_step", 0) + 1
+            continue
+
+        await register_module_step_task(
+            module=step.module,
+            project_id=step.project_id,
+            step_id=step.id,
+            step_code=step.step_code,
+            task_name=task_name,
+            queue=QUEUE_CREATE,
+        )
+        try:
+            _send_task(task_name, args=[step.project_id, step.id], queue=QUEUE_CREATE, countdown=0, priority=settings.DOWNLOAD_TASK_PRIORITY_RECOVER)
+            results["db_step_requeued"] = results.get("db_step_requeued", 0) + 1
+        except Exception:
+            logger.exception("DB fallback 恢复模块步骤失败。step_id=%s", step.id)
+            results["db_step_requeue_failed"] = results.get("db_step_requeue_failed", 0) + 1
+    await db.commit()
+    return results
+
+
+async def _recover_stale_shot_task_sets(db: AsyncSession, *, limit: int) -> dict[str, int]:
+    now = _now()
+    stale_cutoff = now - timedelta(seconds=_lease_seconds())
+    result = await db.execute(
+        select(ShotReplicateTaskSet)
+        .where(
+            ShotReplicateTaskSet.deleted_at.is_(None),
+            ShotReplicateTaskSet.analysis_status.in_([ShotAnalysisStatusEnum.PENDING.value, ShotAnalysisStatusEnum.PROCESSING.value]),
+            ShotReplicateTaskSet.updated_at <= stale_cutoff,
+        )
+        .order_by(ShotReplicateTaskSet.updated_at.asc())
+        .limit(limit)
+        .with_for_update(skip_locked=True)
+    )
+    task_sets = list(result.scalars().all())
+    results: dict[str, int] = {}
+    for task_set in task_sets:
+        await register_shot_task_set_analysis_task(task_set.id)
+        try:
+            _send_task(TASK_SHOT_ANALYZE_ORIGINAL, args=[task_set.id], queue=QUEUE_CREATE, countdown=0, priority=settings.DOWNLOAD_TASK_PRIORITY_RECOVER)
+            results["db_task_set_analysis_requeued"] = results.get("db_task_set_analysis_requeued", 0) + 1
+        except Exception:
+            logger.exception("DB fallback 恢复拆镜原视频分析失败。task_set_id=%s", task_set.id)
+            results["db_task_set_analysis_requeue_failed"] = results.get("db_task_set_analysis_requeue_failed", 0) + 1
+    await db.commit()
+    return results
+
+
+async def _recover_stale_shot_segment_analysis(db: AsyncSession, *, limit: int) -> dict[str, int]:
+    now = _now()
+    stale_cutoff = now - timedelta(seconds=_lease_seconds())
+    result = await db.execute(
+        select(ShotReplicateSegment)
+        .where(
+            ShotReplicateSegment.deleted_at.is_(None),
+            ShotReplicateSegment.segment_video_url.is_not(None),
+            ShotReplicateSegment.analysis_status.in_([ShotSegmentAnalysisStatusEnum.PENDING.value, ShotSegmentAnalysisStatusEnum.PROCESSING.value]),
+            ShotReplicateSegment.updated_at <= stale_cutoff,
+        )
+        .order_by(ShotReplicateSegment.updated_at.asc())
+        .limit(limit)
+        .with_for_update(skip_locked=True)
+    )
+    segments = list(result.scalars().all())
+    results: dict[str, int] = {}
+    for segment in segments:
+        await register_shot_segment_analysis_task(segment.id, task_set_id=segment.task_set_id)
+        try:
+            _send_task(TASK_SHOT_ANALYZE_CUSTOM_SEGMENT, args=[segment.id], queue=QUEUE_CREATE, countdown=0, priority=settings.DOWNLOAD_TASK_PRIORITY_RECOVER)
+            results["db_segment_analysis_requeued"] = results.get("db_segment_analysis_requeued", 0) + 1
+        except Exception:
+            logger.exception("DB fallback 恢复拆镜片段分析失败。segment_id=%s", segment.id)
+            results["db_segment_analysis_requeue_failed"] = results.get("db_segment_analysis_requeue_failed", 0) + 1
+    await db.commit()
+    return results
+
+
+def _merge_counts(target: dict[str, int], items: Iterable[tuple[str, int]]) -> None:
+    for key, value in items:
+        target[key] = target.get(key, 0) + int(value)
+
+
+async def recover_module_async_tasks_once(db: AsyncSession) -> dict[str, Any]:
+    """统一恢复模块异步任务。\n\n    覆盖范围：\n    - hot_opening / shot_replicate 的图片、视频 AI 提词步骤；\n    - shot_replicate 原视频分析；\n    - shot_replicate 自定义片段分析；\n    - shot_replicate split active 注册项。\n\n    拆镜 split 的 DB fallback 仍保留在 shot_replicate_recovery_service，\n    这里主要补 Redis active 恢复和非 split 类任务的 DB fallback。\n    """
+    batch_size = max(1, int(settings.MODULE_ASYNC_RECOVERY_BATCH_SIZE or 100))
+    results: dict[str, int] = {}
+
+    redis_results = await _recover_due_redis_items(db, limit=batch_size)
+    _merge_counts(results, redis_results.items())
+
+    step_results = await _recover_stale_module_steps(db, limit=batch_size)
+    _merge_counts(results, step_results.items())
+
+    task_set_results = await _recover_stale_shot_task_sets(db, limit=batch_size)
+    _merge_counts(results, task_set_results.items())
+
+    segment_results = await _recover_stale_shot_segment_analysis(db, limit=batch_size)
+    _merge_counts(results, segment_results.items())
+
+    return {"checked": sum(results.values()), "results": results}

video-gen-api/app/services/shot_replicate_recovery_service.py

@@ -11,6 +11,7 @@ from app.enums.shot_replicate import ShotSplitStatusEnum
 from app.models.shot_replicate_segment import ShotReplicateSegment
 from app.models.shot_replicate_task_set import ShotReplicateTaskSet
 from app.services.shot_replicate_taskset_service import refresh_task_set_split_summary
+from app.services.module_async_recovery_service import register_shot_split_task
 from app.tasks.celery_app import celery_app
 
 
@@ -81,6 +82,7 @@ async def recover_one_split_segment(db: AsyncSession, segment: ShotReplicateSegm
     await refresh_task_set_split_summary(db, segment.task_set_id)
     await db.commit()
 
+    await register_shot_split_task(segment.id, task_set_id=segment.task_set_id)
     if celery_app:
         split_one_segment.apply_async(
             args=[segment.id],

video-gen-api/app/tasks/__init__.py

@@ -4,6 +4,10 @@ Celery autodiscover imports ``app.tasks``; importing task modules here ensures
 custom named tasks are registered when workers start.
 """
 
+import logging
+
+logger = logging.getLogger("video_gen")
+
 try:
     from app.tasks import (  # noqa: F401
         generation_create_tasks,
@@ -12,7 +16,9 @@ try:
         generation_recovery_tasks,
         hot_opening_replicate_tasks,
         shot_replicate_tasks,
-        shot_replicate_flow_tasks
+        shot_replicate_flow_tasks,
+        module_async_recovery_tasks,
     )
 except Exception:
-    pass
+    logger.exception("Celery 任务模块导入失败，worker 可能出现 unregistered task。")
+    raise

video-gen-api/app/tasks/celery_app.py

@@ -59,6 +59,7 @@ if broker_url:
             "shot_replicate.recover_split_tasks_once": {"queue": "gen_result_download"},
             "generation.recover_download_tasks_once": {"queue": "gen_result_download"},
             "generation.recover_generation_tasks_once": {"queue": "gen_result_download"},
+            "module_async.recover_module_async_tasks_once": {"queue": "gen_result_download"},
             "app.tasks.cleanup.*": {"queue": "default"},
         },
     )
@@ -109,6 +110,7 @@ def on_worker_ready(sender=None, **kwargs):
             recover_generation_tasks_once,
         )
         from app.tasks.shot_replicate_tasks import recover_split_tasks_once
+        from app.tasks.module_async_recovery_tasks import recover_module_async_tasks_once_task
 
         countdown = max(0, int(settings.DOWNLOAD_RECOVERY_STARTUP_DELAY_SECONDS or 0))
 
@@ -127,6 +129,11 @@ def on_worker_ready(sender=None, **kwargs):
             queue="gen_result_download",
             priority=settings.DOWNLOAD_TASK_PRIORITY_RECOVER,
         )
+        recover_module_async_tasks_once_task.apply_async(
+            countdown=countdown + 15,
+            queue="gen_result_download",
+            priority=settings.DOWNLOAD_TASK_PRIORITY_RECOVER,
+        )
 
         logger.info("启动容灾恢复任务已投递。countdown=%s", countdown)
     except Exception:

video-gen-api/app/tasks/generation_create_tasks.py

@@ -1,10 +1,11 @@
 from app.tasks.async_runner import run_async
 import json
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from typing import Any, Optional
 
 from sqlalchemy import select
 
+from app.config import settings
 from app.models.base import async_session
 from app.models.chat_generation_task import ChatGenerationTask
 from app.services.error_codes import extract_error_message
@@ -16,6 +17,8 @@ from app.tasks.celery_app import celery_app
 
 ALLOWED_GENERATION_MODES = {"chatapi_async", "hot_opening_replicate", "shot_replicate"}
 
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
 
 def _get_first_value(obj: Any, *field_names: str) -> Optional[Any]:
     """
@@ -242,8 +245,13 @@ async def _run(task_id: str):
 
                 await enqueue_download_task(db, task, reason="create_result_ready")
             else:
-                from app.tasks.generation_poll_tasks import poll_generation_task
+                from app.tasks.generation_poll_tasks import poll_generation_task, register_poll_active
 
+                await register_poll_active(
+                    task,
+                    reason="create_provider_success",
+                    check_at=_now() + timedelta(seconds=int(settings.POLL_TASK_LEASE_SECONDS or 300)),
+                )
                 poll_generation_task.apply_async(
                     args=[task.id],
                     queue="gen_provider_poll",

video-gen-api/app/tasks/hot_opening_replicate_tasks.py

@@ -1,45 +1,95 @@
 from __future__ import annotations
 
+from typing import Any
+
+from app.enums.hot_opening_replicate import HotOpeningStepCodeEnum, ModuleCodeEnum
 from app.models.base import async_session
 from app.services.hot_opening_replicate_service import run_image_prompt_optimize, run_video_prompt_optimize
+from app.services.module_async_recovery_service import (
+    OBJECT_MODULE_STEP,
+    TASK_HOT_IMAGE_PROMPT,
+    TASK_HOT_VIDEO_PROMPT,
+    acquire_object_lock,
+    cleanup_active_if_terminal,
+    mark_active_started,
+    release_object_lock,
+    register_module_step_task,
+)
 from app.tasks.async_runner import run_async
 from app.tasks.celery_app import celery_app
 
+MODULE = ModuleCodeEnum.HOT_OPENING_REPLICATE.value
+
 
 async def _run_image_prompt(project_id: str, step_id: str | None = None):
+    lock_token: str | None = None
+    if step_id:
+        lock_token = await acquire_object_lock(object_type=OBJECT_MODULE_STEP, object_id=step_id)
+        if not lock_token:
+            # 同一个 step 的重复消息正在被其它 worker 处理：当前消息直接忽略，active registry 保留等待真正任务完成或 lease 过期。
+            return None
+        await register_module_step_task(
+            module=MODULE,
+            project_id=project_id,
+            step_id=step_id,
+            step_code=HotOpeningStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
+            task_name=TASK_HOT_IMAGE_PROMPT,
+        )
+        await mark_active_started(object_type=OBJECT_MODULE_STEP, object_id=step_id)
+    try:
         async with async_session() as db:
-        await run_image_prompt_optimize(db, project_id=project_id, step_id=step_id)
+            result = await run_image_prompt_optimize(db, project_id=project_id, step_id=step_id)
             await db.commit()
+            if step_id:
+                await cleanup_active_if_terminal(db, object_type=OBJECT_MODULE_STEP, object_id=step_id)
+            return result
+    finally:
+        if step_id:
+            await release_object_lock(object_type=OBJECT_MODULE_STEP, object_id=step_id, token=lock_token)
 
 
 async def _run_video_prompt(project_id: str, step_id: str | None = None):
+    lock_token: str | None = None
+    if step_id:
+        lock_token = await acquire_object_lock(object_type=OBJECT_MODULE_STEP, object_id=step_id)
+        if not lock_token:
+            return None
+        await register_module_step_task(
+            module=MODULE,
+            project_id=project_id,
+            step_id=step_id,
+            step_code=HotOpeningStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
+            task_name=TASK_HOT_VIDEO_PROMPT,
+        )
+        await mark_active_started(object_type=OBJECT_MODULE_STEP, object_id=step_id)
+    try:
         async with async_session() as db:
-        await run_video_prompt_optimize(db, project_id=project_id, step_id=step_id)
+            result = await run_video_prompt_optimize(db, project_id=project_id, step_id=step_id)
             await db.commit()
+            if step_id:
+                await cleanup_active_if_terminal(db, object_type=OBJECT_MODULE_STEP, object_id=step_id)
+            return result
+    finally:
+        if step_id:
+            await release_object_lock(object_type=OBJECT_MODULE_STEP, object_id=step_id, token=lock_token)
 
 
 if celery_app:
     @celery_app.task(name="hot_opening.start_image_prompt_optimize", bind=True, max_retries=3, default_retry_delay=30)
     def start_image_prompt_optimize(self, project_id: str, step_id: str | None = None):
-        """手动触发后的图片 AI 提词任务。
+        """手动触发后的图片 AI 提词任务。"""
-
-        该任务路由到现有 gen_chatapi_create 队列，不需要新增 hot_opening worker。
-        """
         return run_async(_run_image_prompt(project_id, step_id))
 
     @celery_app.task(name="hot_opening.start_video_prompt_optimize", bind=True, max_retries=3, default_retry_delay=30)
     def start_video_prompt_optimize(self, project_id: str, step_id: str | None = None):
-        """手动触发后的视频 AI 提词任务。
+        """手动触发后的视频 AI 提词任务。"""
-
-        该任务路由到现有 gen_chatapi_create 队列，不需要新增 hot_opening worker。
-        """
         return run_async(_run_video_prompt(project_id, step_id))
 else:
     class _DisabledTask:
-        def delay(self, *args, **kwargs):
+        def delay(self, *args: Any, **kwargs: Any):
             raise RuntimeError("Celery is disabled")
 
-        def apply_async(self, *args, **kwargs):
+        def apply_async(self, *args: Any, **kwargs: Any):
             raise RuntimeError("Celery is disabled")
 
     start_image_prompt_optimize = _DisabledTask()

video-gen-api/app/tasks/module_async_recovery_tasks.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from typing import Any
+
+from app.models.base import async_session
+from app.services.module_async_recovery_service import recover_module_async_tasks_once
+from app.tasks.async_runner import run_async
+from app.tasks.celery_app import celery_app
+
+
+async def _run_recover_module_async_tasks_once() -> dict[str, Any]:
+    async with async_session() as db:
+        return await recover_module_async_tasks_once(db)
+
+
+if celery_app:
+
+    @celery_app.task(name="module_async.recover_module_async_tasks_once")
+    def recover_module_async_tasks_once_task() -> dict[str, Any]:
+        return run_async(_run_recover_module_async_tasks_once())
+
+else:
+
+    class _DisabledTask:
+        def delay(self, *args: Any, **kwargs: Any) -> None:
+            raise RuntimeError("Celery is disabled")
+
+        def apply_async(self, *args: Any, **kwargs: Any) -> None:
+            raise RuntimeError("Celery is disabled")
+
+    recover_module_async_tasks_once_task = _DisabledTask()

video-gen-api/app/tasks/shot_replicate_flow_tasks.py

@@ -1,21 +1,76 @@
 from __future__ import annotations
 
+from typing import Any
+
+from app.enums.shot_replicate import ModuleCodeEnum, ShotReplicateStepCodeEnum
 from app.models.base import async_session
+from app.services.module_async_recovery_service import (
+    OBJECT_MODULE_STEP,
+    TASK_SHOT_IMAGE_PROMPT,
+    TASK_SHOT_VIDEO_PROMPT,
+    acquire_object_lock,
+    cleanup_active_if_terminal,
+    mark_active_started,
+    release_object_lock,
+    register_module_step_task,
+)
 from app.services.shot_replicate_flow_service import run_image_prompt_optimize, run_video_prompt_optimize
 from app.tasks.async_runner import run_async
 from app.tasks.celery_app import celery_app
 
+MODULE = ModuleCodeEnum.SHOT_REPLICATE.value
+
 
 async def _run_image_prompt(project_id: str, step_id: str | None = None):
+    lock_token: str | None = None
+    if step_id:
+        lock_token = await acquire_object_lock(object_type=OBJECT_MODULE_STEP, object_id=step_id)
+        if not lock_token:
+            return None
+        await register_module_step_task(
+            module=MODULE,
+            project_id=project_id,
+            step_id=step_id,
+            step_code=ShotReplicateStepCodeEnum.IMAGE_PROMPT_OPTIMIZE.value,
+            task_name=TASK_SHOT_IMAGE_PROMPT,
+        )
+        await mark_active_started(object_type=OBJECT_MODULE_STEP, object_id=step_id)
+    try:
         async with async_session() as db:
-        await run_image_prompt_optimize(db, project_id=project_id, step_id=step_id)
+            result = await run_image_prompt_optimize(db, project_id=project_id, step_id=step_id)
             await db.commit()
+            if step_id:
+                await cleanup_active_if_terminal(db, object_type=OBJECT_MODULE_STEP, object_id=step_id)
+            return result
+    finally:
+        if step_id:
+            await release_object_lock(object_type=OBJECT_MODULE_STEP, object_id=step_id, token=lock_token)
 
 
 async def _run_video_prompt(project_id: str, step_id: str | None = None):
+    lock_token: str | None = None
+    if step_id:
+        lock_token = await acquire_object_lock(object_type=OBJECT_MODULE_STEP, object_id=step_id)
+        if not lock_token:
+            return None
+        await register_module_step_task(
+            module=MODULE,
+            project_id=project_id,
+            step_id=step_id,
+            step_code=ShotReplicateStepCodeEnum.VIDEO_PROMPT_OPTIMIZE.value,
+            task_name=TASK_SHOT_VIDEO_PROMPT,
+        )
+        await mark_active_started(object_type=OBJECT_MODULE_STEP, object_id=step_id)
+    try:
         async with async_session() as db:
-        await run_video_prompt_optimize(db, project_id=project_id, step_id=step_id)
+            result = await run_video_prompt_optimize(db, project_id=project_id, step_id=step_id)
             await db.commit()
+            if step_id:
+                await cleanup_active_if_terminal(db, object_type=OBJECT_MODULE_STEP, object_id=step_id)
+            return result
+    finally:
+        if step_id:
+            await release_object_lock(object_type=OBJECT_MODULE_STEP, object_id=step_id, token=lock_token)
 
 
 if celery_app:
@@ -32,10 +87,10 @@ if celery_app:
 else:
 
     class _DisabledTask:
-        def delay(self, *args, **kwargs):
+        def delay(self, *args: Any, **kwargs: Any):
             raise RuntimeError("Celery is disabled")
 
-        def apply_async(self, *args, **kwargs):
+        def apply_async(self, *args: Any, **kwargs: Any):
             raise RuntimeError("Celery is disabled")
 
     start_image_prompt_optimize = _DisabledTask()

video-gen-api/app/tasks/shot_replicate_tasks.py

@@ -20,6 +20,20 @@ from app.models.shot_replicate_segment import ShotReplicateSegment
 from app.models.shot_replicate_task_set import ShotReplicateTaskSet
 from app.services.module_generation_log_service import log_module_error, log_module_event_file, log_module_prompt_event
 from app.services.redis_registry_service import redis_acquire_lock, redis_release_lock
+from app.services.module_async_recovery_service import (
+    OBJECT_SHOT_SEGMENT_ANALYSIS,
+    OBJECT_SHOT_SPLIT_SEGMENT,
+    OBJECT_SHOT_TASK_SET_ANALYSIS,
+    acquire_object_lock,
+    cleanup_active_if_terminal,
+    mark_active_started,
+    postpone_active_task,
+    register_shot_segment_analysis_task,
+    register_shot_split_task,
+    register_shot_task_set_analysis_task,
+    release_object_lock,
+    remove_active_task,
+)
 from app.services.shot_replicate_taskset_service import refresh_task_set_split_summary
 from app.services.shot_video_analysis_service import analyze_video_for_shot_split
 from app.services.shot_video_split_service import split_video_segment_async
@@ -65,6 +79,12 @@ async def _release_split_semaphore(lock_key: str | None, segment_id: str) -> Non
 
 
 async def _run_analyze_original_video(task_set_id: str) -> None:
+    lock_token = await acquire_object_lock(object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id)
+    if not lock_token:
+        return
+    await register_shot_task_set_analysis_task(task_set_id)
+    await mark_active_started(object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id)
+
     task_set_user_id: str | None = None
     video_url: str | None = None
     try:
@@ -77,8 +97,10 @@ async def _run_analyze_original_video(task_set_id: str) -> None:
             )
             task_set = result.scalar_one_or_none()
             if not task_set:
+                await remove_active_task(object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id)
                 return
             if task_set.analysis_status == ShotAnalysisStatusEnum.COMPLETED.value:
+                await remove_active_task(object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id)
                 return
             task_set_user_id = task_set.user_id
             video_url = task_set.video_url
@@ -107,6 +129,7 @@ async def _run_analyze_original_video(task_set_id: str) -> None:
             task_set = result.scalar_one_or_none()
             if not task_set:
                 await db.rollback()
+                await remove_active_task(object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id)
                 return
             result_json = analyzed.result
             task_set.original_video_content = str(result_json.get("原视频内容") or "无")
@@ -119,6 +142,7 @@ async def _run_analyze_original_video(task_set_id: str) -> None:
             task_set.status = ShotTaskSetStatusEnum.ANALYSIS_COMPLETED.value
             task_set.analysis_error_message = None
             await db.commit()
+            await cleanup_active_if_terminal(db, object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id)
 
         log_module_prompt_event(
             event_type="SHOT_ANALYSIS_SUCCESS",
@@ -154,6 +178,9 @@ async def _run_analyze_original_video(task_set_id: str) -> None:
                 task_set.analysis_status = ShotAnalysisStatusEnum.FAILED.value
                 task_set.analysis_error_message = str(exc)
                 await db.commit()
+                await cleanup_active_if_terminal(db, object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id)
+            else:
+                await remove_active_task(object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id)
         log_module_error(
             module=MODULE,
             event_type="SHOT_ANALYSIS_FAILED",
@@ -163,9 +190,17 @@ async def _run_analyze_original_video(task_set_id: str) -> None:
             detail={"task_set_id": task_set_id, "video_url": video_url, "analysis_mode": "full_breakdown"},
             exc=exc,
         )
+    finally:
+        await release_object_lock(object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id, token=lock_token)
 
 
 async def _run_analyze_custom_segment_video(segment_id: str) -> None:
+    lock_token = await acquire_object_lock(object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id)
+    if not lock_token:
+        return
+    await register_shot_segment_analysis_task(segment_id)
+    await mark_active_started(object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id)
+
     user_id: str | None = None
     task_set_id: str | None = None
     video_url: str | None = None
@@ -179,12 +214,15 @@ async def _run_analyze_custom_segment_video(segment_id: str) -> None:
             )
             segment = result.scalar_one_or_none()
             if not segment or not segment.segment_video_url:
+                await remove_active_task(object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id)
                 return
             if segment.analysis_status == ShotSegmentAnalysisStatusEnum.COMPLETED.value:
+                await remove_active_task(object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id)
                 return
             user_id = segment.user_id
             task_set_id = segment.task_set_id
             video_url = segment.segment_video_url
+            await register_shot_segment_analysis_task(segment_id, task_set_id=task_set_id)
             segment.analysis_status = ShotSegmentAnalysisStatusEnum.PROCESSING.value
             segment.analysis_error_message = None
             await db.commit()
@@ -210,6 +248,7 @@ async def _run_analyze_custom_segment_video(segment_id: str) -> None:
             segment = result.scalar_one_or_none()
             if not segment:
                 await db.rollback()
+                await remove_active_task(object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id)
                 return
             result_json = analyzed.result
             segment.original_video_content = str(result_json.get("原视频内容") or "无")
@@ -222,10 +261,11 @@ async def _run_analyze_custom_segment_video(segment_id: str) -> None:
             segment.analysis_status = ShotSegmentAnalysisStatusEnum.COMPLETED.value
             segment.analysis_error_message = None
             await db.commit()
+            await cleanup_active_if_terminal(db, object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id)
 
         log_module_prompt_event(
             event_type="SHOT_SEGMENT_ANALYSIS_SUCCESS",
-            project_id=task_set_id or segment_id,
+            project_id=task_set_id,
             step_id=segment_id,
             user_id=user_id or "",
             module=MODULE,
@@ -258,6 +298,9 @@ async def _run_analyze_custom_segment_video(segment_id: str) -> None:
                 segment.analysis_status = ShotSegmentAnalysisStatusEnum.FAILED.value
                 segment.analysis_error_message = str(exc)
                 await db.commit()
+                await cleanup_active_if_terminal(db, object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id)
+            else:
+                await remove_active_task(object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id)
         log_module_error(
             module=MODULE,
             event_type="SHOT_SEGMENT_ANALYSIS_FAILED",
@@ -268,6 +311,8 @@ async def _run_analyze_custom_segment_video(segment_id: str) -> None:
             detail={"segment_id": segment_id, "task_set_id": task_set_id, "video_url": video_url, "analysis_mode": "summary_only"},
             exc=exc,
         )
+    finally:
+        await release_object_lock(object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id, token=lock_token)
 
 
 async def _run_split_one_segment(segment_id: str) -> None:
@@ -280,6 +325,9 @@ async def _run_split_one_segment(segment_id: str) -> None:
     if not segment_lock_token:
         return
 
+    await register_shot_split_task(segment_id)
+    await mark_active_started(object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)
+
     semaphore_key: str | None = None
     user_id: str | None = None
     task_set_id: str | None = None
@@ -294,8 +342,15 @@ async def _run_split_one_segment(segment_id: str) -> None:
                 message="拆镜 ffmpeg 并发闸门已满，稍后重试",
                 detail={"segment_id": segment_id, "reason": "semaphore_full"},
             )
+            delay = max(1, int(settings.MODULE_ASYNC_REQUEUE_DELAY_SECONDS or 10))
+            await postpone_active_task(
+                object_type=OBJECT_SHOT_SPLIT_SEGMENT,
+                object_id=segment_id,
+                delay_seconds=delay,
+                reason="semaphore_full",
+            )
             if celery_app:
-                split_one_segment.apply_async(args=[segment_id], queue=SPLIT_QUEUE, countdown=10, priority=settings.DOWNLOAD_TASK_PRIORITY_NORMAL)
+                split_one_segment.apply_async(args=[segment_id], queue=SPLIT_QUEUE, countdown=delay, priority=settings.DOWNLOAD_TASK_PRIORITY_NORMAL)
             return
 
         async with async_session() as db:
@@ -307,6 +362,7 @@ async def _run_split_one_segment(segment_id: str) -> None:
             )
             segment = result.scalar_one_or_none()
             if not segment:
+                await remove_active_task(object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)
                 return
             user_id = segment.user_id
             task_set_id = segment.task_set_id
@@ -318,8 +374,10 @@ async def _run_split_one_segment(segment_id: str) -> None:
             )
             task_set = task_set_result.scalar_one_or_none()
             if not task_set:
+                await remove_active_task(object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)
                 return
             if segment.split_status == ShotSplitStatusEnum.COMPLETED.value and segment.segment_video_url:
+                await remove_active_task(object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)
                 return
 
             validate_split_range(
@@ -338,6 +396,8 @@ async def _run_split_one_segment(segment_id: str) -> None:
             task_set.status = ShotTaskSetStatusEnum.SPLITTING.value
             task_set.split_status = ShotSplitStatusEnum.PROCESSING.value
             await db.commit()
+            await register_shot_split_task(segment_id, task_set_id=segment.task_set_id)
+            await mark_active_started(object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)
 
             source_path = task_set.video_path
             date_dir = (segment.created_at or now).strftime("%Y/%m/%d")
@@ -379,6 +439,7 @@ async def _run_split_one_segment(segment_id: str) -> None:
             )
             segment = result.scalar_one_or_none()
             if not segment:
+                await remove_active_task(object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)
                 return
             segment.segment_video_url = split_result.url
             segment.segment_video_path = split_result.path
@@ -389,6 +450,7 @@ async def _run_split_one_segment(segment_id: str) -> None:
             segment.split_last_error = None
             await refresh_task_set_split_summary(db, segment.task_set_id)
             await db.commit()
+            await cleanup_active_if_terminal(db, object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)
 
             log_module_event_file(
                 module=MODULE,
@@ -407,6 +469,7 @@ async def _run_split_one_segment(segment_id: str) -> None:
             )
 
             if segment.source_mode == ShotSegmentSourceModeEnum.CUSTOM.value and celery_app:
+                await register_shot_segment_analysis_task(segment.id, task_set_id=segment.task_set_id)
                 analyze_custom_segment_video.apply_async(args=[segment.id], queue=ANALYSIS_QUEUE, countdown=0)
 
     except Exception as exc:
@@ -421,6 +484,7 @@ async def _run_split_one_segment(segment_id: str) -> None:
             )
             segment = result.scalar_one_or_none()
             if not segment:
+                await remove_active_task(object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)
                 return
             user_id = user_id or segment.user_id
             task_set_id = task_set_id or segment.task_set_id
@@ -437,9 +501,18 @@ async def _run_split_one_segment(segment_id: str) -> None:
             await refresh_task_set_split_summary(db, segment.task_set_id)
             await db.commit()
 
-            if segment.split_status == ShotSplitStatusEnum.RETRY_WAITING.value and celery_app:
+            if segment.split_status == ShotSplitStatusEnum.RETRY_WAITING.value:
                 next_retry_delay = max(1, int(((segment.split_next_retry_at or _now()) - _now()).total_seconds()))
+                await postpone_active_task(
+                    object_type=OBJECT_SHOT_SPLIT_SEGMENT,
+                    object_id=segment_id,
+                    delay_seconds=next_retry_delay,
+                    reason="split_retry_waiting",
+                )
+                if celery_app:
                     split_one_segment.apply_async(args=[segment_id], queue=SPLIT_QUEUE, countdown=next_retry_delay, priority=settings.DOWNLOAD_TASK_PRIORITY_RECOVER)
+            elif final_failed:
+                await cleanup_active_if_terminal(db, object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)
 
         log_module_error(
             module=MODULE,