diff --git a/.gitignore b/.gitignore
index 692a453b..7d9e5bf9 100644
--- a/.gitignore
+++ b/.gitignore
@@ -10,6 +10,7 @@ __pycache__/
 .trae/
 # video-gen-app/dist/
 video-gen-api/dist/
+bak/
 
 # 使用通配符
 *.log
diff --git a/video-gen-api/app/api/v1/generation_ai.py b/video-gen-api/app/api/v1/generation_ai.py
index 309798eb..b32952df 100644
--- a/video-gen-api/app/api/v1/generation_ai.py
+++ b/video-gen-api/app/api/v1/generation_ai.py
@@ -271,6 +271,7 @@ async def list_tasks(
         "必须通过 gen_type 区分图片和视频。"
         "分页对象是生成日期，不是单条记录。"
         "每页最多返回10个生成日期分组，每个日期分组内最多返回该日期下倒序前10条生成记录。"
+        "每条历史记录会返回 generated_resource_id，表示 generated_resources.id；历史脏数据可能为空。"
         "如果某一天 total 大于10，前端可调用 /generation-ai/history/{generated_date} 加载该日期下的后续分页数据。"
     ),
     responses={
@@ -336,6 +337,7 @@ async def list_history_grouped_days(
         "该接口用于前端在历史分组列表中继续加载某一天的后续记录。"
         "例如 /history 接口中某一天 total=18，但 items 只返回前10条，"
         "则前端可以调用本接口 page=2&page_size=10 获取该日期下剩余记录。"
+        "每条历史记录会返回 generated_resource_id，表示 generated_resources.id；历史脏数据可能为空。"
         "该接口同样必须通过 gen_type 区分图片和视频。"
     ),
     responses={
diff --git a/video-gen-api/app/schemas/generation_ai.py b/video-gen-api/app/schemas/generation_ai.py
index 1bf46826..6d7b167c 100644
--- a/video-gen-api/app/schemas/generation_ai.py
+++ b/video-gen-api/app/schemas/generation_ai.py
@@ -253,6 +253,7 @@ class GenerationAITaskOut(BaseModel):
                 "user_id": None,
                 "user_name": None,
                 "project_id": None,
+                "generated_resource_id": "generated_resource_xxx",
                 "gen_type": "image",
                 "generation_mode": "chatapi_async",
                 "pipeline_stage": "done",
@@ -316,6 +317,10 @@ class GenerationAITaskOut(BaseModel):
         None,
         description="项目ID。当前 /generation-ai 任务不绑定项目，通常为 null",
     )
+    generated_resource_id: str | None = Field(
+        None,
+        description="关联的生成资源账本ID，来源于 generated_resources.id；历史脏数据可能为空",
+    )
     gen_type: str = Field(..., description="生成类型：image=图片，video=视频")
     generation_mode: str | None = Field(
         None,
@@ -483,6 +488,7 @@ class GenerationAIRecordHistoryItemOut(BaseModel):
                 "source_type": "generation_record",
                 "project_id": "project_xxx",
                 "project_name": "默认项目",
+                "generated_resource_id": "generated_resource_xxx",
                 "gen_type": "image",
                 "generation_mode": "generation_record",
                 "pipeline_stage": None,
@@ -526,6 +532,10 @@ class GenerationAIRecordHistoryItemOut(BaseModel):
     )
     project_id: str | None = Field(None, description="旧项目ID，来源于 generation_records.project_id")
     project_name: str | None = Field(None, description="旧项目名称，来源于 projects.name；项目不存在时为空")
+    generated_resource_id: str | None = Field(
+        None,
+        description="关联的生成资源账本ID，来源于 generated_resources.id；历史脏数据可能为空",
+    )
     gen_type: str = Field(..., description="生成类型：image=图片，video=视频")
     generation_mode: str | None = Field(
         "generation_record",
diff --git a/video-gen-api/app/services/generation_ai_service.py b/video-gen-api/app/services/generation_ai_service.py
index 424d938a..cb158698 100644
--- a/video-gen-api/app/services/generation_ai_service.py
+++ b/video-gen-api/app/services/generation_ai_service.py
@@ -28,7 +28,12 @@ from app.services.generation_billing_service import (
     OWNER_CHAT_GENERATION_TASK,
     charge_generation_media_by_params,
 )
-from app.services.resource_accounting_service import soft_delete_chat_task_resources
+from app.services.resource_accounting_service import (
+    SOURCE_MODEL_CHAT_TASK,
+    SOURCE_MODEL_GENERATION_RECORD,
+    batch_get_generated_resource_id_map,
+    soft_delete_chat_task_resources,
+)
 from app.services.resource_signed_url_service import build_resource_signed_url
 from app.utils.id_gen import generate_id
 
@@ -320,7 +325,11 @@ async def create_async_generation_task(db: AsyncSession, current_user: User, req
     return task
 
 
-def record_to_out(task: ChatGenerationTask, is_admin: bool = False) -> GenerationAITaskOut:
+def record_to_out(
+    task: ChatGenerationTask,
+    is_admin: bool = False,
+    generated_resource_id: str | None = None,
+) -> GenerationAITaskOut:
     refs = _parse_json(task.media_references)
     snapshot = engine_snapshot_out(_parse_json(task.engine_snapshot_json))
     return GenerationAITaskOut(
@@ -328,6 +337,7 @@ def record_to_out(task: ChatGenerationTask, is_admin: bool = False) -> Generatio
         user_id=task.user_id if is_admin else None,
         user_name=getattr(task, "username", None) if is_admin else None,
         project_id=None,
+        generated_resource_id=generated_resource_id,
         gen_type=task.gen_type,
         generation_mode=task.generation_mode,
         pipeline_stage=task.pipeline_stage,
@@ -499,6 +509,7 @@ def _generation_record_history_base_filters(user_id: str, gen_type: str):
 def generation_record_to_history_out(
     record: GenerationRecord,
     project_name: str | None = None,
+    generated_resource_id: str | None = None,
 ) -> GenerationAIRecordHistoryItemOut:
     refs = _parse_json(record.media_references)
     return GenerationAIRecordHistoryItemOut(
@@ -506,6 +517,7 @@ def generation_record_to_history_out(
         source_type="generation_record",
         project_id=record.project_id,
         project_name=project_name,
+        generated_resource_id=generated_resource_id,
         gen_type=record.gen_type,
         generation_mode="generation_record",
         pipeline_stage=None,
@@ -586,7 +598,8 @@ async def list_generation_record_history_grouped_days(
     )
     day_rows = day_rows_result.all()
 
-    groups = []
+    raw_groups = []
+    all_record_ids: list[str] = []
     for generated_day, day_total in day_rows:
         item_result = await db.execute(
             select(GenerationRecord, Project.name.label("project_name"))
@@ -599,17 +612,31 @@ async def list_generation_record_history_grouped_days(
             .limit(HISTORY_GROUP_ITEM_LIMIT)
         )
         rows = item_result.all()
+        raw_groups.append((generated_day, day_total, rows))
+        all_record_ids.extend(record.id for record, _project_name in rows)
 
-        groups.append(
-            {
-                "generated_date": _history_day_to_str(generated_day),
-                "total": int(day_total or 0),
-                "items": [
-                    generation_record_to_history_out(record, project_name)
-                    for record, project_name in rows
-                ],
-            }
-        )
+    resource_id_map = await batch_get_generated_resource_id_map(
+        db,
+        source_model=SOURCE_MODEL_GENERATION_RECORD,
+        source_ids=all_record_ids,
+        resource_type=gen_type,
+    )
+
+    groups = [
+        {
+            "generated_date": _history_day_to_str(generated_day),
+            "total": int(day_total or 0),
+            "items": [
+                generation_record_to_history_out(
+                    record,
+                    project_name,
+                    generated_resource_id=resource_id_map.get(record.id),
+                )
+                for record, project_name in rows
+            ],
+        }
+        for generated_day, day_total, rows in raw_groups
+    ]
 
     return {
         "total_days": int(total_days or 0),
@@ -660,6 +687,12 @@ async def list_generation_record_history_day_items(
     )
 
     rows = result.all()
+    resource_id_map = await batch_get_generated_resource_id_map(
+        db,
+        source_model=SOURCE_MODEL_GENERATION_RECORD,
+        source_ids=[record.id for record, _project_name in rows],
+        resource_type=gen_type,
+    )
 
     return {
         "generated_date": target_day.strftime("%Y-%m-%d"),
@@ -667,7 +700,11 @@ async def list_generation_record_history_day_items(
         "page": page,
         "page_size": page_size,
         "items": [
-            generation_record_to_history_out(record, project_name)
+            generation_record_to_history_out(
+                record,
+                project_name,
+                generated_resource_id=resource_id_map.get(record.id),
+            )
             for record, project_name in rows
         ],
     }
@@ -729,7 +766,8 @@ async def list_generation_history_grouped_days(
     )
     day_rows = day_rows_result.all()
 
-    groups = []
+    raw_groups = []
+    all_task_ids: list[str] = []
     for generated_day, day_total in day_rows:
         item_result = await db.execute(
             select(ChatGenerationTask)
@@ -741,14 +779,30 @@ async def list_generation_history_grouped_days(
             .limit(HISTORY_GROUP_ITEM_LIMIT)
         )
         tasks = list(item_result.scalars().all())
+        raw_groups.append((generated_day, day_total, tasks))
+        all_task_ids.extend(task.id for task in tasks)
 
-        groups.append(
-            {
-                "generated_date": _history_day_to_str(generated_day),
-                "total": int(day_total or 0),
-                "items": [record_to_out(task) for task in tasks],
-            }
-        )
+    resource_id_map = await batch_get_generated_resource_id_map(
+        db,
+        source_model=SOURCE_MODEL_CHAT_TASK,
+        source_ids=all_task_ids,
+        resource_type=gen_type,
+    )
+
+    groups = [
+        {
+            "generated_date": _history_day_to_str(generated_day),
+            "total": int(day_total or 0),
+            "items": [
+                record_to_out(
+                    task,
+                    generated_resource_id=resource_id_map.get(task.id),
+                )
+                for task in tasks
+            ],
+        }
+        for generated_day, day_total, tasks in raw_groups
+    ]
 
     return {
         "total_days": int(total_days or 0),
@@ -812,13 +866,25 @@ async def list_generation_history_day_items(
     )
 
     tasks = list(result.scalars().all())
+    resource_id_map = await batch_get_generated_resource_id_map(
+        db,
+        source_model=SOURCE_MODEL_CHAT_TASK,
+        source_ids=[task.id for task in tasks],
+        resource_type=gen_type,
+    )
 
     return {
         "generated_date": target_day.strftime("%Y-%m-%d"),
         "total": int(total or 0),
         "page": page,
         "page_size": page_size,
-        "items": [record_to_out(task) for task in tasks],
+        "items": [
+            record_to_out(
+                task,
+                generated_resource_id=resource_id_map.get(task.id),
+            )
+            for task in tasks
+        ],
     }
 
 async def soft_delete_chat_generation_task(
diff --git a/video-gen-api/app/services/module_generation_flow_base_service.py b/video-gen-api/app/services/module_generation_flow_base_service.py
index 2d876d72..1f53cb8f 100644
--- a/video-gen-api/app/services/module_generation_flow_base_service.py
+++ b/video-gen-api/app/services/module_generation_flow_base_service.py
@@ -195,6 +195,35 @@ async def create_module_step(
     return step
 
 
+def _clear_project_final_resources_by_deleted_steps(
+    project: ModuleGenerationProject,
+    *,
+    deleted_step_codes: set[str],
+    config: ModuleGenerationFlowConfig,
+) -> list[str]:
+    """根据被软删的步骤，清空项目表中对应的最终资源字段。
+
+    第 3 步图片生成被软删时，下游第 5 步视频也基于旧图片失效，所以图片、视频、封面都要清空。
+    第 5 步视频生成被软删时，只清空视频和封面，保留第 3 步图片结果。
+    """
+    cleared_fields: list[str] = []
+
+    def clear_field(field_name: str) -> None:
+        if getattr(project, field_name, None):
+            cleared_fields.append(field_name)
+        setattr(project, field_name, None)
+
+    if config.image_generate_step_code in deleted_step_codes:
+        clear_field("final_image_url")
+        clear_field("final_video_url")
+        clear_field("final_video_cover_url")
+    elif config.video_generate_step_code in deleted_step_codes:
+        clear_field("final_video_url")
+        clear_field("final_video_cover_url")
+
+    return cleared_fields
+
+
 async def soft_delete_steps_from_index(
     db: AsyncSession,
     *,
@@ -217,6 +246,13 @@ async def soft_delete_steps_from_index(
         .with_for_update()
     )
     steps = list(result.scalars().all())
+    deleted_step_codes = {step.step_code for step in steps}
+    cleared_project_fields = _clear_project_final_resources_by_deleted_steps(
+        project,
+        deleted_step_codes=deleted_step_codes,
+        config=config,
+    )
+
     for step in steps:
         step.is_current = False
         step.deleted_at = deleted_at
@@ -245,7 +281,11 @@ async def soft_delete_steps_from_index(
             project=project,
             event_type=ModuleEventTypeEnum.SOFT_DELETE_STEPS.value,
             message=f"软删除第 {start_index} 步及之后的旧子任务",
-            detail={"step_ids": [step.id for step in steps]},
+            detail={
+                "step_ids": [step.id for step in steps],
+                "step_codes": [step.step_code for step in steps],
+                "cleared_project_fields": cleared_project_fields,
+            },
         )
     return steps
 
diff --git a/video-gen-api/app/services/resource_accounting_service.py b/video-gen-api/app/services/resource_accounting_service.py
index b6bc20ac..cf798d59 100644
--- a/video-gen-api/app/services/resource_accounting_service.py
+++ b/video-gen-api/app/services/resource_accounting_service.py
@@ -333,6 +333,45 @@ async def record_generation_record_generated_resource(
     )
 
 
+async def batch_get_generated_resource_id_map(
+    db: AsyncSession,
+    *,
+    source_model: str,
+    source_ids: Sequence[str] | Iterable[str],
+    resource_type: str | None = None,
+) -> dict[str, str]:
+    """批量查询来源记录对应的 GeneratedResource.id。
+
+    用于历史列表接口批量回填资源账本 ID，避免按记录一条条查询。
+    如果历史脏数据存在同一个 source_id 对应多条未软删资源账本，按 created_at 倒序取最新一条。
+    """
+    ids = list(dict.fromkeys(str(item) for item in source_ids if item))
+    if not ids:
+        return {}
+
+    normalized_resource_type = (resource_type or "").lower().strip()
+    query = select(GeneratedResource).where(
+        GeneratedResource.source_model == source_model,
+        GeneratedResource.source_id.in_(ids),
+        GeneratedResource.deleted_at.is_(None),
+    )
+    if normalized_resource_type in ("image", "video"):
+        query = query.where(GeneratedResource.resource_type == normalized_resource_type)
+
+    result = await db.execute(
+        query.order_by(
+            GeneratedResource.source_id.asc(),
+            GeneratedResource.created_at.desc(),
+        )
+    )
+
+    resource_id_map: dict[str, str] = {}
+    for resource in result.scalars().all():
+        if resource.source_id not in resource_id_map:
+            resource_id_map[resource.source_id] = resource.id
+    return resource_id_map
+
+
 async def soft_delete_resources_by_source(
     db: AsyncSession,
     *,