fix: 确认生成 API 改为复用 worker.generate_video 渲染路径（适配 develop 分支）

从 main 分支 PR #1308 手动适配到 develop 分支，保留 develop 现有功能：

Schema 层：
- 新增 ConfirmGenerationRequest schema
- CreateGenerationTaskRequest 增加 6 个新字段（is_preview, source_task_id, output_width, output_height, cover_url, custom_title）
- GenerationTaskResponse 增加对应字段

Domain 层：
- GenerationTask 实体增加 source_task_id, output_width, output_height, cover_url, custom_title 字段
- create() 工厂方法增加对应参数

Application 层：
- CreateGenerationTaskCommand 增加对应字段
- UseCase 执行时传递新字段

DB 层：
- GenerationTaskModel 增加 5 个新列（is_preview 已存在于 develop）
- 仓储 _to_domain/create/update 映射更新

API 路由：
- 新增 POST /tasks/{task_id}/confirm 端点
- 使用 safe_enqueue_generation_task 适配 develop 任务入队机制
- 更新 _to_generation_task_response 和 retry 端点

Worker 层：
- _load_task_info 返回新字段
- generate_video 支持动态分辨率（output_width/output_height 覆盖默认分辨率）

Alembic 迁移 054：
- generation_tasks 表增加 source_task_id, output_width, output_height, cover_url, custom_title 列

单元测试：
- tests/unit/test_confirm_generation.py（7 个测试用例全部通过）
- 适配 develop 的 safe_enqueue_generation_task 入队机制

alembic/versions/054_add_confirm_generation_fields.py

