chore: sync storyforge handoff state

tests/test_platform_contracts.py

@@ -2,9 +2,11 @@ from __future__ import annotations
 
 import json
 import os
+import shutil
 import sys
 import tempfile
 import unittest
+import weakref
 from pathlib import Path
 from types import SimpleNamespace
 
@@ -487,9 +489,58 @@ def _seed_domestic(db: Database, owner: dict[str, object], project_row: dict[str
     return account_id
 
 
+def _insert_domestic_creator_account(
+    db: Database,
+    owner: dict[str, object],
+    project_row: dict[str, object],
+    platform: str,
+    *,
+    suffix: str,
+    title: str,
+    handle: str,
+    bio: str,
+    tags: list[str],
+    keywords: list[str],
+) -> str:
+    now = utc_now()
+    account_id = f"{platform}_acct_contract_{suffix}"
+    db.execute(
+        """
+        INSERT INTO content_sources (
+            id, user_id, project_id, source_kind, platform, handle, source_url, title, local_path,
+            metadata_json, created_at, updated_at
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+        """,
+        (
+            account_id,
+            owner["id"],
+            project_row["id"],
+            "creator_account",
+            platform,
+            handle,
+            f"https://example.com/{platform}/profile-{suffix}",
+            title,
+            "",
+            _json(
+                {
+                    "bio": bio,
+                    "description": bio,
+                    "avatar_url": "https://example.com/avatar.png",
+                    "tags": tags,
+                    "keywords": keywords,
+                    "max_items": 5,
+                }
+            ),
+            now,
+            now,
+        ),
+    )
+    return account_id
+
+
 def _build_app(platforms: list[str]) -> tuple[FastAPI, SimpleNamespace, dict[str, object]]:
-    tmpdir = tempfile.TemporaryDirectory()
-    db = Database(str(Path(tmpdir.name) / "storyforge.db"))
+    tmpdir = Path(tempfile.mkdtemp(prefix="storyforge-platform-contracts-"))
+    db = Database(str(tmpdir / "storyforge.db"))
     db.init_schema()
     owner_row, project_row, model_row = _seed_base_account(db)
     legacy = _make_legacy(db, owner_row)
@@ -497,6 +548,7 @@ def _build_app(platforms: list[str]) -> tuple[FastAPI, SimpleNamespace, dict[str
     register_douyin_routes(app, legacy)
     for platform in platforms:
         register_domestic_platform_routes(app, legacy, platform=platform, label=platform)
+    weakref.finalize(app, shutil.rmtree, tmpdir, True)
     app.state._tmpdir = tmpdir
     app.state._legacy = legacy
     app.state._project_row = project_row
@@ -593,6 +645,323 @@ class PlatformContractTests(unittest.TestCase):
             self.assertIn("account", digest_item)
             self.assertIn("video", digest_item)
 
+    def test_kuaishou_creator_center_sync_persists_snapshots_and_analysis_context(self) -> None:
+        app, legacy, seed = _build_app(["kuaishou"])
+        with TestClient(app) as client:
+            sync = client.post(
+                "/v2/kuaishou/accounts/sync",
+                headers={"Authorization": "Bearer dummy"},
+                json={
+                    "project_id": seed["project"]["id"],
+                    "profile_url": "https://www.kuaishou.com/profile/contract-creator",
+                    "title": "快手合同账号",
+                    "handle": "contract_creator",
+                    "manual_profile_payload": {
+                        "nickname": "快手合同账号",
+                        "bio": "擅长创业内容和成交转化",
+                        "avatar_url": "https://example.com/kuaishou/avatar.png",
+                        "follower_count": 32100,
+                    },
+                    "manual_creator_pages": [
+                        {
+                            "url": "https://creator.kuaishou.com/creator/home",
+                            "title": "快手创作者中心",
+                            "payload": {
+                                "creator": {
+                                    "nickname": "快手合同账号",
+                                    "fans_count": 32100,
+                                    "play_count": 987654,
+                                    "content_tags": ["创业", "转化"],
+                                },
+                                "works": {
+                                    "published_count": 48,
+                                    "avg_finish_rate": 0.43,
+                                    "items": [
+                                        {
+                                            "video_id": "ks_work_001",
+                                            "title": "创业成交拆解 1",
+                                            "description": "拆 1 个高转化案例",
+                                            "share_url": "https://www.kuaishou.com/short-video/ks_work_001",
+                                            "cover_url": "https://example.com/kuaishou/work-1.png",
+                                            "duration_sec": 38,
+                                            "published_at": "2026-03-20T10:00:00+00:00",
+                                            "play_count": 82000,
+                                            "like_count": 4300,
+                                            "comment_count": 280,
+                                            "share_count": 140,
+                                            "tags": ["创业", "成交"]
+                                        },
+                                        {
+                                            "video_id": "ks_work_002",
+                                            "title": "口播脚本结构拆解",
+                                            "description": "复盘 3 段爆款口播结构",
+                                            "share_url": "https://www.kuaishou.com/short-video/ks_work_002",
+                                            "cover_url": "https://example.com/kuaishou/work-2.png",
+                                            "duration_sec": 46,
+                                            "published_at": "2026-03-18T10:00:00+00:00",
+                                            "play_count": 65000,
+                                            "like_count": 3100,
+                                            "comment_count": 190,
+                                            "share_count": 96,
+                                            "tags": ["口播", "结构"]
+                                        },
+                                    ],
+                                },
+                            },
+                        }
+                    ],
+                },
+            )
+            self.assertEqual(sync.status_code, 200, sync.text)
+            workspace_payload = sync.json()
+            self.assertEqual(workspace_payload["account"]["platform"], "kuaishou")
+            self.assertIsNotNone(workspace_payload["latest_public_snapshot"])
+            self.assertIsNotNone(workspace_payload["latest_creator_snapshot"])
+            self.assertEqual(workspace_payload["latest_creator_snapshot"]["snapshot_type"], "creator_center")
+            self.assertEqual(workspace_payload["account"]["nickname"], "快手合同账号")
+            self.assertGreaterEqual(workspace_payload["account"]["video_summary"]["count"], 2)
+            account_id = workspace_payload["account"]["id"]
+
+            videos = client.get(
+                f"/v2/kuaishou/accounts/{account_id}/videos",
+                headers={"Authorization": "Bearer dummy"},
+            )
+            self.assertEqual(videos.status_code, 200, videos.text)
+            videos_payload = videos.json()
+            self.assertGreaterEqual(videos_payload["count"], 2)
+            self.assertTrue(videos_payload["items"])
+            self.assertEqual(videos_payload["items"][0]["title"], "创业成交拆解 1")
+            self.assertEqual(videos_payload["items"][0]["stats"]["play"], 82000)
+            self.assertTrue(videos_payload["top_scored_video_ids"])
+            self.assertEqual(videos_payload["top_scored_video_ids"][0], videos_payload["items"][0]["id"])
+
+            snapshots = client.get(
+                f"/v2/kuaishou/accounts/{account_id}/snapshots",
+                headers={"Authorization": "Bearer dummy"},
+            )
+            self.assertEqual(snapshots.status_code, 200, snapshots.text)
+            snapshots_payload = snapshots.json()
+            self.assertGreaterEqual(len(snapshots_payload), 2)
+            creator_snapshot = next(item for item in snapshots_payload if item["snapshot_type"] == "creator_center")
+
+            creator_fields = client.get(
+                f"/v2/kuaishou/accounts/{account_id}/creator-fields",
+                headers={"Authorization": "Bearer dummy"},
+            )
+            self.assertEqual(creator_fields.status_code, 200, creator_fields.text)
+            creator_fields_payload = creator_fields.json()
+            self.assertEqual(creator_fields_payload["id"], creator_snapshot["id"])
+            self.assertEqual(creator_fields_payload["snapshot_type"], "creator_center")
+            self.assertTrue(creator_fields_payload["fields"])
+
+            analyze = client.post(
+                f"/v2/kuaishou/accounts/{account_id}/analysis",
+                headers={"Authorization": "Bearer dummy"},
+                json={
+                    "model_profile_ids": [],
+                    "linked_account_ids": [],
+                    "include_linked_accounts": True,
+                    "include_recent_similar_candidates": True,
+                    "max_videos": 6,
+                    "extra_focus": "更关注创作者中心里的成交与转化指标",
+                    "temperature": 0.35,
+                    "auto_analyze_top_videos": False,
+                    "top_video_analysis_count": 4,
+                },
+            )
+            self.assertEqual(analyze.status_code, 200, analyze.text)
+            analyze_payload = analyze.json()
+            self.assertIn("creator_center", analyze_payload["context"])
+            self.assertEqual(
+                analyze_payload["context"]["creator_center"]["latest_creator_snapshot"]["snapshot_type"],
+                "creator_center",
+            )
+            self.assertEqual(analyze_payload["context"]["requested_model_profile_ids"], [])
+            self.assertEqual(analyze_payload["context"]["selected_model_profile_ids"], [seed["model"]["id"]])
+            self.assertTrue(analyze_payload["context"]["request_options"]["include_linked_accounts"])
+            self.assertEqual(analyze_payload["context"]["linked_accounts"], [])
+            self.assertEqual(analyze_payload["top_video_analyses"], [])
+
+    def test_domestic_analysis_uses_requested_context_and_auto_top_video_followup(self) -> None:
+        app, legacy, seed = _build_app(["xiaohongshu"])
+        source_account_id = _seed_domestic(legacy.db, seed["owner"], seed["project"], "xiaohongshu")
+        linked_account_id = _insert_domestic_creator_account(
+            legacy.db,
+            seed["owner"],
+            seed["project"],
+            "xiaohongshu",
+            suffix="linked",
+            title="Linked Benchmark",
+            handle="xhs_linked",
+            bio="主打创业转化与爆款拆解",
+            tags=["创业", "转化"],
+            keywords=["创业", "转化"],
+        )
+        candidate_account_id = _insert_domestic_creator_account(
+            legacy.db,
+            seed["owner"],
+            seed["project"],
+            "xiaohongshu",
+            suffix="candidate",
+            title="Candidate Creator",
+            handle="xhs_candidate",
+            bio="专注创业口播、成交文案与转化漏斗",
+            tags=["创业", "口播", "转化"],
+            keywords=["口播", "成交"],
+        )
+        now = utc_now()
+        legacy.db.execute(
+            """
+            INSERT INTO xiaohongshu_account_relations (
+                id, user_id, source_account_id, target_account_id, target_profile_url,
+                relation_type, note, search_id, created_at
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                "xhs_relation_contract_linked",
+                seed["owner"]["id"],
+                source_account_id,
+                linked_account_id,
+                "",
+                "benchmark",
+                "linked note",
+                "",
+                now,
+            ),
+        )
+        legacy.db.execute(
+            """
+            INSERT INTO xiaohongshu_similarity_searches (
+                id, user_id, source_account_id, prompt_text, context_json, created_at
+            ) VALUES (?, ?, ?, ?, ?, ?)
+            """,
+            (
+                "xhs_search_contract_recent",
+                seed["owner"]["id"],
+                source_account_id,
+                "recent search",
+                _json({"source_account": "xiaohongshu"}),
+                now,
+            ),
+        )
+        legacy.db.execute(
+            """
+            INSERT INTO xiaohongshu_similarity_candidates (
+                id, search_id, candidate_account_id, candidate_profile_url, heuristic_score,
+                agent_score, rationale_text, dimensions_json, raw_output_json, rank_index, created_at
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                "xhs_candidate_contract_recent",
+                "xhs_search_contract_recent",
+                candidate_account_id,
+                "https://example.com/xiaohongshu/profile-candidate",
+                72,
+                72,
+                "近期相似候选",
+                _json({"tag_overlap": 2}),
+                _json({"candidate_account_id": candidate_account_id, "candidate_profile_url": "https://example.com/xiaohongshu/profile-candidate"}),
+                0,
+                now,
+            ),
+        )
+
+        with TestClient(app) as client:
+            analyze = client.post(
+                f"/v2/xiaohongshu/accounts/{source_account_id}/analysis",
+                headers={"Authorization": "Bearer dummy"},
+                json={
+                    "model_profile_ids": [seed["model"]["id"]],
+                    "linked_account_ids": [linked_account_id],
+                    "include_linked_accounts": True,
+                    "include_recent_similar_candidates": True,
+                    "max_videos": 5,
+                    "extra_focus": "关注转化路径和选题结构",
+                    "temperature": 0.32,
+                    "auto_analyze_top_videos": True,
+                    "top_video_analysis_count": 2,
+                },
+            )
+            self.assertEqual(analyze.status_code, 200, analyze.text)
+            payload = analyze.json()
+            self.assertEqual(payload["context"]["requested_model_profile_ids"], [seed["model"]["id"]])
+            self.assertEqual(payload["context"]["selected_model_profile_ids"], [seed["model"]["id"]])
+            self.assertEqual(payload["context"]["request_options"]["top_video_analysis_count"], 2)
+            self.assertEqual(len(payload["context"]["linked_accounts"]), 1)
+            self.assertEqual(payload["context"]["linked_accounts"][0]["target_account_id"], linked_account_id)
+            self.assertEqual(len(payload["context"]["recent_similar_candidates"]), 1)
+            self.assertEqual(payload["context"]["recent_similar_candidates"][0]["candidate_account_id"], candidate_account_id)
+            self.assertEqual(payload["top_video_analyses"][0]["video_id"], "xiaohongshu_video_contract_2")
+            self.assertEqual(len(payload["top_video_analyses"]), 2)
+
+    def test_domestic_similarity_search_merges_manual_urls_and_linked_candidates(self) -> None:
+        app, legacy, seed = _build_app(["xiaohongshu"])
+        source_account_id = _seed_domestic(legacy.db, seed["owner"], seed["project"], "xiaohongshu")
+        linked_account_id = _insert_domestic_creator_account(
+            legacy.db,
+            seed["owner"],
+            seed["project"],
+            "xiaohongshu",
+            suffix="similarlinked",
+            title="Linked Similar",
+            handle="xhs_similar_linked",
+            bio="创业成交、私域转化、直播承接",
+            tags=["创业", "成交", "转化"],
+            keywords=["直播", "承接"],
+        )
+        now = utc_now()
+        legacy.db.execute(
+            """
+            INSERT INTO xiaohongshu_account_relations (
+                id, user_id, source_account_id, target_account_id, target_profile_url,
+                relation_type, note, search_id, created_at
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                "xhs_relation_contract_similarity",
+                seed["owner"]["id"],
+                source_account_id,
+                linked_account_id,
+                "https://example.com/xiaohongshu/profile-similarlinked",
+                "benchmark",
+                "linked similar note",
+                "",
+                now,
+            ),
+        )
+
+        with TestClient(app) as client:
+            created = client.post(
+                "/v2/xiaohongshu/similar-searches",
+                headers={"Authorization": "Bearer dummy"},
+                json={
+                    "source_account_id": source_account_id,
+                    "candidate_urls": [
+                        "https://example.com/xiaohongshu/external-similar"
+                    ],
+                    "seed_linked_accounts": True,
+                    "search_public_pages": False,
+                    "model_profile_id": seed["model"]["id"],
+                    "max_candidates": 6,
+                    "extra_requirements": "优先找创业成交和口播拆解账号",
+                },
+            )
+            self.assertEqual(created.status_code, 200, created.text)
+            detail = client.get(
+                f"/v2/xiaohongshu/similar-searches/{created.json()['search_id']}",
+                headers={"Authorization": "Bearer dummy"},
+            )
+            self.assertEqual(detail.status_code, 200, detail.text)
+            payload = detail.json()
+            self.assertGreaterEqual(len(payload["candidates"]), 2)
+            self.assertEqual(payload["candidates"][0]["candidate_account_id"], linked_account_id)
+            manual_candidate = next(
+                item for item in payload["candidates"]
+                if item.get("candidate_profile_url") == "https://example.com/xiaohongshu/external-similar"
+            )
+            self.assertEqual(manual_candidate["candidate_account_id"], "")
+            self.assertIn("external", manual_candidate["candidate_nickname"].lower())
+
     def test_douyin_live_first_mutation_routes_are_available(self) -> None:
         app, legacy, seed = _build_app(["xiaohongshu"])
         douyin_account_id = _seed_douyin(legacy.db, seed["owner"], seed["model"])

tests/test_production_baseline.py

@@ -8,6 +8,7 @@ import subprocess
 import sys
 import tempfile
 import unittest
+from unittest import mock
 from pathlib import Path
 from typing import Any
 
@@ -217,6 +218,31 @@ class ProductionBaselineTests(unittest.TestCase):
             "password": login_password,
         }
 
+    def _insert_completed_source_job(self, ctx: dict[str, Any], *, job_id: str, title: str = "Seedance Source") -> str:
+        now = self.db_module.utc_now()
+        self.core.db.execute(
+            """
+            INSERT INTO jobs (
+                id, user_id, project_id, parent_job_id, assistant_id, knowledge_base_id, content_source_id,
+                source_type, line_type, workflow_key, orchestrator, provider_name, provider_task_id,
+                source_url, title, language, status, transcript_text, style_summary, upload_status,
+                error, artifacts_json, result_json, analysis_model_profile_id, created_at, updated_at
+            ) VALUES (?, ?, ?, '', ?, ?, NULL, 'text', 'analysis', 'analysis_pipeline', 'n8n', 'collector', '', '', ?, 'auto', 'completed', '', '', 'completed', '', '{}', '{\"summary\":\"done\"}', ?, ?, ?)
+            """,
+            (
+                job_id,
+                ctx["account_id"],
+                ctx["project_id"],
+                ctx["assistant_id"],
+                ctx["kb_id"],
+                title,
+                ctx["model_id"],
+                now,
+                now,
+            ),
+        )
+        return job_id
+
     def test_auto_session_issues_token_without_manual_credentials(self) -> None:
         ctx = self._seed_context("auto", exhausted=False)
         self.core.WEB_AUTOLOGIN_ENABLED = "1"
@@ -284,6 +310,48 @@ class ProductionBaselineTests(unittest.TestCase):
         self.assertEqual(payload["huobao_configs"]["video"]["items"][0]["provider"], "volcengine")
         self.assertEqual(payload["huobao_configs"]["video"]["items"][0]["api_key_masked"], "secr***oken")
 
+    def test_create_admin_huobao_config_forwards_is_active_flag(self) -> None:
+        ctx = self._seed_context("model_access_create", exhausted=False)
+        headers = {"Authorization": f"Bearer {ctx['token']}"}
+        captured: dict[str, object] = {}
+
+        def fake_huobao_api_request(method: str, path: str, *, payload=None, params=None, timeout: float = 12.0):
+            captured["method"] = method
+            captured["path"] = path
+            captured["payload"] = payload
+            return {
+                "id": "cfg_video_seedance",
+                "service_type": "video",
+                "provider": "volcengine",
+                "name": "Seedance",
+                "base_url": "https://video.example.com",
+                "api_key": "secret-token",
+                "model": ["seedance-2.0-pro"],
+                "priority": 100,
+                "is_active": False,
+            }
+
+        with mock.patch.object(self.core, "huobao_api_request", side_effect=fake_huobao_api_request):
+            response = self.client.post(
+                "/v2/admin/model-access/huobao-configs",
+                headers=headers,
+                json={
+                    "service_type": "video",
+                    "provider": "volcengine",
+                    "name": "Seedance",
+                    "base_url": "https://video.example.com",
+                    "api_key": "secret-token",
+                    "model": ["seedance-2.0-pro"],
+                    "priority": 100,
+                    "is_active": False,
+                    "settings": "",
+                },
+            )
+        self.assertEqual(response.status_code, 200, response.text)
+        self.assertEqual(captured["method"], "POST")
+        self.assertEqual(captured["path"], "/api/v1/ai-configs")
+        self.assertEqual((captured["payload"] or {}).get("is_active"), False)
+
     def test_admin_model_access_runtime_update_changes_effective_healthz_values(self) -> None:
         ctx = self._seed_context("runtime_access", exhausted=False)
         headers = {"Authorization": f"Bearer {ctx['token']}"}
@@ -498,7 +566,6 @@ class ProductionBaselineTests(unittest.TestCase):
             ("POST", "/v2/pipelines/content-source-sync", {"project_id": ctx["project_id"]}, None),
             ("POST", "/v2/reviews", {"project_id": ctx["project_id"], "assistant_id": ctx["assistant_id"], "title": "Review"}, None),
             ("POST", "/v2/pipelines/real-cut", {"project_id": ctx["project_id"], "title": "Cut"}, None),
-            ("POST", "/v2/pipelines/ai-video", {"project_id": ctx["project_id"], "title": "Video", "brief": "Brief"}, None),
             ("POST", f"/v2/assistants/{ctx['assistant_id']}/generate", {"brief": "Copy", "project_id": ctx["project_id"], "knowledge_base_ids": [ctx["kb_id"]]}, None),
             ("POST", "/v2/live-recorder/sources", {"project_id": ctx["project_id"], "source_url": "https://example.com/live", "title": "Live"}, None),
         ]
@@ -507,6 +574,30 @@ class ProductionBaselineTests(unittest.TestCase):
                 response = self.client.request(method, path, headers=headers, json=json_body, files=files)
                 self.assertEqual(response.status_code, 403, response.text)
 
+        with unittest.mock.patch.object(
+            self.core,
+            "huobao_api_request",
+            return_value={
+                "value": [
+                    {
+                        "id": "cfg_quota_video",
+                        "service_type": "video",
+                        "provider": "volcengine",
+                        "name": "Quota Guard",
+                        "base_url": "https://video.example.com",
+                        "model": ["seedance-2.0-pro"],
+                        "is_active": True,
+                    }
+                ]
+            },
+        ):
+            ai_video_response = self.client.post(
+                "/v2/pipelines/ai-video",
+                headers=headers,
+                json={"project_id": ctx["project_id"], "title": "Video", "brief": "Brief"},
+            )
+        self.assertEqual(ai_video_response.status_code, 403, ai_video_response.text)
+
         upload_response = self.client.post(
             "/v2/explore/upload-video",
             headers=headers,
@@ -521,6 +612,115 @@ class ProductionBaselineTests(unittest.TestCase):
         )
         self.assertEqual(upload_response.status_code, 403, upload_response.text)
 
+    def test_ai_video_rejects_when_huobao_video_config_not_ready(self) -> None:
+        ctx = self._seed_context("huobao_not_ready", exhausted=False)
+        headers = {"Authorization": f"Bearer {ctx['token']}"}
+        source_job_id = self._insert_completed_source_job(ctx, job_id="job_huobao_not_ready")
+
+        with unittest.mock.patch.object(
+            self.core,
+            "huobao_api_request",
+            side_effect=self.core.HTTPException(status_code=503, detail="HUOBAO_BASE_URL is not configured"),
+        ):
+            response = self.client.post(
+                "/v2/pipelines/ai-video",
+                headers=headers,
+                json={
+                    "project_id": ctx["project_id"],
+                    "assistant_id": ctx["assistant_id"],
+                    "knowledge_base_id": ctx["kb_id"],
+                    "source_job_id": source_job_id,
+                    "title": "Huobao Not Ready",
+                    "brief": "需要先检查火宝视频配置是否可用。",
+                    "video_provider": "doubao",
+                },
+            )
+
+        self.assertEqual(response.status_code, 503, response.text)
+        self.assertEqual(response.json()["detail"], "AI 视频暂时不可用：Huobao 视频配置未就绪，请先在管理后台完成视频配置。")
+
+    def test_ai_video_rejects_when_no_active_huobao_video_config(self) -> None:
+        ctx = self._seed_context("huobao_inactive", exhausted=False)
+        headers = {"Authorization": f"Bearer {ctx['token']}"}
+        source_job_id = self._insert_completed_source_job(ctx, job_id="job_huobao_inactive")
+
+        with unittest.mock.patch.object(
+            self.core,
+            "huobao_api_request",
+            return_value={
+                "value": [
+                    {
+                        "id": "cfg_disabled_video",
+                        "service_type": "video",
+                        "provider": "volcengine",
+                        "name": "Disabled Seedance",
+                        "base_url": "https://video.example.com",
+                        "model": ["seedance-2.0-pro"],
+                        "is_active": False,
+                    }
+                ]
+            },
+        ):
+            response = self.client.post(
+                "/v2/pipelines/ai-video",
+                headers=headers,
+                json={
+                    "project_id": ctx["project_id"],
+                    "assistant_id": ctx["assistant_id"],
+                    "knowledge_base_id": ctx["kb_id"],
+                    "source_job_id": source_job_id,
+                    "title": "Huobao Inactive",
+                    "brief": "需要启用至少一条视频配置。",
+                    "video_provider": "doubao",
+                },
+            )
+
+        self.assertEqual(response.status_code, 409, response.text)
+        self.assertEqual(response.json()["detail"], "AI 视频暂时不可用：请先在 Huobao 启用至少一条视频配置。")
+
+    def test_ai_video_rejects_when_seedance_model_not_enabled(self) -> None:
+        ctx = self._seed_context("seedance_model_missing", exhausted=False)
+        headers = {"Authorization": f"Bearer {ctx['token']}"}
+        source_job_id = self._insert_completed_source_job(ctx, job_id="job_seedance_model_missing")
+
+        with unittest.mock.patch.object(
+            self.core,
+            "huobao_api_request",
+            return_value={
+                "value": [
+                    {
+                        "id": "cfg_active_video",
+                        "service_type": "video",
+                        "provider": "volcengine",
+                        "name": "Seedance Old",
+                        "base_url": "https://video.example.com",
+                        "model": ["doubao-seedance-1-0-pro-250528"],
+                        "is_active": True,
+                    }
+                ]
+            },
+        ):
+            response = self.client.post(
+                "/v2/pipelines/ai-video",
+                headers=headers,
+                json={
+                    "project_id": ctx["project_id"],
+                    "assistant_id": ctx["assistant_id"],
+                    "knowledge_base_id": ctx["kb_id"],
+                    "source_job_id": source_job_id,
+                    "title": "Seedance Missing",
+                    "brief": "需要启用目标 Seedance 模型。",
+                    "video_provider": "seedance2",
+                    "video_model": "seedance-2.0-pro",
+                },
+            )
+
+        self.assertEqual(response.status_code, 409, response.text)
+        self.assertEqual(
+            response.json()["detail"],
+            "AI 视频暂时不可用：Huobao 启用中的视频配置未包含所选 Seedance 模型 seedance-2.0-pro。",
+        )
+
     def test_successful_analysis_records_usage_and_retry_endpoints_work(self) -> None:
         ctx = self._seed_context("happy", exhausted=False)
         headers = {"Authorization": f"Bearer {ctx['token']}"}
@@ -545,43 +745,38 @@ class ProductionBaselineTests(unittest.TestCase):
         self.assertIsNotNone(usage_row)
         self.assertEqual(text_job["status"], "queued")
 
-        now = self.db_module.utc_now()
-        source_job_id = f"job_seedance_source_{ctx['project_id']}"
-        self.core.db.execute(
-            """
-            INSERT INTO jobs (
-                id, user_id, project_id, parent_job_id, assistant_id, knowledge_base_id, content_source_id,
-                source_type, line_type, workflow_key, orchestrator, provider_name, provider_task_id,
-                source_url, title, language, status, transcript_text, style_summary, upload_status,
-                error, artifacts_json, result_json, analysis_model_profile_id, created_at, updated_at
-            ) VALUES (?, ?, ?, '', ?, ?, NULL, 'text', 'analysis', 'analysis_pipeline', 'n8n', 'collector', '', '', ?, 'auto', 'completed', '', '', 'completed', '', '{}', '{\"summary\":\"done\"}', ?, ?, ?)
-            """,
-            (
-                source_job_id,
-                ctx["account_id"],
-                ctx["project_id"],
-                ctx["assistant_id"],
-                ctx["kb_id"],
-                "Seedance Source",
-                ctx["model_id"],
-                now,
-                now,
-            ),
-        )
-        ai_video_response = self.client.post(
-            "/v2/pipelines/ai-video",
-            headers=headers,
-            json={
-                "project_id": ctx["project_id"],
-                "assistant_id": ctx["assistant_id"],
-                "knowledge_base_id": ctx["kb_id"],
-                "source_job_id": source_job_id,
-                "title": "Seedance 2.0 视频",
-                "brief": "做一条镜头推进感更强的 AI 视频。",
-                "video_provider": "seedance2",
-                "video_model": "",
+        source_job_id = self._insert_completed_source_job(ctx, job_id=f"job_seedance_source_{ctx['project_id']}")
+        with unittest.mock.patch.object(
+            self.core,
+            "huobao_api_request",
+            return_value={
+                "value": [
+                    {
+                        "id": "cfg_seedance_enabled",
+                        "service_type": "video",
+                        "provider": "volcengine",
+                        "name": "Seedance 2.0",
+                        "base_url": "https://video.example.com",
+                        "model": ["seedance-2.0-pro"],
+                        "is_active": True,
+                    }
+                ]
             },
-        )
+        ):
+            ai_video_response = self.client.post(
+                "/v2/pipelines/ai-video",
+                headers=headers,
+                json={
+                    "project_id": ctx["project_id"],
+                    "assistant_id": ctx["assistant_id"],
+                    "knowledge_base_id": ctx["kb_id"],
+                    "source_job_id": source_job_id,
+                    "title": "Seedance 2.0 视频",
+                    "brief": "做一条镜头推进感更强的 AI 视频。",
+                    "video_provider": "seedance2",
+                    "video_model": "",
+                },
+            )
         self.assertEqual(ai_video_response.status_code, 200, ai_video_response.text)
         ai_video_payload = ai_video_response.json()
         self.assertEqual(ai_video_payload["artifacts"]["video_provider"], "seedance2")
@@ -589,6 +784,77 @@ class ProductionBaselineTests(unittest.TestCase):
         self.assertEqual(ai_video_payload["artifacts"]["video_dispatch_provider"], "doubao")
         self.assertEqual(ai_video_payload["artifacts"]["video_dispatch_model"], "seedance-2.0-pro")
 
+    def test_ai_video_seedance_requires_ready_video_config(self) -> None:
+        ctx = self._seed_context("seedance_guard", exhausted=False)
+        headers = {"Authorization": f"Bearer {ctx['token']}"}
+        source_job_id = self._insert_completed_source_job(ctx, job_id=f"job_seedance_guard_{ctx['project_id']}", title="Seedance Guard Source")
+
+        with mock.patch.object(
+            self.core,
+            "huobao_api_request",
+            side_effect=self.core.HTTPException(status_code=503, detail="HUOBAO_BASE_URL is not configured"),
+        ):
+            response = self.client.post(
+                "/v2/pipelines/ai-video",
+                headers=headers,
+                json={
+                    "project_id": ctx["project_id"],
+                    "assistant_id": ctx["assistant_id"],
+                    "knowledge_base_id": ctx["kb_id"],
+                    "source_job_id": source_job_id,
+                    "title": "Seedance 校验失败",
+                    "brief": "做一条 Seedance 视频。",
+                    "video_provider": "seedance2",
+                    "video_model": "seedance-2.0-pro",
+                },
+            )
+        self.assertEqual(response.status_code, 503, response.text)
+        self.assertIn("Huobao 视频配置未就绪", response.text)
+
+    def test_ai_video_seedance_requires_matching_active_model(self) -> None:
+        ctx = self._seed_context("seedance_model_guard", exhausted=False)
+        headers = {"Authorization": f"Bearer {ctx['token']}"}
+        source_job_id = self._insert_completed_source_job(
+            ctx,
+            job_id=f"job_seedance_model_guard_{ctx['project_id']}",
+            title="Seedance Model Guard Source",
+        )
+
+        with mock.patch.object(
+            self.core,
+            "huobao_api_request",
+            return_value={
+                "value": [
+                    {
+                        "id": "cfg_video_default",
+                        "service_type": "video",
+                        "provider": "volcengine",
+                        "name": "Seedance Legacy",
+                        "base_url": "https://video.example.com",
+                        "model": ["doubao-seedance-1-0-pro-250528"],
+                        "is_active": True,
+                    }
+                ]
+            },
+        ):
+            response = self.client.post(
+                "/v2/pipelines/ai-video",
+                headers=headers,
+                json={
+                    "project_id": ctx["project_id"],
+                    "assistant_id": ctx["assistant_id"],
+                    "knowledge_base_id": ctx["kb_id"],
+                    "source_job_id": source_job_id,
+                    "title": "Seedance 模型缺失",
+                    "brief": "做一条 Seedance 视频。",
+                    "video_provider": "seedance2",
+                    "video_model": "seedance-2.0-pro",
+                },
+            )
+        self.assertEqual(response.status_code, 409, response.text)
+        self.assertIn("seedance-2.0-pro", response.text)
+        self.assertIn("未包含", response.text)
+
         now = self.db_module.utc_now()
         failed_jobs = []
         for index in range(2):
@@ -660,6 +926,9 @@ class ProductionBaselineTests(unittest.TestCase):
         )
         backup_path = Path(result.stdout.strip().splitlines()[-1])
         self.assertTrue(backup_path.exists(), result.stdout)
-        with sqlite3.connect(backup_path) as conn:
+        conn = sqlite3.connect(backup_path)
+        try:
             account_count = conn.execute("SELECT COUNT(*) FROM accounts").fetchone()[0]
+        finally:
+            conn.close()
         self.assertGreaterEqual(int(account_count), 1)