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

@@ -377,6 +377,36 @@ def create_generation_task(
                             title_config=request.title_config,
                             db=db,
                         )
+
+                    # 兜底关联编辑计划：前端未传 source_edit_plan_id 时，
+                    # 通过 template_id + user_id 在 DB 层直接查找最新的 plan
+                    if not task.source_edit_plan_id and request.template_id:
+                        try:
+                            from packages.adapters.sqlalchemy_impl.models import EditPlanModel
+
+                            _plan_model = (
+                                db.query(EditPlanModel)
+                                .filter(
+                                    EditPlanModel.template_id == request.template_id,
+                                    EditPlanModel.created_by_user_id == user_id,
+                                )
+                                .order_by(EditPlanModel.created_at.desc())
+                                .first()
+                            )
+                            if _plan_model:
+                                task.source_edit_plan_id = _plan_model.id
+                                generation_task_repository.update(task)
+                                logger.info(
+                                    "[生成任务] 自动关联编辑计划: task_id=%s plan_id=%s",
+                                    task.id,
+                                    _plan_model.id,
+                                )
+                        except Exception:
+                            logger.warning(
+                                "[生成任务] 查找关联编辑计划失败（不影响主流程）: task_id=%s",
+                                task.id,
+                                exc_info=True,
+                            )
                 else:
                     failed_tasks.append(task)
             except UserPendingLimitExceeded as _e:

apps/api/app/services/edit_plan_service.py

@@ -423,6 +423,9 @@ class EditPlanService:
                 )
                 db.add(model)
 