@@ -0,0 +1,81 @@
+"""确认生成 API 改造：为 generation_tasks 表添加 source_task_id、output_width、output_height、cover_url、custom_title 字段
+
+Revision ID: 054_add_confirm_generation_fields
+Revises: 053_generation_task_is_preview
+Create Date: 2026-08-16
+
+Changes:
+1. generation_tasks 表新增 source_task_id（来源预览任务 ID，带索引）
+2. generation_tasks 表新增 output_width / output_height（动态输出分辨率）
+3. generation_tasks 表新增 cover_url / custom_title（自定义封面和标题）
+"""
+
+import sqlalchemy as sa
+
+from alembic import context, op
+
+revision = "054_add_confirm_generation_fields"
+down_revision = "053_generation_task_is_preview"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+
+    if context.get_context().dialect.name == "postgresql":
+        # 幂等检查：source_task_id 列是否已存在
+        result = conn.execute(
+            sa.text(
+                "SELECT column_name FROM information_schema.columns "
+                "WHERE table_name = 'generation_tasks' AND column_name = 'source_task_id'"
+            )
+        )
+        if result.scalar() is not None:
+            return
+
+    # source_task_id
+    op.add_column(
+        "generation_tasks",
+        sa.Column("source_task_id", sa.String(32), nullable=False, server_default=""),
+    )
+
+    # output_width
+    op.add_column(
+        "generation_tasks",
+        sa.Column("output_width", sa.Integer, nullable=False, server_default=sa.text("1280")),
+    )
+
+    # output_height
+    op.add_column(
+        "generation_tasks",
+        sa.Column("output_height", sa.Integer, nullable=False, server_default=sa.text("720")),
+    )
+
+    # cover_url
+    op.add_column(
+        "generation_tasks",
+        sa.Column("cover_url", sa.String(1000), nullable=False, server_default=""),
+    )
+
+    # custom_title
+    op.add_column(
+        "generation_tasks",
+        sa.Column("custom_title", sa.String(500), nullable=False, server_default=""),
+    )
+
+    # 索引
+    op.create_index(
+        "ix_generation_tasks_source_task_id",
+        "generation_tasks",
+        ["source_task_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_generation_tasks_source_task_id", table_name="generation_tasks")
+    op.drop_column("generation_tasks", "custom_title")
+    op.drop_column("generation_tasks", "cover_url")
+    op.drop_column("generation_tasks", "output_height")
+    op.drop_column("generation_tasks", "output_width")
+    op.drop_column("generation_tasks", "source_task_id")

apps/api/app/api/routes/generation_tasks.py

@@ -26,6 +26,7 @@ from app.schemas.generated_video import (
 )
 from app.schemas.generation_task import (
     BatchGenerationTaskResponse,
+    ConfirmGenerationRequest,
     CreateGenerationTaskRequest,
     GenerationTaskResponse,
     ListGenerationTasksResponse,
@@ -62,6 +63,12 @@ def _to_generation_task_response(task) -> GenerationTaskResponse:
         video_title=getattr(task, "video_title", ""),
         resolution=getattr(task, "resolution", ""),
         bgm_config=getattr(task, "bgm_config", {}) or {},
+        is_preview=getattr(task, "is_preview", False),
+        source_task_id=getattr(task, "source_task_id", ""),
+        output_width=getattr(task, "output_width", 1280),
+        output_height=getattr(task, "output_height", 720),
+        cover_url=getattr(task, "cover_url", ""),
+        custom_title=getattr(task, "custom_title", ""),
         logs=getattr(task, "logs", "[]"),
         status=task.status,
         progress=task.progress,
@@ -291,6 +298,12 @@ def create_generation_task(
                     bgm_config=request.bgm_config,
                     auto_retry_enabled=request.auto_retry_enabled,
                     auto_retry_max=request.auto_retry_max,
+                    is_preview=request.is_preview,
+                    source_task_id=request.source_task_id,
+                    output_width=request.output_width,
+                    output_height=request.output_height,
+                    cover_url=request.cover_url,
+                    custom_title=request.custom_title,
                 )
             )
             try:
@@ -331,6 +344,83 @@ def create_generation_task(
     return BatchGenerationTaskResponse(items=items, total=len(items))
 
 
+@router.post("/tasks/{task_id}/confirm", response_model=BatchGenerationTaskResponse)
+def confirm_generation(
+    task_id: str,
+    request: ConfirmGenerationRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    generation_task_repository: Any = Depends(get_generation_task_repository),
+    project_repository: Any = Depends(get_project_repository),
+) -> BatchGenerationTaskResponse:
+    """确认生成 — 基于预览任务创建正式生成任务。
+
+    查找预览任务，复制其配置，创建新的正式生成任务（is_preview=False），
+    使用高分辨率，复用 worker.generate_video 渲染路径。
+    """
+    # 1. 查找源预览任务
+    source_task = generation_task_repository.get(task_id)
+    if source_task is None:
+        raise HTTPException(status_code=404, detail=f"Preview task {task_id} not found")
+
+    # 2. 权限检查
+    if source_task.created_by_user_id and source_task.created_by_user_id != authenticated_user.user.id:
+        raise HTTPException(status_code=403, detail="Access denied to this task")
+    if source_task.project_id:
+        check_project_access(source_task.project_id, authenticated_user.user.id, project_repository)
+
+    # 3. 创建正式生成任务，复制预览任务的配置
+    use_case = CreateGenerationTaskUseCase(generation_task_repository)
+    new_task = use_case.execute(
+        CreateGenerationTaskCommand(
+            project_id=source_task.project_id,
+            asset_library_id=source_task.asset_library_id,
+            strategy_id=source_task.strategy_id,
+            voice_library_id=source_task.voice_library_id,
+            template_id=source_task.template_id,
+            asset_ids=source_task.asset_ids,
+            title_ids=source_task.title_ids,
+            voice_ids=source_task.voice_ids,
+            created_by_user_id=authenticated_user.user.id,
+            source_edit_plan_id=source_task.source_edit_plan_id or "",
+            asset_select_mode=source_task.asset_select_mode,
+            video_title=getattr(source_task, "video_title", ""),
+            resolution=getattr(source_task, "resolution", ""),
+            is_preview=False,
+            source_task_id=task_id,
+            output_width=request.output_width,
+            output_height=request.output_height,
+            cover_url=request.cover_url,
+            custom_title=request.custom_title,
+        )
+    )
+
+    # 4. 调度 worker.generate_video（同一条渲染路径）
+    try:
+        if not safe_enqueue_generation_task(
+            new_task,
+            generation_task_repository,
+            user_id=authenticated_user.user.id,
+            log_prefix="[确认生成]",
+            log_task_status=True,
+        ):
+            logger.warning("[确认生成] 入队失败: task_id=%s", new_task.id)
+    except UserPendingLimitExceeded:
+        raise HTTPException(
+            status_code=429,
+            detail="您的待处理任务过多，请等待完成后再提交",
+        ) from None
+    except GlobalQueueFull:
+        raise HTTPException(
+            status_code=503,
+            detail="系统繁忙，请稍后再试",
+        ) from None
+
+    return BatchGenerationTaskResponse(
+        items=[_to_generation_task_response(new_task)],
+        total=1,
+    )
+
+
 @router.get("/tasks", response_model=ListGenerationTasksResponse)
 def list_generation_tasks(
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
@@ -428,6 +518,12 @@ def retry_generation_task(
             asset_select_mode=getattr(task, "asset_select_mode", ""),
             video_title=getattr(task, "video_title", ""),
             resolution=getattr(task, "resolution", ""),
+            is_preview=getattr(task, "is_preview", False),
+            source_task_id=getattr(task, "source_task_id", ""),
+            output_width=getattr(task, "output_width", 1280),
+            output_height=getattr(task, "output_height", 720),
+            cover_url=getattr(task, "cover_url", ""),
+            custom_title=getattr(task, "custom_title", ""),
         )
     )
     try:

apps/api/app/schemas/generation_task.py

@@ -4,6 +4,15 @@ from datetime import datetime
 from pydantic import BaseModel, Field, field_validator, model_validator
 
 
+class ConfirmGenerationRequest(BaseModel):
+    """确认生成请求体 — 基于预览任务创建正式生成任务"""
+
+    output_width: int = Field(default=1080, description="输出视频宽度")
+    output_height: int = Field(default=1920, description="输出视频高度")
+    cover_url: str = Field(default="", description="自定义封面图片 URL")
+    custom_title: str = Field(default="", description="自定义视频标题")
+
+
 class CreateGenerationTaskRequest(BaseModel):
     """创建生成任务请求。
 
@@ -57,6 +66,13 @@ class CreateGenerationTaskRequest(BaseModel):
         default_factory=dict,
         description="自定义BGM配置，覆盖模板BGM设置。支持 enabled/source/asset_id/preset_id/audio_url/volume 等字段",
     )
+    # ── 预览 / 确认生成 ──
+    is_preview: bool = Field(default=False, description="是否为预览任务")
+    source_task_id: str = Field(default="", description="来源预览任务 ID（确认生成时传入）")
+    output_width: int = Field(default=1280, description="输出视频宽度")
+    output_height: int = Field(default=720, description="输出视频高度")
+    cover_url: str = Field(default="", description="封面图片 URL")
+    custom_title: str = Field(default="", description="自定义视频标题")
 
     @model_validator(mode="after")
     def _check_at_least_one_mode(self) -> "CreateGenerationTaskRequest":
@@ -87,6 +103,12 @@ class GenerationTaskResponse(BaseModel):
     video_title: str = ""
     resolution: str = ""
     bgm_config: dict = Field(default_factory=dict)
+    is_preview: bool = False
+    source_task_id: str = ""
+    output_width: int = 1280
+    output_height: int = 720
+    cover_url: str = ""
+    custom_title: str = ""
     status: str
     progress: float
     result_count: int

apps/worker/worker_app/tasks/generation.py

@@ -1040,6 +1040,11 @@ def _load_task_info(task_id: str) -> dict | None:
             "resolution": getattr(gen_task, "resolution", "") or "",
             "bgm_config": dict(getattr(gen_task, "bgm_config", {}) or {}),
             "is_preview": bool(getattr(gen_task, "is_preview", False)),
+            "source_task_id": getattr(gen_task, "source_task_id", "") or "",
+            "output_width": getattr(gen_task, "output_width", OUTPUT_WIDTH) or OUTPUT_WIDTH,
+            "output_height": getattr(gen_task, "output_height", OUTPUT_HEIGHT) or OUTPUT_HEIGHT,
+            "cover_url": getattr(gen_task, "cover_url", "") or "",
+            "custom_title": getattr(gen_task, "custom_title", "") or "",
             "voice_ids": list(getattr(gen_task, "voice_ids", []) or []),
         }
     finally:
@@ -1431,6 +1436,14 @@ def generate_video(self, task_id: str) -> dict:
 
             # ── 3. 渲染 + 混音 ───────────────────────────────────────────────
             _update_task_progress(task_id, 40, "开始渲染")
+            # 动态分辨率：优先使用 output_width/output_height，其次 resolution 字符串
+            _ow = task_info.get("output_width", OUTPUT_WIDTH) or OUTPUT_WIDTH
+            _oh = task_info.get("output_height", OUTPUT_HEIGHT) or OUTPUT_HEIGHT
+            if _ow != OUTPUT_WIDTH or _oh != OUTPUT_HEIGHT:
+                _resolved_resolution = f"{_ow}x{_oh}"
+            else:
+                _resolved_resolution = task_info.get("resolution", "")
+
             output_path, render_duration = _render_video(
                 task_id=task_id,
                 downloaded_videos=downloaded_videos,
@@ -1441,7 +1454,7 @@ def generate_video(self, task_id: str) -> dict:
                 user_id=user_id,
                 temp_path=temp_path,
                 output_name=output_name,
-                resolution=task_info.get("resolution", ""),
+                resolution=_resolved_resolution,
                 bgm_config=task_info.get("bgm_config", {}),
                 is_preview=task_info.get("is_preview", False),
                 voice_ids=task_info.get("voice_ids", []),

docs/schema-metadata-snapshot.json

@@ -1659,6 +1659,54 @@
           "primary_key": false,
           "type": "DATETIME",
           "unique": false
+        },
+        {
+          "index": false,
+          "name": "is_preview",
+          "nullable": false,
+          "primary_key": false,
+          "type": "BOOLEAN",
+          "unique": false
+        },
+        {
+          "index": true,
+          "name": "source_task_id",
+          "nullable": false,
+          "primary_key": false,
+          "type": "VARCHAR(32)",
+          "unique": false
+        },
+        {
+          "index": false,
+          "name": "output_width",
+          "nullable": false,
+          "primary_key": false,
+          "type": "INTEGER",
+          "unique": false
+        },
+        {
+          "index": false,
+          "name": "output_height",
+          "nullable": false,
+          "primary_key": false,
+          "type": "INTEGER",
+          "unique": false
+        },
+        {
+          "index": false,
+          "name": "cover_url",
+          "nullable": false,
+          "primary_key": false,
+          "type": "VARCHAR(1000)",
+          "unique": false
+        },
+        {
+          "index": false,
+          "name": "custom_title",
+          "nullable": false,
+          "primary_key": false,
+          "type": "VARCHAR(500)",
+          "unique": false
         }
       ],
       "indexes": [
@@ -1717,6 +1765,13 @@
           ],
           "name": "ix_generation_tasks_template_id",
           "unique": false
+        },
+        {
+          "columns": [
+            "source_task_id"
+          ],
+          "name": "ix_generation_tasks_source_task_id",
+          "unique": false
         }
       ],
       "primary_key": [

packages/adapters/sqlalchemy_impl/generation_task_repository.py

@@ -37,6 +37,11 @@ def _to_domain(model: GenerationTaskModel) -> GenerationTask:
         resolution=getattr(model, "resolution", "") or "",
         bgm_config=dict(getattr(model, "bgm_config", {}) or {}),
         is_preview=bool(getattr(model, "is_preview", False)),
+        source_task_id=getattr(model, "source_task_id", "") or "",
+        output_width=getattr(model, "output_width", 1280) or 1280,
+        output_height=getattr(model, "output_height", 720) or 720,
+        cover_url=getattr(model, "cover_url", "") or "",
+        custom_title=getattr(model, "custom_title", "") or "",
         logs=model.logs or "[]",
         created_at=model.created_at,
         updated_at=model.updated_at,
@@ -76,6 +81,11 @@ class SQLAlchemyGenerationTaskRepository:
             resolution=task.resolution or "",
             bgm_config=task.bgm_config or {},
             is_preview=task.is_preview or False,
+            source_task_id=task.source_task_id or "",
+            output_width=task.output_width,
+            output_height=task.output_height,
+            cover_url=task.cover_url or "",
+            custom_title=task.custom_title or "",
             logs=task.logs,
             created_at=task.created_at,
             updated_at=task.updated_at,
@@ -242,6 +252,11 @@ class SQLAlchemyGenerationTaskRepository:
             model.bgm_config = task.bgm_config or {}
         if hasattr(model, "is_preview"):
             model.is_preview = task.is_preview or False
+        model.source_task_id = task.source_task_id or ""
+        model.output_width = task.output_width
+        model.output_height = task.output_height
+        model.cover_url = task.cover_url or ""
+        model.custom_title = task.custom_title or ""
         model.logs = task.logs
         self.session.commit()
         return task

packages/adapters/sqlalchemy_impl/models.py

@@ -293,6 +293,11 @@ class GenerationTaskModel(Base):
     video_title = Column(String(255), nullable=False, default="")
     resolution = Column(String(20), nullable=False, default="")
     is_preview = Column(Boolean, nullable=False, default=False, index=True)
+    source_task_id = Column(String(32), nullable=False, default="", index=True)
+    output_width = Column(Integer, nullable=False, default=1280)
+    output_height = Column(Integer, nullable=False, default=720)
+    cover_url = Column(String(1000), nullable=False, default="")
+    custom_title = Column(String(500), nullable=False, default="")
     bgm_config = Column(JSON, nullable=False, default=dict)
     extra_meta = Column("metadata", JSON, nullable=False, default=dict)
     logs = Column(Text, nullable=False, default="[]", server_default="[]")

packages/application/generation_tasks.py

@@ -27,6 +27,11 @@ class CreateGenerationTaskCommand:
     auto_retry_enabled: bool = False
     auto_retry_max: int = 0
     is_preview: bool = False
+    source_task_id: str = ""
+    output_width: int = 1280
+    output_height: int = 720
+    cover_url: str = ""
+    custom_title: str = ""
 
 
 class CreateGenerationTaskUseCase:
@@ -58,6 +63,11 @@ class CreateGenerationTaskUseCase:
             auto_retry_enabled=command.auto_retry_enabled,
             auto_retry_max=command.auto_retry_max,
             is_preview=command.is_preview,
+            source_task_id=command.source_task_id,
+            output_width=command.output_width,
+            output_height=command.output_height,
+            cover_url=command.cover_url,
+            custom_title=command.custom_title,
         )
         return self.generation_task_repository.create(task)
 

packages/domain/generation_task.py

@@ -116,6 +116,11 @@ class GenerationTask:
     resolution: str = ""
     bgm_config: dict = field(default_factory=dict)
     is_preview: bool = False
+    source_task_id: str = ""
+    output_width: int = 1280
+    output_height: int = 720
+    cover_url: str = ""
+    custom_title: str = ""
     logs: str = "[]"
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
     updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@@ -142,6 +147,11 @@ class GenerationTask:
         auto_retry_enabled: bool = False,
         auto_retry_max: int = 0,
         is_preview: bool = False,
+        source_task_id: str = "",
+        output_width: int = 1280,
+        output_height: int = 720,
+        cover_url: str = "",
+        custom_title: str = "",
     ) -> "GenerationTask":
         if not project_id.strip() and not template_id.strip():
             raise ValueError("project_id 或 template_id 至少需要提供一个")
@@ -167,6 +177,11 @@ class GenerationTask:
             auto_retry_enabled=auto_retry_enabled,
             auto_retry_max=auto_retry_max,
             is_preview=is_preview,
+            source_task_id=source_task_id,
+            output_width=output_width,
+            output_height=output_height,
+            cover_url=cover_url,
+            custom_title=custom_title,
         )
 
     # ── 状态查询 ────────────────────────────────────────────────────────────

tests/unit/test_confirm_generation.py

@@ -0,0 +1,384 @@
+"""确认生成 API 单元测试.
+
+覆盖 POST /tasks/{task_id}/confirm 端点：
+  - 正常确认流程
+  - 预览任务不存在 → 404
+  - 权限不足 → 403
+  - is_preview=False 及分辨率正确
+  - cover_url 和 custom_title 正确传递
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any, Optional
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
+
+from packages.domain.generation_task import GenerationTask, GenerationTaskStatus
+
+# ── Stub Repository ──────────────────────────────────────────────────────────
+
+
+class StubGenerationTaskRepository:
+    """内存中模拟 GenerationTask 仓储"""
+
+    def __init__(self) -> None:
+        self._store: dict[str, Any] = {}
+
+    def create(self, task: Any) -> Any:
+        self._store[task.id] = task
+        return task
+
+    def get(self, task_id: str) -> Optional[Any]:
+        return self._store.get(task_id)
+
+    def update(self, task: Any) -> Any:
+        if task.id not in self._store:
+            raise ValueError(f"GenerationTask {task.id} not found")
+        self._store[task.id] = task
+        return task
+
+    def list_by_project(self, project_id: str) -> list[Any]:
+        return [t for t in self._store.values() if t.project_id == project_id]
+
+    def list_by_user(self, user_id: str) -> list[Any]:
+        return [t for t in self._store.values() if t.created_by_user_id == user_id]
+
+    def count_by_user(self, user_id: str) -> int:
+        return len([t for t in self._store.values() if t.created_by_user_id == user_id])
+
+    def count_pending_by_user(self, user_id: str) -> int:
+        return len([
+            t for t in self._store.values()
+            if t.created_by_user_id == user_id and t.status == GenerationTaskStatus.PENDING
+        ])
+
+    def count_pending_total(self) -> int:
+        return len([t for t in self._store.values() if t.status == GenerationTaskStatus.PENDING])
+
+    def list_recent_by_user(self, user_id: str, limit: int = 5) -> list[Any]:
+        items = [t for t in self._store.values() if t.created_by_user_id == user_id]
+        items.sort(key=lambda t: t.created_at, reverse=True)
+        return items[:limit]
+
+    def list_by_source_edit_plan(self, plan_id: str) -> list[Any]:
+        return [t for t in self._store.values() if (t.source_edit_plan_id or "") == plan_id]
+
+
+# ── Stub Project Repository ──────────────────────────────────────────────────
+
+
+@dataclass
+class FakeProject:
+    id: str = "project-001"
+    owner_user_id: str = "user-001"
+    shared_users: list[str] = field(default_factory=list)
+    name: str = "Test Project"
+
+    def can_access(self, user_id: str) -> bool:
+        return user_id == self.owner_user_id or user_id in self.shared_users
+
+
+class StubProjectRepository:
+    def __init__(self) -> None:
+        self._projects: dict[str, FakeProject] = {}
+
+    def add(self, project: FakeProject) -> None:
+        self._projects[project.id] = project
+
+    def find_by_id(self, project_id: str) -> Optional[FakeProject]:
+        return self._projects.get(project_id)
+
+
+# ── Fixtures ─────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class FakeUser:
+    id: str = "user-001"
+    email: str = "test@example.com"
+
+
+@dataclass
+class FakeAuthenticatedUser:
+    user: FakeUser = field(default_factory=FakeUser)
+    session_id: str | None = None
+    token_type: str | None = None
+
+
+@pytest.fixture
+def gen_task_repo() -> StubGenerationTaskRepository:
+    return StubGenerationTaskRepository()
+
+
+@pytest.fixture
+def project_repo() -> StubProjectRepository:
+    repo = StubProjectRepository()
+    repo.add(FakeProject())
+    return repo
+
+
+@pytest.fixture
+def app(
+    gen_task_repo: StubGenerationTaskRepository,
+    project_repo: StubProjectRepository,
+) -> FastAPI:
+    """构建测试 FastAPI 应用，注入 Stub Repository"""
+    from app.api.routes.generation_tasks import router
+    from app.auth import get_current_user
+    from app.dependencies import (
+        get_asset_library_repository,
+        get_asset_repository,
+        get_generated_video_repository,
+        get_generation_task_repository,
+        get_project_repository,
+    )
+
+    test_app = FastAPI()
+    test_app.include_router(router, prefix="/api/v1")
+
+    def override_get_current_user():
+        return FakeAuthenticatedUser()
+
+    def override_get_generation_task_repository():
+        return gen_task_repo
+
+    def override_get_project_repository():
+        return project_repo
+
+    test_app.dependency_overrides[get_current_user] = override_get_current_user
+    test_app.dependency_overrides[get_generation_task_repository] = override_get_generation_task_repository
+    test_app.dependency_overrides[get_project_repository] = override_get_project_repository
+    # Stubs for repositories not used by confirm endpoint but required by router
+    test_app.dependency_overrides[get_asset_library_repository] = lambda: MagicMock()
+    test_app.dependency_overrides[get_asset_repository] = lambda: MagicMock()
+    test_app.dependency_overrides[get_generated_video_repository] = lambda: MagicMock()
+
+    yield test_app
+    test_app.dependency_overrides.clear()
+
+
+@pytest.fixture
+def client(app: FastAPI) -> TestClient:
+    return TestClient(app)
+
+
+def _make_preview_task(**kwargs: Any) -> GenerationTask:
+    """创建预览任务"""
+    defaults = dict(
+        id="preview-task-001",
+        project_id="project-001",
+        asset_library_id="library-001",
+        strategy_id="one_take",
+        voice_library_id="",
+        template_id="",
+        asset_ids=["asset-1"],
+        title_ids=[],
+        voice_ids=[],
+        status=GenerationTaskStatus.COMPLETED,
+        progress=100.0,
+        result_count=1,
+        error_message="",
+        created_by_user_id="user-001",
+        source_edit_plan_id="",
+        asset_select_mode="all",
+        is_preview=True,
+        source_task_id="",
+        output_width=1280,
+        output_height=720,
+        cover_url="",
+        custom_title="",
+        video_title="",
+        resolution="",
+        bgm_config={},
+    )
+    defaults.update(kwargs)
+    return GenerationTask(**defaults)
+
+
+# ── Tests ────────────────────────────────────────────────────────────────────
+
+
+class TestConfirmGeneration:
+    def test_confirm_success(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """正常确认流程：预览任务存在、权限正确 → 创建正式任务"""
+        preview = _make_preview_task()
+        gen_task_repo.create(preview)
+
+        with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True) as mock_enqueue:
+            resp = client.post(
+                f"/api/v1/tasks/{preview.id}/confirm",
+                json={
+                    "output_width": 1080,
+                    "output_height": 1920,
+                    "cover_url": "https://example.com/cover.jpg",
+                    "custom_title": "我的视频",
+                },
+            )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["total"] == 1
+        item = data["items"][0]
+        assert item["is_preview"] is False
+        assert item["source_task_id"] == preview.id
+        assert item["output_width"] == 1080
+        assert item["output_height"] == 1920
+        assert item["cover_url"] == "https://example.com/cover.jpg"
+        assert item["custom_title"] == "我的视频"
+        # 复制了预览任务的配置
+        assert item["project_id"] == "project-001"
+        assert item["asset_library_id"] == "library-001"
+        assert item["strategy_id"] == "one_take"
+        assert item["asset_ids"] == ["asset-1"]
+
+        # 验证入队函数被调用
+        mock_enqueue.assert_called_once()
+
+    def test_confirm_not_found(self, client: TestClient) -> None:
+        """预览任务不存在 → 404"""
+        resp = client.post(
+            "/api/v1/tasks/nonexistent-task/confirm",
+            json={"output_width": 1080, "output_height": 1920},
+        )
+        assert resp.status_code == 404
+        assert "not found" in resp.json()["detail"]
+
+    def test_confirm_access_denied(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """权限不足 → 403"""
+        preview = _make_preview_task(created_by_user_id="other-user-999")
+        gen_task_repo.create(preview)
+
+        resp = client.post(
+            f"/api/v1/tasks/{preview.id}/confirm",
+            json={"output_width": 1080, "output_height": 1920},
+        )
+        assert resp.status_code == 403
+        assert "denied" in resp.json()["detail"].lower() or "Access" in resp.json()["detail"]
+
+    def test_confirm_preserves_config(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """确认后的任务 is_preview=False，分辨率已更新，其余配置从预览任务复制"""
+        preview = _make_preview_task(
+            voice_library_id="voice-001",
+            template_id="tmpl-001",
+            title_ids=["title-1", "title-2"],
+            voice_ids=["voice-a"],
+        )
+        gen_task_repo.create(preview)
+
+        with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True):
+            resp = client.post(
+                f"/api/v1/tasks/{preview.id}/confirm",
+                json={"output_width": 1920, "output_height": 1080},
+            )
+
+        assert resp.status_code == 200
+        item = resp.json()["items"][0]
+        assert item["is_preview"] is False
+        assert item["source_task_id"] == preview.id
+        assert item["output_width"] == 1920
+        assert item["output_height"] == 1080
+        # 默认封面和标题
+        assert item["cover_url"] == ""
+        assert item["custom_title"] == ""
+        # 复制的配置
+        assert item["voice_library_id"] == "voice-001"
+        assert item["template_id"] == "tmpl-001"
+        assert item["title_ids"] == ["title-1", "title-2"]
+        assert item["voice_ids"] == ["voice-a"]
+
+    def test_confirm_cover_and_title(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """cover_url 和 custom_title 正确传递"""
+        preview = _make_preview_task()
+        gen_task_repo.create(preview)
+
+        with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True):
+            resp = client.post(
+                f"/api/v1/tasks/{preview.id}/confirm",
+                json={
+                    "output_width": 1080,
+                    "output_height": 1920,
+                    "cover_url": "https://cdn.example.com/my-cover.png",
+                    "custom_title": "测试视频标题",
+                },
+            )
+
+        assert resp.status_code == 200
+        item = resp.json()["items"][0]
+        assert item["cover_url"] == "https://cdn.example.com/my-cover.png"
+        assert item["custom_title"] == "测试视频标题"
+
+    def test_confirm_default_resolution(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """不传分辨率时使用 ConfirmGenerationRequest 默认值 1080x1920"""
+        preview = _make_preview_task()
+        gen_task_repo.create(preview)
+
+        with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True):
+            resp = client.post(
+                f"/api/v1/tasks/{preview.id}/confirm",
+                json={},
+            )
+
+        assert resp.status_code == 200
+        item = resp.json()["items"][0]
+        assert item["output_width"] == 1080
+        assert item["output_height"] == 1920
+
+    def test_confirm_creates_new_task_in_repo(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """确认生成的任务确实被存入 repository"""
+        preview = _make_preview_task()
+        gen_task_repo.create(preview)
+
+        initial_count = len(gen_task_repo._store)
+
+        with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True):
+            resp = client.post(
+                f"/api/v1/tasks/{preview.id}/confirm",
+                json={},
+            )
+
+        assert resp.status_code == 200
+        new_task_id = resp.json()["items"][0]["id"]
+        assert new_task_id != preview.id
+        assert len(gen_task_repo._store) == initial_count + 1
+
+        new_task = gen_task_repo.get(new_task_id)
+        assert new_task is not None
+        assert new_task.is_preview is False
+        assert new_task.source_task_id == preview.id