项目/AI生成链路合并

video-gen-api/app/services/generation/ai/engine_service.py

@@ -50,7 +50,7 @@ def normalize_generation_count(value: int | None) -> int:
 
 
 async def get_image_engine(db: AsyncSession, engine_id: str | None) -> ImageEngine:
-    query = select(ImageEngine).where(ImageEngine.is_active == True)
+    query = select(ImageEngine).where(ImageEngine.is_active == True, ImageEngine.deleted_at.is_(None))
     if engine_id:
         query = query.where(ImageEngine.id == engine_id)
     else:
@@ -63,7 +63,7 @@ async def get_image_engine(db: AsyncSession, engine_id: str | None) -> ImageEngi
 
 
 async def get_video_engine(db: AsyncSession, engine_id: str | None) -> VideoEngine:
-    query = select(VideoEngine).where(VideoEngine.is_active == True)
+    query = select(VideoEngine).where(VideoEngine.is_active == True, VideoEngine.deleted_at.is_(None))
     if engine_id:
         query = query.where(VideoEngine.id == engine_id)
     else:

video-gen-api/app/services/generation/ai/image_batch_service.py

@@ -4,7 +4,7 @@ import json
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
-from uuid import uuid4
+from typing import Awaitable, Callable
 
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -18,6 +18,7 @@ from app.enums.generation_task import (
     GenerationType,
 )
 from app.models.chat_generation_task import ChatGenerationTask