+            # flush 让新建 clip 写入当前事务（未 commit），后续查询才能找到它们
+            db.flush()
+
             # 3. 标记有 asset_id 的 clips 为 ready（不 commit）
             pending_with_asset = (
                 db.query(EditPlanClipModel)

apps/worker/worker_app/tasks/generation.py

@@ -1919,12 +1919,11 @@ def generate_video(self, task_id: str) -> dict:
                 _repo = SQLAlchemyGenerationTaskRepository(_session)
                 gen_task = _repo.get(task_id)
                 if gen_task:
-                    gen_task.append_log(  # type: ignore[misc]
-                        "任务失败",
+                    gen_task.append_log(
+                        "render",
                         str(error),
                         level="ERROR",
                         error_type=type(error).__name__,
-                        stage="render",
                     )
                     _flush_logs(task_id, gen_task)
             finally:

tests/unit/test_three_bugs_fix.py

@@ -0,0 +1,404 @@
+"""Regression tests for 3 bug fixes: flush, append_log, plan_id fallback."""
+
+from __future__ import annotations
+
+import json
+import os
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+
+
+# ── Bug 1: db.flush() before pending query ─────────────────────────────────
+
+
+class TestReplaceAllClipsFlush:
+    """replace_all_clips_transactional must flush before querying pending clips."""
+
+    def _make_svc_and_db(self, pending_results):
+        """Helper: create service + db mock. pending_results = list returned by pending query."""
+        from app.services.edit_plan_service import EditPlanService
+
+        db = MagicMock()
+        # Delete query
+        delete_query = MagicMock()
+        delete_query.filter.return_value.delete.return_value = 0
+        # Pending query: single .filter() with multiple conditions
+        ready_query = MagicMock()
+        ready_query.filter.return_value.all.return_value = pending_results
+        db.query.side_effect = [delete_query, ready_query]
+        return db, EditPlanService
+
+    def _setup_clip_mocks(self, mock_clip_cls, mock_model_cls, asset_id="asset-1"):
+        mock_entity = MagicMock()
+        mock_entity.id = "clip-1"
+        mock_entity.plan_id = "plan-1"
+        mock_entity.clip_type = "main"
+        mock_entity.order = 0
+        mock_entity.asset_id = asset_id
+        mock_entity.text_content = ""
+        mock_entity.start_time = 0.0
+        mock_entity.duration = 3.0
+        mock_entity.transition_effect = "cut"
+        mock_entity.transition_duration = 0.0
+        mock_entity.playback_speed = 1.0
+        mock_entity.status.value = "pending"
+        mock_entity.config = {}
+        mock_clip_cls.create.return_value = mock_entity
+        mock_model_cls.return_value = MagicMock()
+        return mock_entity
+
+    @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel")
+    @patch("app.services.edit_plan_service.EditPlanClip")
+    def test_flush_called_between_add_and_query(self, mock_clip_cls, mock_model_cls):
+        """db.flush() must be called after db.add() and before the pending query."""
+        self._setup_clip_mocks(mock_clip_cls, mock_model_cls)
+        db, SvcClass = self._make_svc_and_db([])
+
+        clip_repo = MagicMock()
+        clip_repo.session = db
+        svc = SvcClass.__new__(SvcClass)
+        svc._clip_repo = clip_repo
+
+        svc.replace_all_clips_transactional(
+            "plan-1",
+            [{"asset_id": "asset-1", "start_time": 0.0, "duration": 3.0, "order": 0}],
+        )
+
+        db.flush.assert_called_once()
+        # Verify ordering: add → flush → query → commit
+        method_names = [c[0] for c in db.method_calls]
+        add_idx = method_names.index("add")
+        flush_idx = method_names.index("flush")
+        commit_idx = method_names.index("commit")
+        assert add_idx < flush_idx < commit_idx
+
+    @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel")
+    @patch("app.services.edit_plan_service.EditPlanClip")
+    def test_flush_marks_new_clips_ready(self, mock_clip_cls, mock_model_cls):
+        """After flush, new clips with asset_id are found and marked ready."""
+        self._setup_clip_mocks(mock_clip_cls, mock_model_cls)
+
+        # Use a plain object so we can verify attribute mutation
+        class FakeClip:
+            status = "pending"
+
+        pending_clip = FakeClip()
+        db, SvcClass = self._make_svc_and_db([pending_clip])
+
+        clip_repo = MagicMock()
+        clip_repo.session = db
+        svc = SvcClass.__new__(SvcClass)
+        svc._clip_repo = clip_repo
+
+        svc.replace_all_clips_transactional(
+            "plan-1",
+            [{"asset_id": "asset-1", "start_time": 0.0, "duration": 3.0, "order": 0}],
+        )
+
+        assert pending_clip.status == "ready"
+
+
+# ── Bug 2: append_log no TypeError ─────────────────────────────────────────
+
+
+class TestAppendLogNoConflict:
+    """append_log must not receive duplicate 'stage' parameter."""
+
+    def _make_task(self):
+        from packages.domain.generation_task import GenerationTask
+
+        return GenerationTask.create(
+            project_id="proj-1",
+            asset_library_id="lib-1",
+            strategy_id="one_take",
+            template_id="tpl-1",
+            asset_ids=["a1"],
+            created_by_user_id="user-1",
+        )
+
+    def test_append_log_with_stage_as_first_positional(self):
+        """append_log(stage, message, ...) works correctly."""
+        task = self._make_task()
+        task.append_log("render", "some error", level="ERROR", error_type="RuntimeError")
+
+        entries = json.loads(task.logs)
+        assert len(entries) == 1
+        assert entries[0]["stage"] == "render"
+        assert entries[0]["message"] == "some error"
+        assert entries[0]["level"] == "ERROR"
+        assert entries[0]["error_type"] == "RuntimeError"
+
+    def test_duplicate_stage_raises_type_error(self):
+        """Sanity check: passing stage both positionally and as kwarg raises TypeError."""
+        task = self._make_task()
+        with pytest.raises(TypeError):
+            task.append_log(
+                "任务失败",  # positional → stage
+                "some error",
+                level="ERROR",
+                stage="render",  # duplicate → TypeError
+            )
+
+
+# ── Bug 3: plan_id fallback in create_generation_task ──────────────────────
+
+
+class TestPlanIdFallback:
+    """Formal generation API should fallback to find plan by template_id + user_id."""
+
+    def test_fallback_code_present_in_source(self):
+        """Verify the fallback logic is present in the generation_tasks module."""
+        import inspect
+
+        from apps.api.app.api.routes import generation_tasks
+
+        source = inspect.getsource(generation_tasks.create_generation_task)
+        assert "EditPlanModel" in source
+        assert "兜底关联编辑计划" in source
+        assert "自动关联编辑计划" in source
+
+    def test_fallback_only_runs_when_source_edit_plan_id_empty(self):
+        """Verify the condition checks for empty source_edit_plan_id."""
+        import inspect
+
+        from apps.api.app.api.routes import generation_tasks
+
+        source = inspect.getsource(generation_tasks.create_generation_task)
+        assert "not task.source_edit_plan_id and request.template_id" in source
+
+    def test_list_by_template_method_exists(self):
+        """Verify SQLAlchemyEditPlanRepository.list_by_template is callable."""
+        from packages.adapters.sqlalchemy_impl.edit_plan_repository import (
+            SQLAlchemyEditPlanRepository,
+        )
+
+        mock_db = MagicMock()
+        mock_session = MagicMock()
+        mock_db.query.return_value = mock_session
+        mock_session.filter.return_value = mock_session
+        mock_session.order_by.return_value = mock_session
+        mock_session.offset.return_value = mock_session
+        mock_session.limit.return_value.all.return_value = []
+
+        repo = SQLAlchemyEditPlanRepository(mock_db)
+        result = repo.list_by_template("tpl-1", limit=20)
+        assert result == []
+
+
+class TestPlanIdFallbackExecution:
+    """Test that the fallback logic actually executes when source_edit_plan_id is empty."""
+
+    @staticmethod
+    def _make_mock_task(source_edit_plan_id=""):
+        t = MagicMock()
+        t.id = "task-1"
+        t.project_id = "proj-1"
+        t.asset_library_id = ""
+        t.strategy_id = "one_take"
+        t.voice_library_id = ""
+        t.template_id = "tpl-1"
+        t.asset_ids = []
+        t.title_ids = []
+        t.voice_ids = []
+        t.source_edit_plan_id = source_edit_plan_id
+        t.asset_select_mode = "manual"
+        t.batch_id = ""
+        t.video_title = ""
+        t.resolution = ""
+        t.bgm_config = None
+        t.is_preview = False
+        t.source_task_id = ""
+        t.output_width = 1280
+        t.output_height = 720
+        t.cover_url = ""
+        t.custom_title = ""
+        t.title_config = {}
+        t.logs = "[]"
+        t.status = "pending"
+        t.progress = 0.0
+        t.error_message = ""
+        t.error_info = None
+        t.created_at = "2026-01-01T00:00:00Z"
+        t.updated_at = "2026-01-01T00:00:00Z"
+        t.started_at = None
+        t.completed_at = None
+        t.created_by_user_id = "user-1"
+        t.auto_retry_enabled = False
+        t.auto_retry_max = 0
+        t.auto_retry_count = 0
+        t.result_count = 0
+        return t
+
+    @staticmethod
+    def _make_request(source_edit_plan_id="", template_id="tpl-1"):
+        req = MagicMock()
+        req.template_id = template_id
+        req.source_edit_plan_id = source_edit_plan_id
+        req.asset_ids = []
+        req.asset_select_mode = "manual"
+        req.asset_select_count = 0
+        req.voice_library_id = ""
+        req.title_ids = []
+        req.voice_ids = []
+        req.strategy_id = "one_take"
+        req.count = 1
+        req.video_title = ""
+        req.resolution = ""
+        req.bgm_config = None
+        req.auto_retry_enabled = False
+        req.auto_retry_max = 0
+        req.is_preview = False
+        req.source_task_id = ""
+        req.output_width = 0
+        req.output_height = 0
+        req.cover_url = ""
+        req.custom_title = ""
+        req.title_config = {}
+        req.project_id = None
+        req.asset_library_id = None
+        return req
+
+    def _run_create_task(self, mock_task, mock_request, mock_db, mock_gen_repo):
+        """Helper to run create_generation_task with common mocks."""
+        from app.schemas.generation_task import GenerationTaskResponse
+
+        with (
+            patch(
+                "apps.api.app.api.routes.generation_tasks._resolve_project_and_library", return_value=("proj-1", None)
+            ),
+            patch("apps.api.app.api.routes.generation_tasks.CreateGenerationTaskUseCase") as mock_uc_cls,
+            patch("apps.api.app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True),
+            patch("apps.api.app.api.routes.generation_tasks._to_generation_task_response") as mock_resp_fn,
+        ):
+            mock_uc_cls.return_value.execute.return_value = mock_task
+            mock_resp_fn.return_value = GenerationTaskResponse(
+                id="task-1",
+                project_id="proj-1",
+                asset_library_id="",
+                strategy_id="one_take",
+                voice_library_id="",
+                template_id="tpl-1",
+                asset_ids=[],
+                title_ids=[],
+                voice_ids=[],
+                source_edit_plan_id="",
+                asset_select_mode="",
+                batch_id="",
+                video_title="",
+                resolution="",
+                bgm_config={},
+                is_preview=False,
+                source_task_id="",
+                output_width=1280,
+                output_height=720,
+                cover_url="",
+                custom_title="",
+                title_config={},
+                logs="[]",
+                status="pending",
+                progress=0.0,
+                error_message="",
+                created_at="2026-01-01T00:00:00Z",
+                updated_at="2026-01-01T00:00:00Z",
+                started_at=None,
+                completed_at=None,
+                created_by_user_id="user-1",
+                auto_retry_enabled=False,
+                auto_retry_max=0,
+                auto_retry_count=0,
+                result_count=0,
+            )
+
+            from apps.api.app.api.routes.generation_tasks import create_generation_task
+
+            return create_generation_task(
+                request=mock_request,
+                authenticated_user=MagicMock(user=MagicMock(id="user-1")),
+                generation_task_repository=mock_gen_repo,
+                project_repository=MagicMock(),
+                asset_library_repository=MagicMock(),
+                asset_repository=MagicMock(),
+                db=mock_db,
+            )
+
+    def test_fallback_sets_plan_id_when_empty(self):
+        """When source_edit_plan_id is empty, fallback finds plan via DB query."""
+        mock_task = self._make_mock_task(source_edit_plan_id="")
+        mock_request = self._make_request(source_edit_plan_id="", template_id="tpl-1")
+
+        # Mock DB query chain: db.query(EditPlanModel).filter(...).order_by(...).first()
+        mock_plan_model = MagicMock()
+        mock_plan_model.id = "plan-found-123"
+
+        mock_db = MagicMock()
+        query_chain = MagicMock()
+        query_chain.filter.return_value = query_chain
+        query_chain.order_by.return_value = query_chain
+        query_chain.first.return_value = mock_plan_model
+        mock_db.query.return_value = query_chain
+
+        mock_gen_repo = MagicMock()
+        mock_gen_repo.count_pending_by_user.return_value = 0
+        mock_gen_repo.count_pending_total.return_value = 0
+
+        self._run_create_task(mock_task, mock_request, mock_db, mock_gen_repo)
+
+        assert mock_task.source_edit_plan_id == "plan-found-123"
+        mock_gen_repo.update.assert_called_once_with(mock_task)
+
+    def test_no_fallback_when_plan_id_already_set(self):
+        """When source_edit_plan_id is already set, fallback should NOT run."""
+        mock_task = self._make_mock_task(source_edit_plan_id="plan-already-set")
+        mock_request = self._make_request(source_edit_plan_id="plan-already-set", template_id="tpl-1")
+
+        mock_db = MagicMock()
+        mock_gen_repo = MagicMock()
+        mock_gen_repo.count_pending_by_user.return_value = 0
+        mock_gen_repo.count_pending_total.return_value = 0
+
+        self._run_create_task(mock_task, mock_request, mock_db, mock_gen_repo)
+
+        assert mock_task.source_edit_plan_id == "plan-already-set"
+        mock_gen_repo.update.assert_not_called()
+
+    def test_fallback_no_match_leaves_plan_id_empty(self):
+        """When no plan matches, source_edit_plan_id stays empty."""
+        mock_task = self._make_mock_task(source_edit_plan_id="")
+        mock_request = self._make_request(source_edit_plan_id="", template_id="tpl-1")
+
+        mock_db = MagicMock()
+        query_chain = MagicMock()
+        query_chain.filter.return_value = query_chain
+        query_chain.order_by.return_value = query_chain
+        query_chain.first.return_value = None  # no matching plan
+        mock_db.query.return_value = query_chain
+
+        mock_gen_repo = MagicMock()
+        mock_gen_repo.count_pending_by_user.return_value = 0
+        mock_gen_repo.count_pending_total.return_value = 0
+
+        self._run_create_task(mock_task, mock_request, mock_db, mock_gen_repo)
+
+        assert mock_task.source_edit_plan_id == ""
+        mock_gen_repo.update.assert_not_called()
+
+    def test_fallback_handles_exception_gracefully(self):
+        """When DB query fails, the fallback should not break the main flow."""
+        mock_task = self._make_mock_task(source_edit_plan_id="")
+        mock_request = self._make_request(source_edit_plan_id="", template_id="tpl-1")
+
+        mock_db = MagicMock()
+        mock_db.query.side_effect = Exception("DB connection error")
+
+        mock_gen_repo = MagicMock()
+        mock_gen_repo.count_pending_by_user.return_value = 0
+        mock_gen_repo.count_pending_total.return_value = 0
+
+        self._run_create_task(mock_task, mock_request, mock_db, mock_gen_repo)
+
+        # Task should still be created (fallback error doesn't break main flow)
+        assert mock_task.source_edit_plan_id == ""
+        mock_gen_repo.update.assert_not_called()