fix: AI Code Review 阻塞级问题修复 — 分辨率一致性校验+空指针防护

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

@@ -369,23 +369,40 @@ def confirm_generation(
     if source_task.project_id:
         check_project_access(source_task.project_id, authenticated_user.user.id, project_repository)
 
-    # 3. 如果预览任务已完成，直接复用产物（秒出）
+    # 3. 如果预览任务已完成，检查分辨率一致性后复用产物（秒出）
     if source_task.is_completed and getattr(source_task, "is_preview", False):
-        source_task.mark_confirmed(
-            cover_url=request.cover_url,
-            custom_title=request.custom_title,
-            output_width=request.output_width,
-            output_height=request.output_height,
-        )
-        generation_task_repository.update(source_task)
+        # 校验请求的分辨率是否与预览实际渲染的分辨率一致
+        req_w = request.output_width or 0
+        req_h = request.output_height or 0
+        src_w = getattr(source_task, "output_width", 0) or 0
+        src_h = getattr(source_task, "output_height", 0) or 0
+        resolution_match = (req_w == 0 or req_w == src_w) and (req_h == 0 or req_h == src_h)
+
+        if resolution_match:
+            source_task.mark_confirmed(
+                cover_url=request.cover_url,
+                custom_title=request.custom_title,
+                output_width=request.output_width,
+                output_height=request.output_height,
+            )
+            generation_task_repository.update(source_task)
+            logger.info(
+                "[确认生成] 复用预览产物: task_id=%s, user_id=%s",
+                task_id,
+                authenticated_user.user.id,
+            )
+            return BatchGenerationTaskResponse(
+                items=[_to_generation_task_response(source_task)],
+                total=1,
+            )
+        # 分辨率不一致，跳过复用，走新建任务流程
         logger.info(
-            "[确认生成] 复用预览产物: task_id=%s, user_id=%s",
+            "[确认生成] 分辨率不一致，跳过复用: task_id=%s, src=%sx%s, req=%sx%s",
             task_id,
-            authenticated_user.user.id,
-        )
-        return BatchGenerationTaskResponse(
-            items=[_to_generation_task_response(source_task)],
-            total=1,
+            src_w,
+            src_h,
+            req_w,
+            req_h,
         )
 
     # 4. 预览任务未完成，创建新的正式任务走渲染流程

apps/api/app/api/routes/templates_editor/generation.py

@@ -206,6 +206,9 @@ def _find_reusable_preview_task(gen_task_repo, plan_id: str, plan) -> "object |
         completed_at = getattr(task, "completed_at", None)
         if completed_at and hasattr(plan, "updated_at"):
             plan_updated = plan.updated_at
+            # 如果 plan.updated_at 为空，无法判断是否修改过，跳过
+            if plan_updated is None:
+                continue
             # 如果 plan 在预览完成后又被修改了，不能复用
             if plan_updated > completed_at:
                 continue

tests/unit/test_confirm_generation.py

@@ -198,8 +198,8 @@ def _make_preview_task(**kwargs: Any) -> GenerationTask:
         asset_select_mode="all",
         is_preview=True,
         source_task_id="",
-        output_width=1280,
-        output_height=720,
+        output_width=1080,
+        output_height=1920,
         cover_url="",
         custom_title="",
         video_title="",
@@ -405,3 +405,28 @@ class TestConfirmGenerationErrors:
         item = resp.json()["items"][0]
         assert item["output_width"] == 1080
         assert item["output_height"] == 1920
+
+    def test_confirm_skips_reuse_when_resolution_mismatch(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """请求的分辨率与预览渲染的分辨率不一致时，跳过复用，创建新任务"""
+        preview = _make_preview_task(output_width=1080, output_height=1920)
+        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={"output_width": 720, "output_height": 1280},
+            )
+
+        assert resp.status_code == 200
+        item = resp.json()["items"][0]
+        # 创建了新任务（而非复用）
+        assert item["id"] != preview.id
+        assert item["is_preview"] is False
+        assert item["source_task_id"] == preview.id
+        assert len(gen_task_repo._store) == initial_count + 1

tests/unit/test_unified_render_reuse.py

@@ -268,3 +268,20 @@ class TestMarkConfirmed:
         task.mark_confirmed(output_width=0, output_height=0)
         assert task.output_width == 1920
         assert task.output_height == 1080
+
+    def test_skips_when_plan_updated_at_is_none(self):
+        from app.api.routes.templates_editor.generation import _find_reusable_preview_task
+
+        repo = StubGenTaskRepo()
+        now = datetime.now(timezone.utc)
+        task = _make_task(
+            source_edit_plan_id="plan-1",
+            is_preview=True,
+            status=GenerationTaskStatus.COMPLETED,
+            completed_at=now - timedelta(minutes=5),
+        )
+        repo.create(task)
+
+        plan = _make_plan(updated_at=None)
+        result = _find_reusable_preview_task(repo, "plan-1", plan)
+        assert result is None