+from app.services.generation.pipeline.db_lock_service import execute_with_lock_timeout
 from app.services.generation.ai.task_group_service import aggregate_main_task_status, load_children_map
 from app.services.generation.log_service import log_task_event
 from app.services.generation.provider_service import (
@@ -27,6 +28,7 @@ from app.services.generation.provider_service import (
 from app.services.generation.refund_service import mark_chat_generation_task_failed_and_refund_once
 from app.services.image_gen import ImageProviderError
 from app.services.operation_log_service import build_exception_detail, log_operation_event
+from app.services.redis_registry_service import RedisExecutionLockError
 from app.utils.id_gen import generate_id
 
 
@@ -84,8 +86,15 @@ def _task_snapshot(main: ChatGenerationTask) -> SimpleNamespace:
     )
 
 
-async def _claim_image_main_batch(db: AsyncSession, main_task_id: str) -> ImageBatchClaim:
-    result = await db.execute(
+async def _claim_image_main_batch(
+    db: AsyncSession,
+    main_task_id: str,
+    *,
+    execution_token: str,
+) -> ImageBatchClaim:
+    result = await execute_with_lock_timeout(
+        db,
+
         select(ChatGenerationTask)
         .where(
             ChatGenerationTask.id == main_task_id,
@@ -146,7 +155,7 @@ async def _claim_image_main_batch(db: AsyncSession, main_task_id: str) -> ImageB
         await db.commit()
         return ImageBatchClaim(False, main_task_id, reason="deadline_expired")
 
-    claim_token = uuid4().hex
+    claim_token = execution_token
     main.provider_create_claim_token = claim_token
     main.provider_create_started_at = now
     main.provider_create_lease_until = now + timedelta(seconds=IMAGE_PROVIDER_CLAIM_LEASE_SECONDS)
@@ -232,7 +241,9 @@ async def _fail_claimed_main(
         await db.rollback()
     except Exception:
         pass
-    result = await db.execute(
+    result = await execute_with_lock_timeout(
+        db,
+
         select(ChatGenerationTask)
         .where(
             ChatGenerationTask.id == main_task_id,
@@ -296,7 +307,9 @@ async def _split_children(
     provider_result: dict,
     provider_items: list[dict],
 ) -> list[str]:
-    result = await db.execute(
+    result = await execute_with_lock_timeout(
+        db,
+
         select(ChatGenerationTask)
         .where(
             ChatGenerationTask.id == main_task_id,
@@ -343,8 +356,12 @@ async def _split_children(
         index = int(item.get("generation_index") or 0)
         if index < 1 or index > expected_count:
             raise RuntimeError(f"无效的图片生成序号: {index}")
+        child_created_at = datetime.now(timezone.utc)
         child = ChatGenerationTask(
             id=generate_id(),
+            created_at=child_created_at,
+            resource_generation_started_at=child_created_at,
+            generation_attempt_no=1,
             user_id=main.user_id,
             original_prompt=main.original_prompt,
             optimized_prompt=main.optimized_prompt,
@@ -433,13 +450,23 @@ async def _enqueue_child_downloads(db: AsyncSession, child_ids: list[str]) -> di
     return {"enqueued": enqueued, "failed": failed}
 
 
-async def run_image_main_batch(db: AsyncSession, main_task: ChatGenerationTask) -> list[str]:
+async def run_image_main_batch(
+    db: AsyncSession,
+    main_task: ChatGenerationTask,
+    *,
+    execution_token: str,
+    execution_guard: Callable[[], Awaitable[None]],
+) -> list[str]:
     """单次同步组图，全部成功后原子拆分 child。
 
     绝不在组图 API 失败后退化为 N 次单图请求。
     """
     main_task_id = str(main_task.id)
-    claim = await _claim_image_main_batch(db, main_task_id)
+    claim = await _claim_image_main_batch(
+        db,
+        main_task_id,
+        execution_token=execution_token,
+    )
     if claim.existing_child_ids is not None:
         await _enqueue_child_downloads(db, claim.existing_child_ids)
         return claim.existing_child_ids
@@ -463,6 +490,7 @@ async def run_image_main_batch(db: AsyncSession, main_task: ChatGenerationTask)
             claim.runtime_engine,
             generation_count=generation_count,
         )
+        await execution_guard()
         provider_items = _validate_provider_batch(provider_result, generation_count)
         log_operation_event(
             domain="generation_ai_batch",
@@ -480,7 +508,10 @@ async def run_image_main_batch(db: AsyncSession, main_task: ChatGenerationTask)
                 "fallback_to_single_requests": False,
             },
         )
+    except RedisExecutionLockError:
+        raise
     except Exception as exc:
+        await execution_guard()
         message = exc.safe_message if isinstance(exc, ImageProviderError) else str(exc)
         await _fail_claimed_main(
             db,
@@ -493,6 +524,7 @@ async def run_image_main_batch(db: AsyncSession, main_task: ChatGenerationTask)
         return []
 
     try:
+        await execution_guard()
         child_ids = await _split_children(
             db,
             main_task_id=main_task_id,
@@ -500,7 +532,10 @@ async def run_image_main_batch(db: AsyncSession, main_task: ChatGenerationTask)
             provider_result=provider_result,
             provider_items=provider_items,
         )
+    except RedisExecutionLockError:
+        raise
     except Exception as exc:
+        await execution_guard()
         await _fail_claimed_main(
             db,
             main_task_id=main_task_id,

video-gen-api/app/services/generation/ai/service.py

@@ -95,12 +95,12 @@ async def list_generation_ai_engine_options(db: AsyncSession) -> GenerationAIEng
     """获取当前启用的图片/视频生成引擎，供前端创建任务时选择 engine_id。"""
     image_result = await db.execute(
         select(ImageEngine)
-        .where(ImageEngine.is_active == True)
+        .where(ImageEngine.is_active == True, ImageEngine.deleted_at.is_(None))
         .order_by(ImageEngine.priority.desc())
     )
     video_result = await db.execute(
         select(VideoEngine)
-        .where(VideoEngine.is_active == True)
+        .where(VideoEngine.is_active == True, VideoEngine.deleted_at.is_(None))
         .order_by(VideoEngine.priority.desc())
     )
 
@@ -274,7 +274,7 @@ def record_to_out(
         text_tokens_used=task.text_tokens_used or 0,
         image_tokens_used=task.image_tokens_used or 0,
         video_tokens_used=task.video_tokens_used or 0,
-        retry_count=task.retry_count or 0,
+        retry_count=task.manual_retry_count or 0,
         poll_count=task.poll_count or 0,
         error_message=task.error_message if is_main else _resolve_error_message(task.error_message),
         created_at=task.created_at,

video-gen-api/app/services/generation/ai/task_create_service.py

@@ -142,8 +142,16 @@ def _base_task_kwargs(
     credits_cost: float = 0.0,
     idempotency_key: str | None = None,
 ) -> dict[str, Any]:
+    resource_started_at = (
+        deadline_at - timedelta(minutes=int(settings.CHATAPI_ASYNC_IMAGE_DEADLINE_MINUTES or 30))
+        if gen_type == GenerationType.IMAGE.value
+        else deadline_at - timedelta(hours=int(settings.CHATAPI_ASYNC_VIDEO_FINAL_DEADLINE_HOURS or 24))
+    )
     return {
         "id": task_id,
+        "created_at": resource_started_at,
+        "resource_generation_started_at": resource_started_at,
+        "generation_attempt_no": 1,
         "user_id": user_id,
         "original_prompt": req.original_prompt,
         "gen_type": gen_type,
@@ -515,12 +523,10 @@ async def enqueue_created_generation_tasks(
 ) -> list[str]:
     """在业务事务提交后投递任务；返回投递失败的任务ID。
 
-    投递失败会在补偿事务中将对应任务置为失败并幂等退款，视频子任务
-    同时触发主任务状态汇总。调用方不应在初始事务提交前调用本函数。
+    投递失败时保留 queued 状态和已提交的计费/快照，交由 generation recovery
+    补投，避免 broker 短暂故障被误判为业务生成失败并提前退款。
     """
-    from app.services.generation.ai.task_group_service import aggregate_parent_for_child
     from app.services.generation.log_service import log_task_event
-    from app.services.generation.refund_service import mark_chat_generation_task_failed_and_refund_once
     from app.tasks.generation_create_tasks import chatapi_create_generation_task
 
     normalized_ids = list(dict.fromkeys(str(item) for item in task_ids if item))
@@ -530,6 +536,7 @@ async def enqueue_created_generation_tasks(
             ChatGenerationTask.user_id,
             ChatGenerationTask.parent_task_id,
             ChatGenerationTask.generation_index,
+            ChatGenerationTask.generation_attempt_no,
         ).where(ChatGenerationTask.id.in_(normalized_ids))
     ) if normalized_ids else None
     task_meta = {
@@ -537,6 +544,7 @@ async def enqueue_created_generation_tasks(
             "user_id": str(row.user_id),
             "parent_task_id": str(row.parent_task_id) if row.parent_task_id else None,
             "generation_index": row.generation_index,
+            "generation_attempt_no": int(row.generation_attempt_no or 1),
         }
         for row in (meta_result.all() if meta_result is not None else [])
     }
@@ -555,7 +563,11 @@ async def enqueue_created_generation_tasks(
             detail={"generation_index": meta.get("generation_index")},
         )
         try:
-            chatapi_create_generation_task.delay(task_id)
+            chatapi_create_generation_task.apply_async(
+                args=[task_id],
+                kwargs={"owner_type": "chat_generation_task", "generation_attempt_no": int(meta.get("generation_attempt_no") or 1)},
+                queue="gen_chatapi_create",
+            )
             await log_task_event(
                 task_id=task_id,
                 event_type="CHILD_ENQUEUE_SUCCESS",
@@ -575,33 +587,25 @@ async def enqueue_created_generation_tasks(
             )
         except Exception as exc:
             failed_ids.append(task_id)
-            await db.rollback()
-            failed_task = await mark_chat_generation_task_failed_and_refund_once(
-                db,
-                task_id=task_id,
-                error_message=f"任务队列投递失败: {exc}",
-                pipeline_stage="failed",
-            )
-            await aggregate_parent_for_child(db, failed_task)
-            await db.commit()
             await log_task_event(
                 task_id=task_id,
+                generation_attempt_no=int(meta.get("generation_attempt_no") or 1),
                 event_type="CHILD_ENQUEUE_FAILED",
-                to_status="failed",
-                to_stage="failed",
+                to_status="generating",
+                to_stage="queued",
                 message=str(exc),
-                detail={"task_id": task_id},
+                detail={"task_id": task_id, "recoverable": True},
             )
             log_operation_event(
                 domain="generation_ai_batch",
                 event_type="CHILD_ENQUEUE_FAILED",
                 event_status="failed",
                 source="api",
-                user_id=getattr(failed_task, "user_id", None),
-                group_id=getattr(failed_task, "parent_task_id", None) or task_id,
+                user_id=meta.get("user_id"),
+                group_id=meta.get("parent_task_id") or task_id,
                 task_id=task_id,
                 message=str(exc),
-                detail={"physical_files_deleted": False},
+                detail={"recoverable": True, "physical_files_deleted": False},
                 error=str(exc),
             )
     return failed_ids

video-gen-api/app/services/generation/ai/task_group_service.py

@@ -14,6 +14,7 @@ from app.enums.generation_task import (
     GenerationMode,
 )
 from app.models.chat_generation_task import ChatGenerationTask
+from app.services.generation.pipeline.db_lock_service import execute_with_lock_timeout
 from app.services.operation_log_service import log_operation_event
 from app.services.video_upscale.guard_service import assert_no_recoverable_failed_upscale_tasks
 from app.services.resource_accounting_service import (
@@ -157,7 +158,9 @@ async def aggregate_main_task_status(
     *,
     parent_task_id: str,
 ) -> ChatGenerationTask | None:
-    result = await db.execute(
+    result = await execute_with_lock_timeout(
+        db,
+
         select(ChatGenerationTask)
         .where(
             ChatGenerationTask.id == parent_task_id,
@@ -229,8 +232,10 @@ async def aggregate_main_task_status(
         main.text_tokens_used = sum(int(child.text_tokens_used or 0) for child in children)
         main.image_tokens_used = sum(int(child.image_tokens_used or 0) for child in children)
         main.video_tokens_used = sum(int(child.video_tokens_used or 0) for child in children)
-        main.retry_count = sum(int(child.retry_count or 0) for child in children)
+        # main 的手动重试次数只代表 main 自身，不能累加 child 的轮询/重试次数。
+        main.retry_count = int(main.manual_retry_count or 0)
         main.poll_count = sum(int(child.poll_count or 0) for child in children)
+        main.poll_error_count = sum(int(child.poll_error_count or 0) for child in children)
 
     await db.flush()
     log_operation_event(
@@ -272,7 +277,9 @@ async def soft_delete_child_tasks_batch(
     if not ids:
         return 0
     deleted_at = deleted_at or datetime.now(timezone.utc)
-    result = await db.execute(
+    result = await execute_with_lock_timeout(
+        db,
+
         select(ChatGenerationTask)
         .where(
             ChatGenerationTask.id.in_(ids),
@@ -374,7 +381,9 @@ async def soft_delete_top_level_task_group(
     deleted_at: datetime | None = None,
 ) -> int:
     deleted_at = deleted_at or datetime.now(timezone.utc)
-    result = await db.execute(
+    result = await execute_with_lock_timeout(
+        db,
+
         select(ChatGenerationTask)
         .where(
             ChatGenerationTask.id == task_id,

video-gen-api/app/services/generation/billing_service.py

@@ -558,6 +558,7 @@ async def charge_generation_media_for_record(
     project_name: str | None = None,
     description_prefix: str = "AI创作-",
     attempt_no: int | None = None,
+    engine_id: str | None = None,
 ) -> BillingSummary:
     return await charge_generation_media_by_params(
         db,
@@ -567,6 +568,7 @@ async def charge_generation_media_for_record(
         image_size=record.image_size,
         duration=record.duration,
         resolution=record.resolution,
+        engine_id=engine_id or getattr(record, "engine_id", None),
         project_name=project_name,
         description_prefix=description_prefix,
         owner_type=OWNER_GENERATION_RECORD,

video-gen-api/app/services/generation/download_service.py

@@ -1,12 +1,15 @@
 from __future__ import annotations
 
+import json
 import os
 import uuid
 from dataclasses import dataclass
 from datetime import datetime, timezone
+from typing import Awaitable, Callable
+from urllib.parse import urlparse
 
 from app.config import settings
-from app.models.chat_generation_task import ChatGenerationTask
+from app.services.generation.pipeline.owner_service import GenerationOwner
 from app.services.image_gen import download_image
 from app.services.provider_limit import provider_limit
 from app.services.resource_accounting_service import safe_file_size
@@ -34,7 +37,7 @@ def _to_aware_utc(value: datetime | None) -> datetime | None:
     return value.astimezone(timezone.utc)
 
 
-def _build_storage_date_dir(record: ChatGenerationTask) -> str:
+def _build_storage_date_dir(record: GenerationOwner) -> str:
     fixed = (getattr(record, "download_storage_date_dir", None) or "").strip().strip("/")
     if fixed:
         return fixed
@@ -42,6 +45,32 @@ def _build_storage_date_dir(record: ChatGenerationTask) -> str:
     return created_at.strftime("%Y/%m/%d")
 
 
+
+
+def _normalize_image_extension(record: GenerationOwner) -> str:
+    output_format = ""
+    try:
+        snapshot = json.loads(getattr(record, "engine_snapshot_json", None) or "{}")
+        if isinstance(snapshot, dict):
+            output_format = str(snapshot.get("output_format") or "").strip().lower()
+    except Exception:
+        output_format = ""
+    if output_format in {"jpg", "jpeg"}:
+        return "jpg"
+    if output_format in {"png", "webp"}:
+        return output_format
+
+    remote_url = str(getattr(record, "remote_result_url", None) or "")
+    try:
+        suffix = os.path.splitext(urlparse(remote_url).path or "")[1].lower().lstrip(".")
+    except Exception:
+        suffix = ""
+    if suffix in {"jpg", "jpeg"}:
+        return "jpg"
+    if suffix in {"png", "webp"}:
+        return suffix
+    return "jpg"
+
 def _make_part_path(final_path: str) -> str:
     return f"{final_path}.{uuid.uuid4().hex}.part"
 
@@ -65,16 +94,27 @@ def _safe_remove(path: str | None) -> None:
         pass
 
 
-async def _download_image_atomically(remote_url: str, final_path: str) -> str:
+async def _download_image_atomically(
+    remote_url: str,
+    final_path: str,
+    *,
+    execution_guard: Callable[[], Awaitable[None]] | None = None,
+) -> str:
     if _is_valid_file(final_path):
         return final_path
 
     os.makedirs(os.path.dirname(final_path), exist_ok=True)
     part_path = _make_part_path(final_path)
     try:
-        await download_image(remote_url, part_path)
+        await download_image(
+            remote_url,
+            part_path,
+            execution_guard=execution_guard,
+        )
         if not _is_valid_file(part_path):
             raise RuntimeError("图片下载完成但临时文件为空")
+        if execution_guard is not None:
+            await execution_guard()
         os.replace(part_path, final_path)
         return final_path
     except Exception:
@@ -82,16 +122,27 @@ async def _download_image_atomically(remote_url: str, final_path: str) -> str:
         raise
 
 
-async def _download_video_atomically(remote_url: str, final_path: str) -> str:
+async def _download_video_atomically(
+    remote_url: str,
+    final_path: str,
+    *,
+    execution_guard: Callable[[], Awaitable[None]] | None = None,
+) -> str:
     if _is_valid_file(final_path):
         return final_path
 
     os.makedirs(os.path.dirname(final_path), exist_ok=True)
     part_path = _make_part_path(final_path)
     try:
-        await download_video(remote_url, part_path)
+        await download_video(
+            remote_url,
+            part_path,
+            execution_guard=execution_guard,
+        )
         if not _is_valid_file(part_path):
             raise RuntimeError("视频下载完成但临时文件为空")
+        if execution_guard is not None:
+            await execution_guard()
         os.replace(part_path, final_path)
         return final_path
     except Exception:
@@ -101,7 +152,11 @@ async def _download_video_atomically(remote_url: str, final_path: str) -> str:
 
 
 
-async def download_video_upscale_source(record: ChatGenerationTask) -> DownloadedGenerationResult:
+async def download_video_upscale_source(
+    record: GenerationOwner,
+    *,
+    execution_guard: Callable[[], Awaitable[None]] | None = None,
+) -> DownloadedGenerationResult:
     """下载超分源视频。
 
     源视频只供后处理使用，不生成封面，也不作为用户 GeneratedResource。
@@ -119,10 +174,16 @@ async def download_video_upscale_source(record: ChatGenerationTask) -> Downloade
         part_path = build_part_mp4_path(dest)
         try:
             async with provider_limit("result_download", settings.RESULT_DOWNLOAD_MAX_CONCURRENCY):
-                await download_video(record.remote_result_url, part_path)
+                await download_video(
+                    record.remote_result_url,
+                    part_path,
+                    execution_guard=execution_guard,
+                )
             if not _is_valid_file(part_path):
                 raise RuntimeError("超分源视频下载完成但临时文件为空")
             await probe_video(part_path)
+            if execution_guard is not None:
+                await execution_guard()
             os.replace(part_path, dest)
         except Exception:
             _safe_remove(part_path)
@@ -138,7 +199,11 @@ async def download_video_upscale_source(record: ChatGenerationTask) -> Downloade
     )
 
 
-async def download_generation_result(record: ChatGenerationTask) -> DownloadedGenerationResult:
+async def download_generation_result(
+    record: GenerationOwner,
+    *,
+    execution_guard: Callable[[], Awaitable[None]] | None = None,
+) -> DownloadedGenerationResult:
     if not record.remote_result_url:
         raise ValueError("缺少远程结果URL")
 
@@ -147,13 +212,18 @@ async def download_generation_result(record: ChatGenerationTask) -> DownloadedGe
     if record.gen_type == "image":
         dest_dir = os.path.join(settings.STORAGE_IMAGE_LOCAL_PATH, date_dir)
         os.makedirs(dest_dir, exist_ok=True)
-        dest = os.path.join(dest_dir, f"{record.id}.png")
+        extension = _normalize_image_extension(record)
+        dest = os.path.join(dest_dir, f"{record.id}.{extension}")
 
         async with provider_limit("result_download", settings.RESULT_DOWNLOAD_MAX_CONCURRENCY):
-            await _download_image_atomically(record.remote_result_url if record.remote_result_url else "", dest)
+            await _download_image_atomically(
+                record.remote_result_url if record.remote_result_url else "",
+                dest,
+                execution_guard=execution_guard,
+            )
 
         return DownloadedGenerationResult(
-            url=f"/generate/images/{date_dir}/{record.id}.png",
+            url=f"/generate/images/{date_dir}/{record.id}.{extension}",
             storage_path=dest,
             file_size_bytes=safe_file_size(dest),
             resource_type="image",
@@ -164,14 +234,20 @@ async def download_generation_result(record: ChatGenerationTask) -> DownloadedGe
     dest = os.path.join(dest_dir, f"{record.id}.mp4")
 
     async with provider_limit("result_download", settings.RESULT_DOWNLOAD_MAX_CONCURRENCY):
-        await _download_video_atomically(record.remote_result_url if record.remote_result_url else "", dest)
+        await _download_video_atomically(
+            record.remote_result_url if record.remote_result_url else "",
+            dest,
+            execution_guard=execution_guard,
+        )
 
     cover_url, cover_storage_path = create_video_cover_for_local_video(
         record_id=record.id,
         video_path=dest,
         date_dir=date_dir,
-        log_prefix=f"ChatGenerationTask视频封面生成 task_id={record.id}",
+        log_prefix=f"生成资源视频封面 task_id={record.id}",
     )
+    if execution_guard is not None:
+        await execution_guard()
 
     return DownloadedGenerationResult(
         url=f"/generate/videos/{date_dir}/{record.id}.mp4",

video-gen-api/app/services/generation/history_delete_service.py

@@ -23,6 +23,7 @@ from app.models.module_generation_step import ModuleGenerationStep
 from app.models.shot_replicate_segment import ShotReplicateSegment
 from app.models.user import User
 from app.schemas.generation_ai import GenerationAIHistoryBatchDeleteOut
+from app.services.generation.pipeline.db_lock_service import execute_with_lock_timeout
 from app.services.generation.ai.task_group_service import soft_delete_child_tasks_batch
 from app.services.module_generation_flow_base_service import is_active_chat_generation_task
 from app.services.module_generation_log_service import log_module_event_file
@@ -105,7 +106,9 @@ async def _load_chat_tasks_for_steps(
     task_ids = _task_id_list(steps)
     if not task_ids:
         return {}
-    result = await db.execute(
+    result = await execute_with_lock_timeout(
+        db,
+
         select(ChatGenerationTask)
         .where(
             ChatGenerationTask.id.in_(task_ids),
@@ -194,7 +197,9 @@ async def _delete_generation_records(
     ids: list[str],
     deleted_at: datetime,
 ) -> GenerationAIHistoryBatchDeleteOut:
-    result = await db.execute(
+    result = await execute_with_lock_timeout(
+        db,
+
         select(GenerationRecord)
         .where(
             GenerationRecord.id.in_(ids),
@@ -238,7 +243,9 @@ async def _delete_chat_tasks(
     ids: list[str],
     deleted_at: datetime,
 ) -> GenerationAIHistoryBatchDeleteOut:
-    result = await db.execute(
+    result = await execute_with_lock_timeout(
+        db,
+
         select(ChatGenerationTask)
         .where(
             ChatGenerationTask.id.in_(ids),
@@ -325,7 +332,9 @@ async def _load_module_projects_by_ids(
     source: GenerationHistorySourceEnum,
     project_ids: list[str],
 ) -> list[ModuleGenerationProject]:
-    result = await db.execute(
+    result = await execute_with_lock_timeout(
+        db,
+
         select(ModuleGenerationProject)
         .where(
             ModuleGenerationProject.id.in_(project_ids),
@@ -351,7 +360,8 @@ async def _soft_delete_module_projects(
     if not project_ids:
         return [], [], 0
 
-    step_result = await db.execute(
+    step_result = await execute_with_lock_timeout(
+        db,
         select(ModuleGenerationStep)
         .where(
             ModuleGenerationStep.project_id.in_(project_ids),
@@ -452,7 +462,9 @@ async def _delete_shot_segments(
     ids: list[str],
     deleted_at: datetime,
 ) -> GenerationAIHistoryBatchDeleteOut:
-    result = await db.execute(
+    result = await execute_with_lock_timeout(
+        db,
+
         select(ShotReplicateSegment)
         .where(
             ShotReplicateSegment.id.in_(ids),

video-gen-api/app/services/generation/log_service.py

@@ -4,9 +4,13 @@ import hashlib
 import json
 from typing import Any
 
+from app.enums.generation_task import GenerationMode, GenerationOwnerType
 from app.models.base import async_session
+from app.models.chat_generation_task import ChatGenerationTask
 from app.models.chat_generation_task_event import ChatGenerationTaskEvent
 from app.models.chat_provider_call_log import ChatProviderCallLog
+from app.models.generation_record import GenerationRecord
+from app.services.operation_log_service import build_exception_detail, log_operation_event, sanitize_log_value
 from app.utils.id_gen import generate_id
 
 MAX_EXCERPT_CHARS = 2000
@@ -22,12 +26,9 @@ def _safe_json(data: Any) -> str | None:
 
 
 def _excerpt(data: Any, limit: int = MAX_EXCERPT_CHARS) -> str | None:
-    text = _safe_json(data)
+    text = _safe_json(sanitize_log_value(data))
     if text is None:
         return None
-    # Avoid storing secrets in logs.
-    text = text.replace("Authorization", "Authorization-REDACTED")
-    text = text.replace("api_key", "api_key_REDACTED")
     if len(text) > limit:
         return text[:limit] + "...[truncated]"
     return text
@@ -40,12 +41,74 @@ def _hash(data: Any) -> str | None:
     return hashlib.sha256(text.encode("utf-8")).hexdigest()
 
 
+def _owner_fields(
+    obj: Any | None,
+    *,
+    owner_type: str | None,
+    owner_id: str | None,
+    task_id: str | None,
+    record_id: str | None,
+    generation_attempt_no: int | None,
+    generation_mode: str | None,
+) -> dict[str, Any] | None:
+    if isinstance(obj, GenerationRecord) or record_id:
+        resolved_owner_type = GenerationOwnerType.GENERATION_RECORD.value
+        resolved_owner_id = record_id or owner_id or getattr(obj, "id", None)
+        resolved_task_id = None
+        resolved_record_id = resolved_owner_id
+        resolved_mode = generation_mode or GenerationMode.GENERATION_RECORD.value
+    elif isinstance(obj, ChatGenerationTask) or task_id:
+        resolved_owner_type = GenerationOwnerType.CHAT_GENERATION_TASK.value
+        resolved_owner_id = task_id or owner_id or getattr(obj, "id", None)
+        resolved_task_id = resolved_owner_id
+        resolved_record_id = None
+        resolved_mode = generation_mode or getattr(obj, "generation_mode", GenerationMode.CHATAPI_ASYNC.value)
+    else:
+        resolved_owner_type = owner_type or GenerationOwnerType.CHAT_GENERATION_TASK.value
+        resolved_owner_id = owner_id
+        if resolved_owner_type == GenerationOwnerType.GENERATION_RECORD.value:
+            resolved_task_id = None
+            resolved_record_id = resolved_owner_id
+            resolved_mode = generation_mode or GenerationMode.GENERATION_RECORD.value
+        else:
+            resolved_task_id = resolved_owner_id
+            resolved_record_id = None
+            resolved_mode = generation_mode or GenerationMode.CHATAPI_ASYNC.value
+    if not resolved_owner_id:
+        return None
+    return {
+        "owner_type": resolved_owner_type,
+        "owner_id": str(resolved_owner_id),
+        "task_id": str(resolved_task_id) if resolved_task_id else None,
+        "generation_record_id": str(resolved_record_id) if resolved_record_id else None,
+        "generation_attempt_no": int(generation_attempt_no or getattr(obj, "generation_attempt_no", 1) or 1),
+        "generation_mode": str(resolved_mode or ""),
+    }
+
+
+def _fallback_log(event_type: str, fields: dict[str, Any] | None, exc: Exception) -> None:
+    fields = fields or {}
+    log_operation_event(
+        domain="generation_pipeline",
+        event_type="PIPELINE_DB_LOG_FAILED",
+        event_status="failed",
+        task_id=fields.get("owner_id"),
+        message=f"数据库生成日志写入失败: {event_type}",
+        detail=build_exception_detail(exc, fields),
+        error=str(exc),
+    )
+
+
 async def log_task_event(
     task: Any | None = None,
     *,
     record: Any | None = None,
+    owner_type: str | None = None,
+    owner_id: str | None = None,
     task_id: str | None = None,
     record_id: str | None = None,
+    generation_attempt_no: int | None = None,
+    generation_mode: str | None = None,
     event_type: str,
     from_status: str | None = None,
     to_status: str | None = None,
@@ -54,17 +117,52 @@ async def log_task_event(
     message: str | None = None,
     detail: Any = None,
 ) -> None:
-    """Write task event in a separate transaction; failure must not affect main flow."""
+    """Write an owner-scoped event in a separate transaction."""
+    obj = task or record
+    fields = _owner_fields(
+        obj,
+        owner_type=owner_type,
+        owner_id=owner_id,
+        task_id=task_id,
+        record_id=record_id,
+        generation_attempt_no=generation_attempt_no,
+        generation_mode=generation_mode,
+    )
+    if not fields:
+        return
+    upper_event = str(event_type or "").upper()
+    failed_event = any(marker in upper_event for marker in ("FAILED", "TIMEOUT", "ERROR"))
+    log_operation_event(
+        domain="generation_pipeline",
+        event_type=event_type,
+        event_status="failed" if failed_event else "success",
+        source="pipeline",
+        user_id=str(getattr(obj, "user_id", "") or "") or None,
+        project_id=str(getattr(obj, "project_id", "") or "") or None,
+        task_id=fields["owner_id"],
+        message=message,
+        detail={
+            "owner_type": fields["owner_type"],
+            "owner_id": fields["owner_id"],
+            "generation_attempt_no": fields["generation_attempt_no"],
+            "generation_mode": fields["generation_mode"],
+            "from_status": from_status,
+            "to_status": to_status,
+            "from_stage": from_stage,
+            "to_stage": to_stage,
+            "detail_excerpt": _excerpt(detail),
+        },
+        error=message if failed_event else None,
+    )
     try:
-        obj = task or record
-        tid = task_id or record_id or (obj.id if obj else None)
-        if not tid:
-            return
         async with async_session() as db:
             db.add(ChatGenerationTaskEvent(
                 id=generate_id(),
-                task_id=tid,
-                generation_mode=getattr(obj, "generation_mode", "chatapi_async"),
+                owner_type=fields["owner_type"],
+                task_id=fields["task_id"],
+                generation_record_id=fields["generation_record_id"],
+                generation_attempt_no=fields["generation_attempt_no"],
+                generation_mode=fields["generation_mode"],
                 event_type=event_type,
                 from_status=from_status,
                 to_status=to_status,
@@ -74,16 +172,20 @@ async def log_task_event(
                 detail_json=_excerpt(detail),
             ))
             await db.commit()
-    except Exception:
-        return
+    except Exception as exc:
+        _fallback_log(event_type, fields, exc)
 
 
 async def log_provider_call(
     task: Any | None = None,
     *,
     record: Any | None = None,
+    owner_type: str | None = None,
+    owner_id: str | None = None,
     task_id: str | None = None,
     record_id: str | None = None,
+    generation_attempt_no: int | None = None,
+    generation_mode: str | None = None,
     provider: str | None,
     api_type: str,
     model: str | None = None,
@@ -100,17 +202,28 @@ async def log_provider_call(
     error_code: str | None = None,
     error_message: str | None = None,
 ) -> None:
-    """Write provider call log in a separate transaction; failure must not affect main flow."""
+    """Write an owner-scoped provider call log in a separate transaction."""
+    obj = task or record
+    fields = _owner_fields(
+        obj,
+        owner_type=owner_type,
+        owner_id=owner_id,
+        task_id=task_id,
+        record_id=record_id,
+        generation_attempt_no=generation_attempt_no,
+        generation_mode=generation_mode,
+    )
+    if not fields:
+        return
     try:
-        obj = task or record
-        tid = task_id or record_id or (obj.id if obj else None)
-        if not tid:
-            return
         async with async_session() as db:
             db.add(ChatProviderCallLog(
                 id=generate_id(),
-                task_id=tid,
-                generation_mode=getattr(obj, "generation_mode", "chatapi_async"),
+                owner_type=fields["owner_type"],
+                task_id=fields["task_id"],
+                generation_record_id=fields["generation_record_id"],
+                generation_attempt_no=fields["generation_attempt_no"],
+                generation_mode=fields["generation_mode"],
                 provider=provider,
                 api_type=api_type,
                 model=model,
@@ -130,5 +243,5 @@ async def log_provider_call(
                 error_message=error_message,
             ))
             await db.commit()
-    except Exception:
-        return
+    except Exception as exc:
+        _fallback_log(api_type, fields, exc)

video-gen-api/app/services/generation/pipeline/__init__.py

@@ -0,0 +1 @@
+"""Shared Celery generation pipeline for ChatGenerationTask and GenerationRecord."""

video-gen-api/app/services/generation/pipeline/db_lock_service.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import HTTPException
+from sqlalchemy import text
+from sqlalchemy.exc import DBAPIError, OperationalError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import settings
+
+
+class DatabaseRowLockBusy(HTTPException, RuntimeError):
+    """A short-lived PostgreSQL row/table lock could not be acquired in time."""
+
+    def __init__(self, message: str = "当前任务正在被其他流程处理，请稍后重试") -> None:
+        super().__init__(status_code=409, detail=message)
+
+
+_LOCK_NOT_AVAILABLE_SQLSTATE = "55P03"
+
+
+def _sqlstate_from_exception(exc: BaseException | None) -> str | None:
+    current: BaseException | None = exc
+    seen: set[int] = set()
+    while current is not None and id(current) not in seen:
+        seen.add(id(current))
+        for attr in ("sqlstate", "pgcode"):
+            value = getattr(current, attr, None)
+            if value:
+                return str(value)
+        current = getattr(current, "orig", None) or getattr(current, "__cause__", None)
+    return None
+
+
+def is_postgres_lock_timeout(exc: BaseException) -> bool:
+    if _sqlstate_from_exception(exc) == _LOCK_NOT_AVAILABLE_SQLSTATE:
+        return True
+    message = str(exc).lower()
+    return "lock timeout" in message or "could not obtain lock" in message
+
+
+def raise_if_database_lock_busy(exc: BaseException) -> None:
+    if is_postgres_lock_timeout(exc):
+        raise DatabaseRowLockBusy("数据库任务行正在被其他事务处理，请稍后重试") from exc
+
+
+async def apply_short_lock_timeout(
+    db: AsyncSession,
+    *,
+    seconds: int | None = None,
+) -> None:
+    """Apply a transaction-local PostgreSQL lock wait limit.
+
+    It deliberately does not change the global database configuration and is a
+    no-op on non-PostgreSQL test/development databases.
+    """
+    bind = db.get_bind()
+    if bind is None or bind.dialect.name != "postgresql":
+        return
+    timeout_seconds = max(
+        1,
+        int(
+            seconds
+            if seconds is not None
+            else getattr(settings, "GENERATION_DB_LOCK_TIMEOUT_SECONDS", 5)
+            or 5
+        ),
+    )
+    await db.execute(text(f"SET LOCAL lock_timeout = '{timeout_seconds}s'"))
+
+
+async def execute_with_lock_timeout(
+    db: AsyncSession,
+    statement: Any,
+    *,
+    seconds: int | None = None,
+):
+    await apply_short_lock_timeout(db, seconds=seconds)
+    try:
+        return await db.execute(statement)
+    except (OperationalError, DBAPIError) as exc:
+        raise_if_database_lock_busy(exc)
+        raise

video-gen-api/app/services/generation/pipeline/enqueue_service.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from app.enums.celery_queue import CeleryQueue
+from app.enums.generation_task import ChatGenerationTaskEventType
+from app.services.generation.log_service import log_task_event
+from app.services.generation.pipeline.owner_service import GenerationOwner, owner_type_of
+
+
+async def enqueue_generation_create(
+    owner: GenerationOwner,
+    *,
+    reason: str,
+) -> None:
+    """Commit caller-owned state before invoking this function."""
+    from app.tasks.generation_create_tasks import chatapi_create_generation_task
+
+    owner_type = owner_type_of(owner)
+    attempt_no = int(getattr(owner, "generation_attempt_no", 1) or 1)
+    try:
+        chatapi_create_generation_task.apply_async(
+            args=[str(owner.id)],
+            kwargs={"owner_type": owner_type, "generation_attempt_no": attempt_no},
+            queue=CeleryQueue.GEN_CHATAPI_CREATE.value,
+            task_id=f"generation-create:{owner_type}:{owner.id}:attempt:{attempt_no}",
+        )
+        await log_task_event(
+            owner,
+            event_type=ChatGenerationTaskEventType.GENERATION_RECORD_ENQUEUE_SUCCESS.value,
+            message="资源生成创建任务已投递",
+            detail={"reason": reason, "queue": CeleryQueue.GEN_CHATAPI_CREATE.value},
+        )
+    except Exception as exc:
+        await log_task_event(
+            owner,
+            event_type=ChatGenerationTaskEventType.GENERATION_RECORD_ENQUEUE_FAILED.value,
+            message=str(exc),
+            detail={"reason": reason, "queue": CeleryQueue.GEN_CHATAPI_CREATE.value},
+        )
+        raise

video-gen-api/app/services/generation/pipeline/generation_record_service.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.enums.generation_status import GenerationRecordPipelineStage, GenerationStatus
+from app.models.generation_record import GenerationRecord
+from app.models.image_engine import ImageEngine
+from app.models.video_engine import VideoEngine
+from app.services.generation.ai.engine_service import build_image_snapshot, build_video_snapshot
+from app.services.generation.pipeline.enqueue_service import enqueue_generation_create
+from app.services.generation.pipeline.lifecycle_service import reset_execution_fields
+
+
+def _json(data: dict) -> str:
+    return json.dumps(data, ensure_ascii=False, default=str)
+
+
+def prepare_generation_record_execution(
+    record: GenerationRecord,
+    *,
+    engine: ImageEngine | VideoEngine,
+    attempt_no: int,
+) -> None:
+    now = datetime.now(timezone.utc)
+    reset_execution_fields(record, started_at=now, attempt_no=attempt_no)
+    record.engine_id = engine.id
+    if record.gen_type == "image":
+        record.engine_snapshot_json = _json(build_image_snapshot(
+            engine,
+            record.image_size or getattr(engine, "default_size", "2K") or "2K",
+            record.image_proportion or "1:1",
+            record.image_px or "2048x2048",
+        ))
+    else:
+        record.engine_snapshot_json = _json(build_video_snapshot(
+            engine,
+            record.aspect_ratio or "16:9",
+            record.resolution or "480p",
+            int(record.duration or 4),
+        ))
+    record.status = GenerationStatus.generating.value
+    record.pipeline_stage = GenerationRecordPipelineStage.QUEUED.value
+
+
+async def commit_and_enqueue_generation_record(
+    db: AsyncSession,
+    record: GenerationRecord,
+    *,
+    reason: str,
+) -> None:
+    await db.commit()
+    try:
+        await enqueue_generation_create(record, reason=reason)
+    except Exception:
+        # queued stage and all execution metadata are already committed; recovery will retry.
+        return

video-gen-api/app/services/generation/pipeline/lifecycle_service.py

@@ -0,0 +1,149 @@
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import settings
+from app.enums.generation_status import GenerationRecordPipelineStage
+from app.enums.generation_task import ChatGenerationPipelineStage, GenerationType
+from app.models.chat_generation_task import ChatGenerationTask
+from app.models.generation_record import GenerationRecord
+from app.models.base import async_session
+from app.services.generation.pipeline.db_lock_service import DatabaseRowLockBusy
+from app.services.generation.pipeline.owner_service import GenerationOwner
+from app.services.generation.refund_service import (
+    mark_chat_generation_task_failed_and_refund_once,
+    mark_generation_record_failed_and_refund_once,
+)
+
+
+def utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def generation_deadline(*, gen_type: str, started_at: datetime) -> datetime:
+    if str(gen_type or "").lower() == GenerationType.IMAGE.value:
+        return started_at + timedelta(minutes=max(1, int(settings.CHATAPI_ASYNC_IMAGE_DEADLINE_MINUTES or 30)))
+    return started_at + timedelta(hours=max(1, int(settings.CHATAPI_ASYNC_VIDEO_FINAL_DEADLINE_HOURS or 24)))
+
+
+def reset_execution_fields(owner: GenerationOwner, *, started_at: datetime, attempt_no: int) -> None:
+    owner.generation_attempt_no = max(1, int(attempt_no or 1))
+    owner.resource_generation_started_at = started_at
+    owner.deadline_at = generation_deadline(gen_type=owner.gen_type, started_at=started_at)
+    owner.error_message = None
+    owner.seedance_task_id = None
+    if hasattr(owner, "provider_task_id"):
+        owner.provider_task_id = None
+    owner.remote_result_url = None
+    owner.provider_response_json = None
+    owner.provider_create_claim_token = None
+    owner.provider_create_lease_until = None
+    owner.provider_create_started_at = None
+    owner.retry_count = int(getattr(owner, "manual_retry_count", 0) or 0)
+    owner.poll_error_count = 0
+    owner.poll_count = 0
+    owner.last_poll_at = None
+    owner.poll_started_at = None
+    owner.next_poll_at = None
+    owner.poll_interval_seconds = 0
+    owner.poll_claim_token = None
+    owner.poll_lease_until = None
+    owner.download_celery_task_id = None
+    owner.download_enqueued_at = None
+    owner.download_started_at = None
+    owner.download_claim_token = None
+    owner.download_lease_until = None
+    owner.download_next_retry_at = None
+    owner.download_attempt_count = 0
+    owner.download_last_error = None
+    owner.download_storage_date_dir = None
+    owner.generated_at = None
+    owner.image_url = None
+    owner.video_url = None
+    owner.video_cover_url = None
+
+
+async def notify_owner_finished(db: AsyncSession, owner: GenerationOwner) -> None:
+    """Run module hooks after the generation owner transaction has committed.
+
+    Module project/step rows have their own short locks. A transient lock conflict
+    must not turn an already completed generation into a download/provider retry,
+    so hooks use a fresh short transaction with a small local retry window.
+    """
+    _ = db
+    if not isinstance(owner, ChatGenerationTask):
+        return
+    from app.services.generation.module_hook_service import notify_chat_generation_task_finished
+    from app.services.generation.ai.task_group_service import aggregate_parent_for_child
+    from app.services.operation_log_service import log_operation_event
+
+    task_id = str(owner.id)
+    attempt_no = int(owner.generation_attempt_no or 1)
+    last_error: Exception | None = None
+    for retry_index in range(3):
+        try:
+            async with async_session() as hook_db:
+                result = await hook_db.execute(
+                    select(ChatGenerationTask).where(
+                        ChatGenerationTask.id == task_id,
+                        ChatGenerationTask.deleted_at.is_(None),
+                    ).limit(1)
+                )
+                fresh_task = result.scalar_one_or_none()
+                if fresh_task is None:
+                    return
+                await notify_chat_generation_task_finished(hook_db, fresh_task)
+                await aggregate_parent_for_child(hook_db, fresh_task)
+                await hook_db.commit()
+                return
+        except DatabaseRowLockBusy as exc:
+            last_error = exc
+            await asyncio.sleep(1 + retry_index)
+        except Exception as exc:
+            last_error = exc
+            break
+
+    log_operation_event(
+        domain="generation_pipeline",
+        event_type="MODULE_HOOK_DEFERRED_MANUAL_CHECK",
+        event_status="failed",
+        source="pipeline",
+        task_id=task_id,
+        message="生成任务已进入终态，但模块状态回填失败，需要人工排查",
+        detail={
+            "generation_attempt_no": attempt_no,
+            "retry_count": 3,
+        },
+        error=str(last_error or "unknown module hook error"),
+    )
+
+
+async def mark_owner_failed_and_refund_once(
+    db: AsyncSession,
+    owner: GenerationOwner,
+    *,
+    error_message: str,
+    pipeline_stage: str,
+) -> GenerationOwner:
+    if isinstance(owner, ChatGenerationTask):
+        await mark_chat_generation_task_failed_and_refund_once(
+            db,
+            task=owner,
+            error_message=error_message,
+            pipeline_stage=pipeline_stage,
+            generation_attempt_no=int(owner.generation_attempt_no or 1),
+        )
+        return owner
+
+    owner.pipeline_stage = pipeline_stage if pipeline_stage in {item.value for item in GenerationRecordPipelineStage} else GenerationRecordPipelineStage.FAILED.value
+    await mark_generation_record_failed_and_refund_once(
+        db,
+        record=owner,
+        error_message=error_message,
+        generation_attempt_no=int(owner.generation_attempt_no or 1),
+    )
+    return owner

video-gen-api/app/services/generation/pipeline/owner_service.py

@@ -0,0 +1,192 @@
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass
+from typing import TypeAlias
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.enums.generation_status import GenerationStatus
+from app.enums.generation_task import (
+    ChatGenerationTaskStatus,
+    GenerationMode,
+    GenerationOwnerType,
+)
+from app.models.chat_generation_task import ChatGenerationTask
+from app.models.generation_record import GenerationRecord
+from app.services.generation.pipeline.db_lock_service import (
+    DatabaseRowLockBusy,
+    apply_short_lock_timeout,
+    raise_if_database_lock_busy,
+)
+
+GenerationOwner: TypeAlias = ChatGenerationTask | GenerationRecord
+
+
+@dataclass(frozen=True, slots=True)
+class GenerationOwnerRef:
+    owner_type: str
+    owner_id: str
+    generation_attempt_no: int | None = None
+
+    @classmethod
+    def from_owner(cls, owner: GenerationOwner) -> "GenerationOwnerRef":
+        return cls(
+            owner_type=owner_type_of(owner),
+            owner_id=str(owner.id),
+            generation_attempt_no=int(getattr(owner, "generation_attempt_no", 1) or 1),
+        )
+
+
+def normalize_owner_type(value: str | GenerationOwnerType | None) -> str:
+    if isinstance(value, GenerationOwnerType):
+        return value.value
+    text = str(value or "").strip().lower()
+    if not text:
+        return GenerationOwnerType.CHAT_GENERATION_TASK.value
+    if text not in {item.value for item in GenerationOwnerType}:
+        raise ValueError(f"不支持的生成任务所有者类型: {value}")
+    return text
+
+
+def owner_type_of(owner: GenerationOwner) -> str:
+    if isinstance(owner, ChatGenerationTask):
+        return GenerationOwnerType.CHAT_GENERATION_TASK.value
+    if isinstance(owner, GenerationRecord):
+        return GenerationOwnerType.GENERATION_RECORD.value
+    raise TypeError(f"不支持的生成任务对象: {type(owner)!r}")
+
+
+def owner_mode(owner: GenerationOwner) -> str:
+    if isinstance(owner, ChatGenerationTask):
+        return str(owner.generation_mode or GenerationMode.CHATAPI_ASYNC.value)
+    return GenerationMode.GENERATION_RECORD.value
+
+
+def owner_provider_task_id(owner: GenerationOwner) -> str | None:
+    return str(getattr(owner, "provider_task_id", None) or getattr(owner, "seedance_task_id", None) or "") or None
+
+
+def set_owner_provider_task_id(owner: GenerationOwner, value: str | None) -> None:
+    if hasattr(owner, "provider_task_id"):
+        owner.provider_task_id = value
+    owner.seedance_task_id = value
+
+
+def owner_is_generating(owner: GenerationOwner) -> bool:
+    if isinstance(owner, ChatGenerationTask):
+        return owner.status == ChatGenerationTaskStatus.GENERATING.value
+    return owner.status == GenerationStatus.generating.value
+
+
+def owner_is_completed(owner: GenerationOwner) -> bool:
+    if isinstance(owner, ChatGenerationTask):
+        return owner.status == ChatGenerationTaskStatus.COMPLETED.value
+    return owner.status == GenerationStatus.completed.value
+
+
+def owner_is_failed(owner: GenerationOwner) -> bool:
+    if isinstance(owner, ChatGenerationTask):
+        return owner.status == ChatGenerationTaskStatus.FAILED.value
+    return owner.status == GenerationStatus.failed.value
+
+
+def set_owner_generating(owner: GenerationOwner) -> None:
+    owner.status = ChatGenerationTaskStatus.GENERATING.value if isinstance(owner, ChatGenerationTask) else GenerationStatus.generating.value
+
+
+def set_owner_completed(owner: GenerationOwner) -> None:
+    owner.status = ChatGenerationTaskStatus.COMPLETED.value if isinstance(owner, ChatGenerationTask) else GenerationStatus.completed.value
+
+
+def set_owner_failed(owner: GenerationOwner) -> None:
+    owner.status = ChatGenerationTaskStatus.FAILED.value if isinstance(owner, ChatGenerationTask) else GenerationStatus.failed.value
+
+
+def is_attempt_current(owner: GenerationOwner, attempt_no: int | None) -> bool:
+    if attempt_no is None:
+        return True
+    return int(getattr(owner, "generation_attempt_no", 1) or 1) == int(attempt_no)
+
+
+async def load_generation_owner(
+    db: AsyncSession,
+    *,
+    owner_type: str | GenerationOwnerType | None,
+    owner_id: str,
+    for_update: bool = False,
+    include_deleted: bool = False,
+) -> GenerationOwner | None:
+    normalized = normalize_owner_type(owner_type)
+    if normalized == GenerationOwnerType.CHAT_GENERATION_TASK.value:
+        query = select(ChatGenerationTask).where(ChatGenerationTask.id == owner_id)
+        if not include_deleted:
+            query = query.where(ChatGenerationTask.deleted_at.is_(None))
+    else:
+        query = select(GenerationRecord).where(GenerationRecord.id == owner_id)
+        if not include_deleted:
+            query = query.where(GenerationRecord.deleted_at.is_(None))
+    if for_update:
+        await apply_short_lock_timeout(db)
+        query = query.with_for_update().execution_options(populate_existing=True)
+    try:
+        result = await db.execute(query.limit(1))
+    except Exception as exc:
+        raise_if_database_lock_busy(exc)
+        raise
+    return result.scalar_one_or_none()
+
+
+
+async def load_generation_owner_for_update_retry(
+    db: AsyncSession,
+    *,
+    owner_type: str | GenerationOwnerType | None,
+    owner_id: str,
+    attempts: int = 3,
+) -> GenerationOwner | None:
+    """Retry a short owner row lock inside the same Worker.
+
+    This is intended after an external call/download has already completed so a
+    transient row lock does not force the Worker to repeat that external side effect.
+    """
+    last_error: DatabaseRowLockBusy | None = None
+    for retry_index in range(max(1, int(attempts or 1))):
+        try:
+            return await load_generation_owner(
+                db,
+                owner_type=owner_type,
+                owner_id=owner_id,
+                for_update=True,
+            )
+        except DatabaseRowLockBusy as exc:
+            last_error = exc
+            await db.rollback()
+            if retry_index + 1 < max(1, int(attempts or 1)):
+                await asyncio.sleep(1 + retry_index)
+    raise last_error or DatabaseRowLockBusy()
+
+def redis_owner_item_id(owner_type: str | GenerationOwnerType | None, owner_id: str, attempt_no: int | None = None) -> str:
+    normalized = normalize_owner_type(owner_type)
+    if attempt_no is None:
+        return f"{normalized}:{owner_id}"
+    return f"{normalized}:{owner_id}:attempt:{int(attempt_no)}"
+
+
+def parse_redis_owner_item_id(value: str) -> GenerationOwnerRef:
+    text = str(value or "").strip()
+    for owner_type in (GenerationOwnerType.CHAT_GENERATION_TASK.value, GenerationOwnerType.GENERATION_RECORD.value):
+        prefix = f"{owner_type}:"
+        if text.startswith(prefix):
+            rest = text[len(prefix):]
+            marker = ":attempt:"
+            if marker in rest:
+                owner_id, attempt = rest.rsplit(marker, 1)
+                try:
+                    return GenerationOwnerRef(owner_type, owner_id, int(attempt))
+                except ValueError:
+                    return GenerationOwnerRef(owner_type, owner_id, None)
+            return GenerationOwnerRef(owner_type, rest, None)
+    # Historical Redis/Celery identifiers always belonged to ChatGenerationTask.
+    return GenerationOwnerRef(GenerationOwnerType.CHAT_GENERATION_TASK.value, text, None)

video-gen-api/app/services/generation/pipeline/recovery_repository.py

@@ -0,0 +1,148 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import settings
+from app.enums.generation_status import GenerationRecordPipelineStage, GenerationStatus
+from app.enums.generation_task import GenerationOwnerType
+from app.models.generation_record import GenerationRecord
+from app.services.generation.pipeline.owner_service import GenerationOwnerRef
+from app.services.redis_registry_service import ensure_aware_utc
+
+
+@dataclass(frozen=True, slots=True)
+class GenerationRecordRecoveryCursor:
+    owner_id: str
+
+
+@dataclass(slots=True)
+class GenerationRecordRecoveryBatch:
+    create: list[GenerationOwnerRef]
+    poll: list[GenerationOwnerRef]
+    download: list[GenerationOwnerRef]
+    next_cursor: GenerationRecordRecoveryCursor | None
+
+
+async def find_generation_record_recovery_batch(
+    db: AsyncSession,
+    *,
+    limit: int,
+    cursor: GenerationRecordRecoveryCursor | None = None,
+) -> GenerationRecordRecoveryBatch:
+    """按稳定游标读取恢复分流所需列，避免大字段加载和 offset 扫描。"""
+    stages = {
+        GenerationRecordPipelineStage.QUEUED.value,
+        GenerationRecordPipelineStage.PREPARING.value,
+        GenerationRecordPipelineStage.CREATING_PROVIDER_TASK.value,
+        GenerationRecordPipelineStage.WAITING_REMOTE.value,
+        GenerationRecordPipelineStage.POLLING.value,
+        GenerationRecordPipelineStage.RESULT_READY.value,
+        GenerationRecordPipelineStage.DOWNLOAD_QUEUED.value,
+        GenerationRecordPipelineStage.DOWNLOADING.value,
+        GenerationRecordPipelineStage.RETRY_WAITING.value,
+    }
+    page_size = max(1, int(limit))
+    now = datetime.now(timezone.utc)
+    queue_timeout = timedelta(
+        seconds=max(1, int(settings.DOWNLOAD_TASK_QUEUE_TIMEOUT_SECONDS or 300))
+    )
+    query = select(
+        GenerationRecord.id,
+        GenerationRecord.generation_attempt_no,
+        GenerationRecord.seedance_task_id,
+        GenerationRecord.remote_result_url,
+        GenerationRecord.pipeline_stage,
+        GenerationRecord.provider_create_lease_until,
+        GenerationRecord.next_poll_at,
+        GenerationRecord.poll_lease_until,
+        GenerationRecord.download_enqueued_at,
+        GenerationRecord.download_lease_until,
+        GenerationRecord.download_next_retry_at,
+    ).where(
+        GenerationRecord.deleted_at.is_(None),
+        GenerationRecord.status == GenerationStatus.generating.value,
+        GenerationRecord.pipeline_stage.in_(stages),
+    )
+    if cursor is not None:
+        query = query.where(GenerationRecord.id > cursor.owner_id)
+    result = await db.execute(
+        query.order_by(GenerationRecord.id.asc()).limit(page_size)
+    )
+    rows = list(result.all())
+
+    create: list[GenerationOwnerRef] = []
+    poll: list[GenerationOwnerRef] = []
+    download: list[GenerationOwnerRef] = []
+    for (
+        owner_id,
+        attempt_no,
+        provider_task_id,
+        remote_result_url,
+        pipeline_stage,
+        provider_create_lease_until,
+        next_poll_at,
+        poll_lease_until,
+        download_enqueued_at,
+        download_lease_until,
+        download_next_retry_at,
+    ) in rows:
+        ref = GenerationOwnerRef(
+            GenerationOwnerType.GENERATION_RECORD.value,
+            str(owner_id),
+            int(attempt_no or 1),
+        )
+        stage = str(pipeline_stage or "")
+        if str(remote_result_url or "").strip():
+            if stage == GenerationRecordPipelineStage.RESULT_READY.value:
+                download.append(ref)
+            elif stage == GenerationRecordPipelineStage.DOWNLOAD_QUEUED.value:
+                checked_enqueued_at = ensure_aware_utc(download_enqueued_at)
+                if (
+                    checked_enqueued_at is None
+                    or checked_enqueued_at + queue_timeout <= now
+                ):
+                    download.append(ref)
+            elif stage == GenerationRecordPipelineStage.DOWNLOADING.value:
+                if (
+                    ensure_aware_utc(download_lease_until) is None
+                    or ensure_aware_utc(download_lease_until) <= now
+                ):
+                    download.append(ref)
+            elif stage == GenerationRecordPipelineStage.RETRY_WAITING.value:
+                if (
+                    ensure_aware_utc(download_next_retry_at) is None
+                    or ensure_aware_utc(download_next_retry_at) <= now
+                ):
+                    download.append(ref)
+        elif str(provider_task_id or "").strip():
+            checked_next_poll_at = ensure_aware_utc(next_poll_at)
+            checked_poll_lease_until = ensure_aware_utc(poll_lease_until)
+            if (
+                (checked_next_poll_at is None or checked_next_poll_at <= now)
+                and (
+                    checked_poll_lease_until is None
+                    or checked_poll_lease_until <= now
+                )
+            ):
+                poll.append(ref)
+        else:
+            if (
+                ensure_aware_utc(provider_create_lease_until) is None
+                or ensure_aware_utc(provider_create_lease_until) <= now
+            ):
+                create.append(ref)
+
+    next_cursor = None
+    if len(rows) == page_size:
+        last = rows[-1]
+        next_cursor = GenerationRecordRecoveryCursor(owner_id=str(last.id))
+    return GenerationRecordRecoveryBatch(
+        create=create,
+        poll=poll,
+        download=download,
+        next_cursor=next_cursor,
+    )

video-gen-api/app/services/generation/poll_schedule_service.py

@@ -5,7 +5,7 @@ from datetime import datetime, timedelta, timezone
 
 from app.config import settings
 from app.enums.generation_task import GenerationType
-from app.models.chat_generation_task import ChatGenerationTask
+from app.services.generation.pipeline.owner_service import GenerationOwner
 from app.services.redis_registry_service import ensure_aware_utc
 
 
@@ -23,7 +23,7 @@ def utc_now() -> datetime:
     return datetime.now(timezone.utc)
 
 
-def is_video_generation_task(task: ChatGenerationTask) -> bool:
+def is_video_generation_task(task: GenerationOwner) -> bool:
     return str(getattr(task, "gen_type", "") or "").lower() == GenerationType.VIDEO.value
 
 
@@ -33,7 +33,7 @@ def video_final_deadline_from(now: datetime | None = None) -> datetime:
     return current_time + timedelta(hours=hours)
 
 
-def ensure_video_poll_fields(task: ChatGenerationTask, *, now: datetime | None = None) -> None:
+def ensure_video_poll_fields(task: GenerationOwner, *, now: datetime | None = None) -> None:
     """补齐视频轮询调度字段，兼容历史任务。"""
     if not is_video_generation_task(task):
         return
@@ -42,17 +42,22 @@ def ensure_video_poll_fields(task: ChatGenerationTask, *, now: datetime | None =
     if ensure_aware_utc(getattr(task, "poll_started_at", None)) is None:
         task.poll_started_at = current_time
     if ensure_aware_utc(getattr(task, "deadline_at", None)) is None:
-        task.deadline_at = video_final_deadline_from(current_time)
+        resource_started_at = (
+            ensure_aware_utc(getattr(task, "resource_generation_started_at", None))
+            or ensure_aware_utc(getattr(task, "created_at", None))
+            or current_time
+        )
+        task.deadline_at = video_final_deadline_from(resource_started_at)
     if getattr(task, "poll_interval_seconds", None) is None:
         task.poll_interval_seconds = 0
 
 
-def is_final_poll_due(task: ChatGenerationTask, *, now: datetime | None = None) -> bool:
+def is_final_poll_due(task: GenerationOwner, *, now: datetime | None = None) -> bool:
     deadline_at = ensure_aware_utc(getattr(task, "deadline_at", None))
     return bool(deadline_at and deadline_at <= (now or utc_now()))
 
 
-def is_poll_not_due(task: ChatGenerationTask, *, now: datetime | None = None, tolerance_seconds: int = 1) -> bool:
+def is_poll_not_due(task: GenerationOwner, *, now: datetime | None = None, tolerance_seconds: int = 1) -> bool:
     """判断当前 poll 任务是否早于 next_poll_at。只对视频降频轮询生效。"""
     if not is_video_generation_task(task):
         return False
@@ -73,7 +78,7 @@ def _clamp_positive_seconds(value: int | float | None, default: int) -> int:
 
 
 def build_video_pending_poll_schedule(
-    task: ChatGenerationTask,
+    task: GenerationOwner,
     *,
     now: datetime | None = None,
 ) -> PollScheduleDecision:
@@ -133,7 +138,7 @@ def build_video_pending_poll_schedule(
 
 
 def build_default_poll_schedule(
-    task: ChatGenerationTask,
+    task: GenerationOwner,
     *,
     now: datetime | None = None,
     delay_seconds: int | None = None,

video-gen-api/app/services/generation/prompt_service.py

@@ -75,7 +75,7 @@ async def _build_user_content(record: ChatGenerationTask, db: AsyncSession | Non
 async def _get_model_config(db: AsyncSession) -> ModelConfig:
     result = await db.execute(
         select(ModelConfig)
-        .where(ModelConfig.is_active == True)
+        .where(ModelConfig.is_active == True, ModelConfig.deleted_at.is_(None))
         .order_by(ModelConfig.priority.desc())
         .limit(1)
     )

video-gen-api/app/services/generation/provider_service.py

@@ -11,6 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.config import settings
 from app.models.chat_generation_task import ChatGenerationTask
+from app.services.generation.pipeline.owner_service import GenerationOwner, owner_provider_task_id
 from app.models.image_engine import ImageEngine
 from app.models.video_engine import VideoEngine
 from app.services.generation.log_service import log_provider_call
@@ -39,7 +40,7 @@ def _try_json(value: Any) -> Any:
         return None
 
 
-async def get_runtime_engine(db: AsyncSession, task: ChatGenerationTask) -> Any:
+async def get_runtime_engine(db: AsyncSession, task: GenerationOwner) -> Any:
     """使用任务快照冻结历史参数，只从当前引擎记录读取密钥。"""
     snapshot = _loads(task.engine_snapshot_json)
     if not task.engine_id:
@@ -50,7 +51,7 @@ async def get_runtime_engine(db: AsyncSession, task: ChatGenerationTask) -> Any:
         result = await db.execute(select(VideoEngine).where(VideoEngine.id == task.engine_id).limit(1))
     engine = result.scalar_one_or_none()
     if not engine:
-        raise ValueError("引擎不存在或已删除")
+        raise ValueError("任务绑定的引擎不存在")
     return SimpleNamespace(
         id=task.engine_id,
         name=snapshot.get("name") or engine.name,
@@ -89,7 +90,7 @@ async def get_runtime_engine(db: AsyncSession, task: ChatGenerationTask) -> Any:
     )
 
 
-async def create_provider_task(db: AsyncSession, task: ChatGenerationTask) -> dict:
+async def create_provider_task(db: AsyncSession, task: GenerationOwner) -> dict:
     if task.gen_type == "video":
         return await _create_video_task(db, task)
     if task.gen_type == "image":
@@ -97,12 +98,14 @@ async def create_provider_task(db: AsyncSession, task: ChatGenerationTask) -> di
     raise ValueError(f"不支持的生成类型: {task.gen_type}")
 
 
-async def _create_video_task(db: AsyncSession, task: ChatGenerationTask) -> dict:
+async def _create_video_task(db: AsyncSession, task: GenerationOwner) -> dict:
     engine = await get_runtime_engine(db, task)
+    # Close the engine lookup transaction before the long provider HTTP call.
+    await db.commit()
     started = time.perf_counter()
     async with provider_limit("ark_video_create", settings.ARK_VIDEO_CREATE_MAX_CONCURRENCY):
         try:
-            provider_task_id = await submit_video_task(None, engine, task, include_media_references=True)
+            provider_task_id = await submit_video_task(None, engine, task, include_media_references=isinstance(task, ChatGenerationTask))
             response = {"task_id": provider_task_id}
             await log_provider_call(
                 task,
@@ -132,11 +135,13 @@ async def _create_video_task(db: AsyncSession, task: ChatGenerationTask) -> dict
 
 async def create_image_sync_batch_result(
     db: AsyncSession,
-    task: ChatGenerationTask,
+    task: GenerationOwner,
     *,
     generation_count: int,
 ) -> ImageProviderBatchResult:
     engine = await get_runtime_engine(db, task)
+    # Do not keep a database transaction open while the synchronous provider call runs.
+    await db.commit()
     return await create_image_sync_batch_result_with_engine(
         task,
         engine,
@@ -145,7 +150,7 @@ async def create_image_sync_batch_result(
 
 
 async def create_image_sync_batch_result_with_engine(
-    task: ChatGenerationTask,
+    task: GenerationOwner,
     engine: Any,
     *,
     generation_count: int,
@@ -164,7 +169,7 @@ async def create_image_sync_batch_result_with_engine(
                 None,
                 engine,
                 task,
-                include_media_references=True,
+                include_media_references=isinstance(task, ChatGenerationTask),
                 generation_count=count,
             )
             response_data = result.get("response_data") or result
@@ -197,7 +202,7 @@ async def create_image_sync_batch_result_with_engine(
             raise
 
 
-async def create_image_sync_result(db: AsyncSession, task: ChatGenerationTask) -> dict:
+async def create_image_sync_result(db: AsyncSession, task: GenerationOwner) -> dict:
     result = await create_image_sync_batch_result(db, task, generation_count=1)
     items = result.get("items") or []
     if len(items) != 1:
@@ -216,9 +221,13 @@ async def create_image_sync_result(db: AsyncSession, task: ChatGenerationTask) -
     }
 
 
-async def poll_provider_task(db: AsyncSession, task: ChatGenerationTask) -> dict:
+async def poll_provider_task(db: AsyncSession, task: GenerationOwner) -> dict:
     engine = await get_runtime_engine(db, task)
-    task_id = task.seedance_task_id or task.provider_task_id
+    # Polling may block on the remote provider; release the lookup transaction first.
+    await db.commit()
+    task_id = owner_provider_task_id(task)
+    if not task_id:
+        raise ValueError("缺少供应商任务ID")
     if task.gen_type == "video":
         async with provider_limit("ark_video_poll", settings.ARK_VIDEO_POLL_MAX_CONCURRENCY):
             return await poll_task_status(engine, task_id)

video-gen-api/app/services/generation/recovery_service.py

@@ -16,6 +16,7 @@ from app.enums.generation_task import (
     ChatGenerationTaskEventType,
     ChatGenerationTaskStatus,
     GenerationMode,
+    GenerationOwnerType,
     GenerationType,
 )
 from app.models.chat_generation_task import ChatGenerationTask
@@ -27,9 +28,15 @@ from app.services.celery_download_recovery_service import (
     remove_download_active,
 )
 from app.services.generation.log_service import log_task_event
-from app.services.generation.module_hook_service import notify_chat_generation_task_finished
+from app.services.generation.pipeline.db_lock_service import apply_short_lock_timeout
+from app.services.generation.pipeline.lifecycle_service import notify_owner_finished
 from app.services.generation.poll_schedule_service import ensure_video_poll_fields, is_poll_not_due, is_video_generation_task
 from app.services.generation.refund_service import mark_chat_generation_task_failed_and_refund_once
+from app.services.generation.pipeline.owner_service import (
+    load_generation_owner,
+    parse_redis_owner_item_id,
+    redis_owner_item_id,
+)
 from app.services.redis_registry_service import (
     redis_get_due_registry_ids,
     redis_get_registry_payloads,
@@ -42,6 +49,26 @@ logger = logging.getLogger("video_gen")
 POLL_QUEUE = CeleryQueue.GEN_PROVIDER_POLL.value
 
 
+def _chat_registry_id(task: ChatGenerationTask) -> str:
+    return redis_owner_item_id(
+        GenerationOwnerType.CHAT_GENERATION_TASK.value,
+        str(task.id),
+        int(getattr(task, "generation_attempt_no", 1) or 1),
+    )
+
+
+async def _load_chat_task_for_update(
+    db: AsyncSession, task_id: str
+) -> ChatGenerationTask | None:
+    owner = await load_generation_owner(
+        db,
+        owner_type=GenerationOwnerType.CHAT_GENERATION_TASK.value,
+        owner_id=str(task_id),
+        for_update=True,
+    )
+    return owner if isinstance(owner, ChatGenerationTask) else None
+
+
 def _now() -> datetime:
     return datetime.now(timezone.utc)
 
@@ -140,26 +167,47 @@ async def recover_one_download_task(
     if not task:
         return "skip_missing_task"
     if task.generation_mode not in ALLOWED_GENERATION_MODES:
-        await remove_download_active(task.id)
+        await remove_download_active(_chat_registry_id(task))
         return "clean_invalid_mode"
     if _is_final_task_state(task):
-        await remove_download_active(task.id)
+        await remove_download_active(_chat_registry_id(task))
         return "clean_final_state"
     if task.status != ChatGenerationTaskStatus.GENERATING.value:
-        await remove_download_active(task.id)
+        task_id = str(task.id)
+        generation_attempt_no = int(task.generation_attempt_no or 1)
+        generation_mode = str(task.generation_mode or "")
+        status = task.status
+        stage = task.pipeline_stage
+        registry_id = _chat_registry_id(task)
+        await db.rollback()
+        await remove_download_active(registry_id)
         await log_task_event(
-            task,
+            owner_type=GenerationOwnerType.CHAT_GENERATION_TASK.value,
+            owner_id=task_id,
+            task_id=task_id,
+            generation_attempt_no=generation_attempt_no,
+            generation_mode=generation_mode,
             event_type=ChatGenerationTaskEventType.DOWNLOAD_SKIP_NOT_GENERATING.value,
             message=f"{source} 下载恢复跳过：任务不是 generating",
-            detail={"status": task.status, "stage": task.pipeline_stage},
+            detail={"status": status, "stage": stage},
         )
         return "clean_not_generating"
     if not task.remote_result_url:
+        task_id = str(task.id)
+        generation_attempt_no = int(task.generation_attempt_no or 1)
+        generation_mode = str(task.generation_mode or "")
+        status = task.status
+        stage = task.pipeline_stage
+        await db.rollback()
         await log_task_event(
-            task,
+            owner_type=GenerationOwnerType.CHAT_GENERATION_TASK.value,
+            owner_id=task_id,
+            task_id=task_id,
+            generation_attempt_no=generation_attempt_no,
+            generation_mode=generation_mode,
             event_type=ChatGenerationTaskEventType.DOWNLOAD_SKIP_NO_REMOTE_RESULT_URL.value,
             message=f"{source} 下载恢复跳过：缺少 remote_result_url",
-            detail={"status": task.status, "stage": task.pipeline_stage},
+            detail={"status": status, "stage": stage},
         )
         return "skip_no_remote_result_url"
 
@@ -167,38 +215,38 @@ async def recover_one_download_task(
     redis_payload = payload or {}
 
     if stage == ChatGenerationPipelineStage.RESULT_READY.value:
-        await log_task_event(
-            task,
-            event_type=ChatGenerationTaskEventType.DOWNLOAD_RECOVERY_ENQUEUE.value,
-            message=f"{source} 发现 result_ready 未完成下载，启动时恢复投递下载任务",
-            detail={"payload": redis_payload},
-        )
         await enqueue_download_task(
             db,
             task,
             recover=True,
             reason=f"{source}_result_ready",
         )
+        await log_task_event(
+            task,
+            event_type=ChatGenerationTaskEventType.DOWNLOAD_RECOVERY_ENQUEUE.value,
+            message=f"{source} 发现 result_ready 未完成下载，启动时恢复投递下载任务",
+            detail={"payload": redis_payload},
+        )
         return "recover_result_ready"
 
     if stage == DOWNLOAD_STAGE_QUEUED:
         if _is_queue_timeout(task, current_time):
-            await log_task_event(
-                task,
-                event_type=ChatGenerationTaskEventType.DOWNLOAD_RECOVERY_ENQUEUE.value,
-                message=f"{source} 发现 download_queued 长时间未消费，启动时恢复投递下载任务",
-                detail={"payload": redis_payload},
-            )
             await enqueue_download_task(
                 db,
                 task,
                 recover=True,
                 reason=f"{source}_download_queued_timeout",
             )
+            await log_task_event(
+                task,
+                event_type=ChatGenerationTaskEventType.DOWNLOAD_RECOVERY_ENQUEUE.value,
+                message=f"{source} 发现 download_queued 长时间未消费，启动时恢复投递下载任务",
+                detail={"payload": redis_payload},
+            )
             return "recover_queued_timeout"
 
         await postpone_download_active_check(
-            record_id=task.id,
+            record_id=_chat_registry_id(task),
             payload=payload,
             check_at=_queue_timeout_at(task, current_time),
         )
@@ -206,22 +254,22 @@ async def recover_one_download_task(
 
     if stage == DOWNLOAD_STAGE_DOWNLOADING:
         if _is_expired(task.download_lease_until, current_time):
-            await log_task_event(
-                task,
-                event_type=ChatGenerationTaskEventType.DOWNLOAD_RECOVERY_ENQUEUE.value,
-                message=f"{source} 发现 downloading lease 过期，启动时恢复投递下载任务",
-                detail={"payload": redis_payload},
-            )
             await enqueue_download_task(
                 db,
                 task,
                 recover=True,
                 reason=f"{source}_downloading_lease_expired",
             )
+            await log_task_event(
+                task,
+                event_type=ChatGenerationTaskEventType.DOWNLOAD_RECOVERY_ENQUEUE.value,
+                message=f"{source} 发现 downloading lease 过期，启动时恢复投递下载任务",
+                detail={"payload": redis_payload},
+            )
             return "recover_downloading_expired"
 
         await postpone_download_active_check(
-            record_id=task.id,
+            record_id=_chat_registry_id(task),
             payload=payload,
             check_at=task.download_lease_until,
         )
@@ -229,22 +277,22 @@ async def recover_one_download_task(
 
     if stage == DOWNLOAD_STAGE_RETRY_WAITING:
         if _is_expired(task.download_next_retry_at, current_time):
-            await log_task_event(
-                task,
-                event_type=ChatGenerationTaskEventType.DOWNLOAD_RECOVERY_ENQUEUE.value,
-                message=f"{source} 发现 retry_waiting 到期，启动时恢复投递下载任务",
-                detail={"payload": redis_payload},
-            )
             await enqueue_download_task(
                 db,
                 task,
                 recover=True,
                 reason=f"{source}_retry_waiting_due",
             )
+            await log_task_event(
+                task,
+                event_type=ChatGenerationTaskEventType.DOWNLOAD_RECOVERY_ENQUEUE.value,
+                message=f"{source} 发现 retry_waiting 到期，启动时恢复投递下载任务",
+                detail={"payload": redis_payload},
+            )
             return "recover_retry_due"
 
         await postpone_download_active_check(
-            record_id=task.id,
+            record_id=_chat_registry_id(task),
             payload=payload,
             check_at=task.download_next_retry_at,
         )
@@ -267,33 +315,42 @@ async def recover_download_tasks_once(db: AsyncSession) -> dict[str, Any]:
     )
     payloads = await get_download_active_payloads(due_ids)
 
-    for task_id in due_ids:
-        result = await db.execute(
-            select(ChatGenerationTask)
-            .where(
-                ChatGenerationTask.id == task_id,
-                ChatGenerationTask.deleted_at.is_(None),
-            )
-            .with_for_update()
-            .limit(1)
-        )
-        task = result.scalar_one_or_none()
+    download_refs = {
+        registry_item_id: parse_redis_owner_item_id(registry_item_id)
+        for registry_item_id in due_ids
+    }
+    for registry_item_id, ref in download_refs.items():
+        if ref.owner_type != GenerationOwnerType.CHAT_GENERATION_TASK.value:
+            continue
+        task = await _load_chat_task_for_update(db, ref.owner_id)
         if task is None:
-            await remove_download_active(task_id)
+            await db.rollback()
+            await remove_download_active(registry_item_id)
             action = "clean_missing_task"
+        elif (
+            ref.generation_attempt_no is not None
+            and int(task.generation_attempt_no or 1)
+            != int(ref.generation_attempt_no)
+        ):
+            await db.rollback()
+            await remove_download_active(registry_item_id)
+            action = "clean_stale_download_attempt"
         else:
+            current_registry_id = _chat_registry_id(task)
+            if registry_item_id != current_registry_id:
+                await remove_download_active(registry_item_id)
             checked_ids.add(task.id)
             action = await recover_one_download_task(
                 db,
                 task,
-                payload=payloads.get(task_id),
+                payload=payloads.get(registry_item_id),
                 source="startup_redis",
             )
         results[action] = results.get(action, 0) + 1
 
     # DB fallback：不依赖 Redis active 注册表。
     fallback_result = await db.execute(
-        select(ChatGenerationTask)
+        select(ChatGenerationTask.id)
         .where(
             ChatGenerationTask.deleted_at.is_(None),
             ChatGenerationTask.generation_mode.in_(list(ALLOWED_GENERATION_MODES)),
@@ -310,12 +367,15 @@ async def recover_download_tasks_once(db: AsyncSession) -> dict[str, Any]:
         )
         .order_by(ChatGenerationTask.updated_at.asc())
         .limit(int(settings.DOWNLOAD_RECOVERY_BATCH_SIZE or 100))
-        .with_for_update(skip_locked=True)
     )
-    fallback_tasks = fallback_result.scalars().all()
+    fallback_ids = [str(value) for value in fallback_result.scalars().all()]
 
-    for task in fallback_tasks:
-        if task.id in checked_ids:
+    for task_id in fallback_ids:
+        if task_id in checked_ids:
+            continue
+        task = await _load_chat_task_for_update(db, task_id)
+        if task is None:
+            await db.rollback()
             continue
         action = await recover_one_download_task(
             db,
@@ -324,7 +384,7 @@ async def recover_download_tasks_once(db: AsyncSession) -> dict[str, Any]:
             source="startup_db",
         )
         results[action] = results.get(action, 0) + 1
-        checked_ids.add(task.id)
+        checked_ids.add(task_id)
 
     return {"checked": len(checked_ids), "results": results}
 
@@ -341,11 +401,9 @@ async def _mark_timeout(
         error_message=error_message,
         pipeline_stage=ChatGenerationPipelineStage.TIMEOUT.value,
     )
-    await notify_chat_generation_task_finished(db, task)
-    from app.services.generation.ai.task_group_service import aggregate_parent_for_child
-    await aggregate_parent_for_child(db, task)
     await db.commit()
-    await _remove_poll_active(task.id)
+    await notify_owner_finished(db, task)
+    await _remove_poll_active(_chat_registry_id(task))
     await log_task_event(
         task,
         event_type=ChatGenerationTaskEventType.TASK_TIMEOUT.value,
@@ -369,11 +427,9 @@ async def _mark_failed(
         error_message=error_message,
         pipeline_stage=ChatGenerationPipelineStage.FAILED.value,
     )
-    await notify_chat_generation_task_finished(db, task)
-    from app.services.generation.ai.task_group_service import aggregate_parent_for_child
-    await aggregate_parent_for_child(db, task)
     await db.commit()
-    await _remove_poll_active(task.id)
+    await notify_owner_finished(db, task)
+    await _remove_poll_active(_chat_registry_id(task))
     await log_task_event(task, event_type=event_type, message=task.error_message, detail=detail)
     return "mark_failed"
 
@@ -403,19 +459,19 @@ async def recover_one_generation_task(
     if not task:
         return "skip_missing_task"
     if task.generation_mode not in ALLOWED_GENERATION_MODES:
-        await _remove_poll_active(task.id)
+        await _remove_poll_active(_chat_registry_id(task))
         return "clean_invalid_mode"
     if _is_final_task_state(task):
-        await _remove_poll_active(task.id)
+        await _remove_poll_active(_chat_registry_id(task))
         return "clean_final_state"
     if task.status != ChatGenerationTaskStatus.GENERATING.value:
-        await _remove_poll_active(task.id)
+        await _remove_poll_active(_chat_registry_id(task))
         return "clean_not_generating"
 
     if bool(getattr(task, "video_upscale_enabled_snapshot", False)) and str(task.pipeline_stage or "").startswith("upscale_"):
         # 原视频已经进入超分流水线，后续由 video_upscale 恢复扫描处理。
         # 这里禁止再次投递原结果下载，避免覆盖保留的 source.mp4 或提前生成用户资源。
-        await _remove_poll_active(task.id)
+        await _remove_poll_active(_chat_registry_id(task))
         return "delegate_video_upscale_recovery"
 
     has_remote_result = bool(str(task.remote_result_url or "").strip())
@@ -425,7 +481,13 @@ async def recover_one_generation_task(
     # 最高优先级：只要远程结果 URL 已经落库，说明生成侧已经成功。
     # 不管当前 pipeline_stage 是 queued/creating/waiting/result_ready/download_*，恢复时都不能重复 create 或 poll。
     if has_remote_result:
-        await _remove_poll_active(task.id)
+        await _remove_poll_active(_chat_registry_id(task))
+        await enqueue_download_task(
+            db,
+            task,
+            recover=True,
+            reason=f"{source}_has_remote_result_url",
+        )
         await log_task_event(
             task,
             event_type=ChatGenerationTaskEventType.GENERATION_RECOVERY_ENQUEUE.value,
@@ -436,12 +498,6 @@ async def recover_one_generation_task(
                 "deadline_expired": is_deadline_expired,
             },
         )
-        await enqueue_download_task(
-            db,
-            task,
-            recover=True,
-            reason=f"{source}_has_remote_result_url",
-        )
         return "recover_download_has_remote_result"
 
     # 已经过 deadline 且没有结果 URL：
@@ -459,7 +515,7 @@ async def recover_one_generation_task(
             )
             poll_generation_task.apply_async(
                 args=[task.id],
-                kwargs={"force_due": True},
+                kwargs={"force_due": True, "owner_type": GenerationOwnerType.CHAT_GENERATION_TASK.value, "generation_attempt_no": int(task.generation_attempt_no or 1)},
                 queue=POLL_QUEUE,
                 countdown=0,
             )
@@ -470,13 +526,11 @@ async def recover_one_generation_task(
             )
             return "recover_deadline_final_poll"
 
-        await log_task_event(
+        return await _mark_timeout(
+            db,
             task,
-            event_type=ChatGenerationTaskEventType.GENERATION_RECOVERY_TIMEOUT.value,
-            message=f"{source} 发现任务已到 deadline，且没有 remote_result_url/供应商任务ID，按超时失败处理",
-            detail={"pipeline_stage": task.pipeline_stage, "payload": redis_payload},
+            error_message=f"{source} 发现任务已到 deadline，且没有远程结果或供应商任务ID",
         )
-        return await _mark_timeout(db, task)
 
     # 未过 deadline：有供应商任务 ID 才允许恢复到 poll 队列。
     # 视频任务如果 next_poll_at 未到期，不提前 poll，只刷新 active 注册表等待 Beat dispatcher 到期投递。
@@ -523,7 +577,7 @@ async def recover_one_generation_task(
         )
         poll_generation_task.apply_async(
             args=[task.id],
-            kwargs={"force_due": True},
+            kwargs={"force_due": True, "owner_type": GenerationOwnerType.CHAT_GENERATION_TASK.value, "generation_attempt_no": int(task.generation_attempt_no or 1)},
             queue=POLL_QUEUE,
             countdown=0,
         )
@@ -552,9 +606,11 @@ async def recover_one_generation_task(
             ChatGenerationPipelineStage.CREATING_PROVIDER_TASK.value,
         ):
             task.pipeline_stage = ChatGenerationPipelineStage.QUEUED.value
-            await db.commit()
+        # Release the recovery row lock before writing an event through the
+        # independent logging session or talking to the broker.
+        await db.commit()
 
-        await _remove_poll_active(task.id)
+        await _remove_poll_active(_chat_registry_id(task))
         await log_task_event(
             task,
             event_type=ChatGenerationTaskEventType.GENERATION_RECOVERY_ENQUEUE.value,
@@ -563,6 +619,7 @@ async def recover_one_generation_task(
         )
         chatapi_create_generation_task.apply_async(
             args=[task.id],
+            kwargs={"owner_type": GenerationOwnerType.CHAT_GENERATION_TASK.value, "generation_attempt_no": int(task.generation_attempt_no or 1)},
             queue=CeleryQueue.GEN_CHATAPI_CREATE.value,
             countdown=0,
         )
@@ -572,7 +629,7 @@ async def recover_one_generation_task(
     if task.pipeline_stage == ChatGenerationPipelineStage.RESULT_READY.value:
         task.pipeline_stage = ChatGenerationPipelineStage.QUEUED.value
         await db.commit()
-        await _remove_poll_active(task.id)
+        await _remove_poll_active(_chat_registry_id(task))
         await log_task_event(
             task,
             event_type=ChatGenerationTaskEventType.GENERATION_RECOVERY_ENQUEUE.value,
@@ -581,6 +638,7 @@ async def recover_one_generation_task(
         )
         chatapi_create_generation_task.apply_async(
             args=[task.id],
+            kwargs={"owner_type": GenerationOwnerType.CHAT_GENERATION_TASK.value, "generation_attempt_no": int(task.generation_attempt_no or 1)},
             queue=CeleryQueue.GEN_CHATAPI_CREATE.value,
             countdown=0,
         )
@@ -621,28 +679,33 @@ async def recover_generation_tasks_once(db: AsyncSession) -> dict[str, Any]:
         if image_main_cursor:
             image_main_query = image_main_query.where(ChatGenerationTask.id > image_main_cursor)
         image_main_result = await db.execute(
-            image_main_query.order_by(ChatGenerationTask.id.asc())
+            image_main_query.with_only_columns(ChatGenerationTask.id)
+            .order_by(ChatGenerationTask.id.asc())
             .limit(image_main_batch_size)
-            .with_for_update()
         )
-        image_mains = list(image_main_result.scalars().all())
-        if not image_mains:
+        image_main_ids = [str(value) for value in image_main_result.scalars().all()]
+        if not image_main_ids:
             break
 
-        for main in image_mains:
-            main_id = str(main.id)
+        child_parent_result = await db.execute(
+            select(ChatGenerationTask.parent_task_id)
+            .where(
+                ChatGenerationTask.parent_task_id.in_(image_main_ids),
+                ChatGenerationTask.generation_mode == GenerationMode.CHATAPI_CHILD.value,
+            )
+            .distinct()
+        )
+        split_parent_ids = {str(value) for value in child_parent_result.scalars().all() if value}
+
+        for main_id in image_main_ids:
             image_main_cursor = main_id
             checked_ids.add(main_id)
+            main = await _load_chat_task_for_update(db, main_id)
+            if main is None:
+                await db.rollback()
+                continue
 
-            child_result = await db.execute(
-                select(ChatGenerationTask.id)
-                .where(
-                    ChatGenerationTask.parent_task_id == main_id,
-                    ChatGenerationTask.generation_mode == GenerationMode.CHATAPI_CHILD.value,
-                )
-                .limit(1)
-            )
-            if child_result.scalar_one_or_none() is not None:
+            if main_id in split_parent_ids:
                 main.provider_create_claim_token = None
                 main.provider_create_lease_until = None
                 await db.commit()
@@ -653,7 +716,7 @@ async def recover_generation_tasks_once(db: AsyncSession) -> dict[str, Any]:
             lease_until = ensure_aware_utc(main.provider_create_lease_until)
             lease_alive = bool(main.provider_create_claim_token and lease_until and lease_until > now)
             if lease_alive:
-                await db.commit()
+                await db.rollback()
                 results["image_main_claim_alive"] = results.get("image_main_claim_alive", 0) + 1
                 continue
 
@@ -670,19 +733,24 @@ async def recover_generation_tasks_once(db: AsyncSession) -> dict[str, Any]:
                 results["image_main_timeout"] = results.get("image_main_timeout", 0) + 1
                 continue
 
+            claim_expired = False
             if main.provider_create_claim_token or main.provider_create_lease_until:
                 main.provider_create_claim_token = None
                 main.provider_create_lease_until = None
                 main.pipeline_stage = ChatGenerationPipelineStage.QUEUED.value
+                claim_expired = True
+            attempt_no = int(main.generation_attempt_no or 1)
+            await db.commit()
+            if claim_expired:
                 await log_task_event(
                     main,
                     event_type=ChatGenerationTaskEventType.IMAGE_MAIN_CLAIM_EXPIRED.value,
                     message="图片主任务供应商执行租约已过期，恢复重新投递",
                 )
-            await db.commit()
             try:
                 chatapi_create_generation_task.apply_async(
                     args=[main_id],
+                    kwargs={"owner_type": GenerationOwnerType.CHAT_GENERATION_TASK.value, "generation_attempt_no": attempt_no},
                     queue=CeleryQueue.GEN_CHATAPI_CREATE.value,
                     countdown=0,
                 )
@@ -691,7 +759,7 @@ async def recover_generation_tasks_once(db: AsyncSession) -> dict[str, Any]:
                 logger.exception("恢复投递图片主任务失败 task_id=%s: %s", main_id, exc)
                 results["recover_image_main_enqueue_failed"] = results.get("recover_image_main_enqueue_failed", 0) + 1
 
-        if len(image_mains) < image_main_batch_size:
+        if len(image_main_ids) < image_main_batch_size:
             break
 
     due_poll_ids = await redis_get_due_registry_ids(
@@ -705,26 +773,34 @@ async def recover_generation_tasks_once(db: AsyncSession) -> dict[str, Any]:
         log_context="poll_active",
     )
 
-    for task_id in due_poll_ids:
-        result = await db.execute(
-            select(ChatGenerationTask)
-            .where(
-                ChatGenerationTask.id == task_id,
-                ChatGenerationTask.deleted_at.is_(None),
-            )
-            .with_for_update()
-            .limit(1)
-        )
-        task = result.scalar_one_or_none()
+    poll_refs = {
+        registry_item_id: parse_redis_owner_item_id(registry_item_id)
+        for registry_item_id in due_poll_ids
+    }
+    for registry_item_id, ref in poll_refs.items():
+        if ref.owner_type != GenerationOwnerType.CHAT_GENERATION_TASK.value:
+            continue
+        task = await _load_chat_task_for_update(db, ref.owner_id)
         if task is None:
-            await _remove_poll_active(task_id)
+            await db.rollback()
+            await _remove_poll_active(registry_item_id)
             action = "clean_missing_poll_task"
+        elif (
+            ref.generation_attempt_no is not None
+            and int(task.generation_attempt_no or 1) != int(ref.generation_attempt_no)
+        ):
+            await db.rollback()
+            await _remove_poll_active(registry_item_id)
+            action = "clean_stale_poll_attempt"
         else:
+            current_registry_id = _chat_registry_id(task)
+            if registry_item_id != current_registry_id:
+                await _remove_poll_active(registry_item_id)
             checked_ids.add(task.id)
             action = await recover_one_generation_task(
                 db,
                 task,
-                payload=poll_payloads.get(task_id),
+                payload=poll_payloads.get(registry_item_id),
                 source="startup_poll_redis",
             )
         results[action] = results.get(action, 0) + 1
@@ -735,7 +811,7 @@ async def recover_generation_tasks_once(db: AsyncSession) -> dict[str, Any]:
 
     for _round in range(max_rounds):
         query_result = await db.execute(
-            select(ChatGenerationTask)
+            select(ChatGenerationTask.id)
             .where(
                 ChatGenerationTask.deleted_at.is_(None),
                 ChatGenerationTask.generation_mode.in_(list(ALLOWED_GENERATION_MODES)),
@@ -753,15 +829,18 @@ async def recover_generation_tasks_once(db: AsyncSession) -> dict[str, Any]:
             )
             .order_by(ChatGenerationTask.updated_at.asc())
             .limit(batch_size)
-            .with_for_update(skip_locked=True)
         )
-        tasks = query_result.scalars().all()
-        if not tasks:
+        task_ids = [str(value) for value in query_result.scalars().all()]
+        if not task_ids:
             break
 
         progressed_this_round = 0
-        for task in tasks:
-            if task.id in checked_ids:
+        for task_id in task_ids:
+            if task_id in checked_ids:
+                continue
+            task = await _load_chat_task_for_update(db, task_id)
+            if task is None:
+                await db.rollback()
                 continue
             action = await recover_one_generation_task(
                 db,
@@ -770,11 +849,11 @@ async def recover_generation_tasks_once(db: AsyncSession) -> dict[str, Any]:
                 source="startup_db",
             )
             results[action] = results.get(action, 0) + 1
-            checked_ids.add(task.id)
+            checked_ids.add(task_id)
             total_db_checked += 1
             progressed_this_round += 1
 
-        if len(tasks) < batch_size or progressed_this_round <= 0:
+        if len(task_ids) < batch_size or progressed_this_round <= 0:
             break
 
     # 子任务可能在 worker 中断前已进入终态但主任务尚未汇总，按稳定游标完整重算全部主任务。
@@ -820,6 +899,7 @@ async def dispatch_due_poll_tasks_once(db: AsyncSession) -> dict[str, Any]:
     batch_size = max(1, int(settings.POLL_DUE_DISPATCH_BATCH_SIZE or 100))
     poll_lease_expired_at = current_time - timedelta(seconds=int(settings.POLL_TASK_LEASE_SECONDS or 300))
 
+    await apply_short_lock_timeout(db)
     query_result = await db.execute(
         select(ChatGenerationTask)
         .where(
@@ -846,6 +926,7 @@ async def dispatch_due_poll_tasks_once(db: AsyncSession) -> dict[str, Any]:
     dispatched_task_ids: list[str] = []
     dispatched_due_next_poll_at_by_id: dict[str, datetime | None] = {}
     dispatched_queue_hold_until_by_id: dict[str, datetime] = {}
+    post_commit_logs: list[dict[str, Any]] = []
 
     for task in tasks:
         action = "skip_unknown"
@@ -863,13 +944,14 @@ async def dispatch_due_poll_tasks_once(db: AsyncSession) -> dict[str, Any]:
                     continue
 
             if not (task.seedance_task_id or task.provider_task_id):
-                # dispatcher 不负责重新 create；没有 provider id 的异常状态交给启动容灾或 create 任务处理。
-                await log_task_event(
-                    task,
-                    event_type=ChatGenerationTaskEventType.POLL_DISPATCH_SKIP.value,
-                    message="视频到期轮询调度跳过：缺少外部任务ID",
-                    detail={"pipeline_stage": task.pipeline_stage, "next_poll_at": task.next_poll_at},
-                )
+                # Defer FK-backed event logging until the batch row locks are released.
+                post_commit_logs.append({
+                    "task_id": str(task.id),
+                    "generation_attempt_no": int(task.generation_attempt_no or 1),
+                    "event_type": ChatGenerationTaskEventType.POLL_DISPATCH_SKIP.value,
+                    "message": "视频到期轮询调度跳过：缺少外部任务ID",
+                    "detail": {"pipeline_stage": task.pipeline_stage, "next_poll_at": task.next_poll_at},
+                })
                 action = "skip_no_provider_task_id"
                 continue
 
@@ -886,16 +968,28 @@ async def dispatch_due_poll_tasks_once(db: AsyncSession) -> dict[str, Any]:
         except Exception as exc:
             logger.exception("视频到期轮询调度单条处理失败。task_id=%s", getattr(task, "id", None))
             action = "error"
-            await log_task_event(
-                task,
-                event_type=ChatGenerationTaskEventType.POLL_DISPATCH_SKIP.value,
-                message=f"视频到期轮询调度单条处理失败：{exc}",
-            )
+            post_commit_logs.append({
+                "task_id": str(getattr(task, "id", "") or ""),
+                "generation_attempt_no": int(getattr(task, "generation_attempt_no", 1) or 1),
+                "event_type": ChatGenerationTaskEventType.POLL_DISPATCH_SKIP.value,
+                "message": f"视频到期轮询调度单条处理失败：{exc}",
+                "detail": None,
+            })
         finally:
             results[action] = results.get(action, 0) + 1
 
     await db.commit()
 
+    for item in post_commit_logs:
+        if item["task_id"]:
+            await log_task_event(
+                task_id=item["task_id"],
+                generation_attempt_no=item["generation_attempt_no"],
+                event_type=item["event_type"],
+                message=item["message"],
+                detail=item["detail"],
+            )
+
     fresh_tasks = []
     if dispatched_task_ids:
         fresh_result = await db.execute(
@@ -931,7 +1025,7 @@ async def dispatch_due_poll_tasks_once(db: AsyncSession) -> dict[str, Any]:
         )
         poll_generation_task.apply_async(
             args=[task.id],
-            kwargs={"force_due": True},
+            kwargs={"force_due": True, "owner_type": GenerationOwnerType.CHAT_GENERATION_TASK.value, "generation_attempt_no": int(task.generation_attempt_no or 1)},
             queue=POLL_QUEUE,
             countdown=0,
         )

video-gen-api/app/services/generation/refund_service.py

@@ -7,6 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.models.chat_generation_task import ChatGenerationTask
 from app.models.credit_record import CreditRecord
 from app.models.generation_record import GenerationRecord
+from app.services.generation.pipeline.db_lock_service import execute_with_lock_timeout
 from app.services.credits import refund_credits
 from app.services.credit_record_meta_service import build_refund_meta_from_charge
 from app.services.generation.billing_service import (
@@ -25,46 +26,52 @@ def _round2(value: float | int | None) -> float:
     return round(float(value or 0), 2)
 
 
-async def _has_refund_for_biz_key(db: AsyncSession, *, user_id: str, charge_biz_key: str) -> bool:
-    result = await db.execute(
-        select(CreditRecord.id)
-        .where(
-            CreditRecord.user_id == user_id,
-            CreditRecord.type == "refund",
-            CreditRecord.refund_for_biz_key == charge_biz_key,
-        )
-        .limit(1)
-    )
-    return result.scalar_one_or_none() is not None
-
-
 async def _find_unrefunded_media_charges(
     db: AsyncSession,
     *,
     user_id: str,
     owner_type: str,
     owner_id: str,
+    generation_attempt_no: int | None = None,
 ) -> list[CreditRecord]:
-    """查找当前任务下所有未退款的媒体生成扣费流水。"""
-    pattern = f"{owner_type}:{owner_id}:attempt:%:{CHARGE_MEDIA}:charge"
+    """批量查找指定任务、指定生成轮次下尚未退款的媒体扣费。"""
     result = await db.execute(
         select(CreditRecord)
         .where(
             CreditRecord.user_id == user_id,
             CreditRecord.related_id == owner_id,
             CreditRecord.type == "consume",
-            CreditRecord.biz_key.like(pattern),
+            CreditRecord.biz_key.is_not(None),
         )
         .order_by(CreditRecord.created_at.asc())
     )
-    charges = list(result.scalars().all())
-    unrefunded: list[CreditRecord] = []
-    for charge in charges:
-        if not charge.biz_key:
+
+    charges: list[CreditRecord] = []
+    for charge in result.scalars().all():
+        parsed = parse_credit_biz_key(charge.biz_key)
+        if not parsed:
             continue
-        if not await _has_refund_for_biz_key(db, user_id=user_id, charge_biz_key=charge.biz_key):
-            unrefunded.append(charge)
-    return unrefunded
+        if parsed.get("owner_type") != owner_type or parsed.get("owner_id") != owner_id:
+            continue
+        if parsed.get("charge_kind") != CHARGE_MEDIA or parsed.get("action") != "charge":
+            continue
+        if generation_attempt_no is not None and int(parsed.get("attempt_no") or 0) != int(generation_attempt_no):
+            continue
+        charges.append(charge)
+
+    charge_keys = [str(charge.biz_key) for charge in charges if charge.biz_key]
+    if not charge_keys:
+        return []
+
+    refund_result = await db.execute(
+        select(CreditRecord.refund_for_biz_key).where(
+            CreditRecord.user_id == user_id,
+            CreditRecord.type == "refund",
+            CreditRecord.refund_for_biz_key.in_(charge_keys),
+        )
+    )
+    refunded_keys = {str(value) for (value,) in refund_result.all() if value}
+    return [charge for charge in charges if charge.biz_key not in refunded_keys]
 
 
 async def refund_unrefunded_media_charges(
@@ -74,8 +81,9 @@ async def refund_unrefunded_media_charges(
     owner_type: str,
     owner_id: str,
     description_prefix: str,
+    generation_attempt_no: int | None = None,
 ) -> float:
-    """回退当前任务所有未退款媒体扣费流水。
+    """回退当前生成轮次尚未退款的媒体扣费流水。
 
     容灾考虑：
     - 不依赖 retry_count 推断当前轮次。
@@ -88,6 +96,7 @@ async def refund_unrefunded_media_charges(
         user_id=user_id,
         owner_type=owner_type,
         owner_id=owner_id,
+        generation_attempt_no=generation_attempt_no,
     )
     for charge in charges:
         parsed = parse_credit_biz_key(charge.biz_key)
@@ -124,6 +133,7 @@ async def mark_generation_record_failed_and_refund_once(
     record_id: str | None = None,
     record: GenerationRecord | None = None,
     error_message: str | None = None,
+    generation_attempt_no: int | None = None,
 ) -> GenerationRecord | None:
     """把 GenerationRecord 标记为最终失败并幂等退回媒体生成积分。
 
@@ -132,7 +142,9 @@ async def mark_generation_record_failed_and_refund_once(
     if record is None:
         if not record_id:
             return None
-        result = await db.execute(
+        result = await execute_with_lock_timeout(
+            db,
+
             select(GenerationRecord)
             .where(GenerationRecord.id == record_id, GenerationRecord.deleted_at.is_(None))
             .with_for_update()
@@ -155,6 +167,7 @@ async def mark_generation_record_failed_and_refund_once(
         owner_type=OWNER_GENERATION_RECORD,
         owner_id=record.id,
         description_prefix="生成记录",
+        generation_attempt_no=generation_attempt_no or int(getattr(record, "generation_attempt_no", 1) or 1),
     )
     if refunded_amount > 0:
         # GenerationRecord.credits_cost 只代表视频/图片生成媒体积分。
@@ -171,12 +184,15 @@ async def mark_chat_generation_task_failed_and_refund_once(
     task: ChatGenerationTask | None = None,
     error_message: str | None = None,
     pipeline_stage: str = "failed",
+    generation_attempt_no: int | None = None,
 ) -> ChatGenerationTask | None:
     """把 ChatGenerationTask 标记为最终失败并幂等退回媒体生成积分。"""
     if task is None:
         if not task_id:
             return None
-        result = await db.execute(
+        result = await execute_with_lock_timeout(
+            db,
+
             select(ChatGenerationTask)
             .where(
                 ChatGenerationTask.id == task_id,
@@ -204,6 +220,7 @@ async def mark_chat_generation_task_failed_and_refund_once(
         owner_type=OWNER_CHAT_GENERATION_TASK,
         owner_id=task.id,
         description_prefix="任务生成",
+        generation_attempt_no=generation_attempt_no or int(getattr(task, "generation_attempt_no", 1) or 1),
     )
 
     if refunded_amount > 0:

video-gen-api/app/services/generation/task_factory_service.py

@@ -132,6 +132,9 @@ async def create_chat_generation_task_for_module(
         snapshot["generation_count"] = 1
         task = ChatGenerationTask(
             id=task_id,
+            created_at=now,
+            resource_generation_started_at=now,
+            generation_attempt_no=1,
             user_id=current_user.id,
             original_prompt=original_prompt,
             optimized_prompt=optimized_prompt,
@@ -195,6 +198,9 @@ async def create_chat_generation_task_for_module(
         snapshot["video_upscale_enabled_snapshot"] = upscale_enabled_snapshot
         task = ChatGenerationTask(
             id=task_id,
+            created_at=now,
+            resource_generation_started_at=now,
+            generation_attempt_no=1,
             user_id=current_user.id,
             original_prompt=original_prompt,
             optimized_prompt=optimized_prompt,