diff --git a/tests/unit/test_shared_ai_service.py b/tests/unit/test_shared_ai_service.py
new file mode 100755
index 000000000..f46002dbf
--- /dev/null
+++ b/tests/unit/test_shared_ai_service.py
@@ -0,0 +1,453 @@
+"""shared.ai_service 单元测试.
+
+主要测试纯逻辑部分：_parse_recommend_response / _fallback_recommend_clips / _call_ai_cover_service.
+"""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import patch
+
+import pytest
+from shared.ai_service import (
+    _call_ai_cover_service,
+    _fallback_recommend_clips,
+    _parse_recommend_response,
+)
+
+# ── _parse_recommend_response 测试 ────────────────────────────────────────
+
+
+class TestParseRecommendResponseBasic:
+    """基础解析测试."""
+
+    def test_parse_valid_json(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {
+                        "clip_type": "intro",
+                        "order": 0,
+                        "text_content": "开场",
+                        "duration": 3.0,
+                        "transition_effect": "fade",
+                        "asset_id": "asset1",
+                        "start_time": 0.0,
+                        "config": {},
+                    },
+                    {
+                        "clip_type": "outro",
+                        "order": 1,
+                        "text_content": "结尾",
+                        "duration": 2.0,
+                        "transition_effect": "fade",
+                        "asset_id": "",
+                        "start_time": 0.0,
+                        "config": {},
+                    },
+                ],
+                "title": "测试视频",
+                "confidence": 0.85,
+            }
+        )
+        result = _parse_recommend_response(content, ["asset1"], 30.0)
+        assert result is not None
+        assert len(result["clips"]) == 2
+        assert result["confidence"] == 0.85
+        assert result["total_duration"] == 5.0
+        assert result["config"]["title"]["text"] == "测试视频"
+        assert result["config"]["title"]["ai_auto"] is True
+
+    def test_parse_none_returns_none(self):
+        result = _parse_recommend_response(None, ["a1"], 30.0)  # type: ignore[arg-type]
+        assert result is None
+
+    def test_parse_empty_string_returns_none(self):
+        result = _parse_recommend_response("", ["a1"], 30.0)
+        assert result is None
+
+    def test_parse_whitespace_only_returns_none(self):
+        result = _parse_recommend_response("   ", ["a1"], 30.0)
+        assert result is None
+
+    def test_parse_invalid_json_returns_none(self):
+        result = _parse_recommend_response("not json", ["a1"], 30.0)
+        assert result is None
+
+    def test_parse_non_dict_json_returns_none(self):
+        result = _parse_recommend_response("[1, 2, 3]", ["a1"], 30.0)
+        assert result is None
+
+
+class TestParseRecommendResponseClips:
+    """clips 解析测试."""
+
+    def test_parse_no_clips_returns_none(self):
+        content = json.dumps({"title": "test", "clips": []})
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is None
+
+    def test_parse_clips_not_list_returns_none(self):
+        content = json.dumps({"clips": "not a list"})
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is None
+
+    def test_parse_clips_sorted_by_order(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "outro", "order": 2, "duration": 2, "asset_id": "a1"},
+                    {"clip_type": "intro", "order": 0, "duration": 3, "asset_id": "a1"},
+                    {"clip_type": "showcase", "order": 1, "duration": 5, "asset_id": "a1"},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert len(result["clips"]) == 3
+        assert result["clips"][0]["clip_type"] == "intro"
+        assert result["clips"][1]["clip_type"] == "showcase"
+        assert result["clips"][2]["clip_type"] == "outro"
+
+    def test_parse_clips_renumbered_continuously(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "intro", "order": 10, "duration": 2, "asset_id": "a1"},
+                    {"clip_type": "outro", "order": 20, "duration": 2, "asset_id": "a1"},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["clips"][0]["order"] == 0
+        assert result["clips"][1]["order"] == 1
+
+    def test_parse_skips_invalid_clip_dicts(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"},
+                    "not a dict",
+                    {"clip_type": "outro", "order": 2, "duration": 2, "asset_id": "a1"},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert len(result["clips"]) == 2
+
+
+class TestParseRecommendResponseFields:
+    """各字段解析与边界测试."""
+
+    def test_parse_duration_clamped_min(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "intro", "order": 0, "duration": 0.5, "asset_id": "a1"},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["clips"][0]["duration"] == 1.0
+
+    def test_parse_duration_clamped_max(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "intro", "order": 0, "duration": 100, "asset_id": "a1"},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["clips"][0]["duration"] == 30.0
+
+    def test_parse_start_time_clamped_min(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1", "start_time": -5.0},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["clips"][0]["start_time"] == 0.0
+
+    def test_parse_asset_id_not_in_list_empty(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "unknown_asset"},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1", "a2"], 30.0)
+        assert result is not None
+        assert result["clips"][0]["asset_id"] == ""
+
+    def test_parse_asset_id_in_list_kept(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a2"},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1", "a2"], 30.0)
+        assert result is not None
+        assert result["clips"][0]["asset_id"] == "a2"
+
+    def test_parse_default_values(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"order": 0},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        clip = result["clips"][0]
+        assert clip["clip_type"] == "showcase"
+        assert clip["text_content"] == ""
+        assert clip["duration"] == 3.0
+        assert clip["transition_effect"] == "cut"
+        assert clip["asset_id"] == ""
+        assert clip["start_time"] == 0.0
+        assert clip["config"] == {}
+
+
+class TestParseRecommendResponseMarkdown:
+    """Markdown 代码块包裹的 JSON 测试."""
+
+    def test_parse_markdown_json(self):
+        content = (
+            "```json\n"
+            + json.dumps(
+                {
+                    "clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
+                    "title": "md test",
+                }
+            )
+            + "\n```"
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert len(result["clips"]) == 1
+        assert result["config"]["title"]["text"] == "md test"
+
+    def test_parse_backticks_no_language(self):
+        content = (
+            "```\n"
+            + json.dumps(
+                {
+                    "clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
+                }
+            )
+            + "\n```"
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert len(result["clips"]) == 1
+
+
+class TestParseRecommendResponseConfidence:
+    """confidence 解析测试."""
+
+    def test_parse_confidence_normal(self):
+        content = json.dumps(
+            {
+                "clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
+                "confidence": 0.85,
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["confidence"] == 0.85
+
+    def test_parse_confidence_clamped_min(self):
+        content = json.dumps(
+            {
+                "clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
+                "confidence": -0.5,
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["confidence"] == 0.0
+
+    def test_parse_confidence_clamped_max(self):
+        content = json.dumps(
+            {
+                "clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
+                "confidence": 1.5,
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["confidence"] == 1.0
+
+    def test_parse_confidence_default(self):
+        content = json.dumps(
+            {
+                "clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["confidence"] == 0.7
+
+
+class TestParseRecommendResponseConfig:
+    """config 生成测试."""
+
+    def test_parse_no_title_no_ai_auto(self):
+        content = json.dumps(
+            {
+                "clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        # 没有 title 时，config 的 title.text 保持默认（DEFAULT_EDIT_PLAN_CONFIG 中的值）
+        assert "title" in result["config"]
+
+    def test_parse_config_is_deep_copy(self):
+        content = json.dumps(
+            {
+                "clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
+                "title": "test",
+            }
+        )
+        result1 = _parse_recommend_response(content, ["a1"], 30.0)
+        result2 = _parse_recommend_response(content, ["a1"], 30.0)
+        # 修改其中一个不影响另一个
+        result1["config"]["title"]["text"] = "modified"
+        assert result2["config"]["title"]["text"] != "modified"
+
+
+class TestParseRecommendResponseTotalDuration:
+    """total_duration 计算测试."""
+
+    def test_parse_total_duration_sum(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "intro", "order": 0, "duration": 3.5, "asset_id": "a1"},
+                    {"clip_type": "showcase", "order": 1, "duration": 5.2, "asset_id": "a1"},
+                    {"clip_type": "outro", "order": 2, "duration": 2.0, "asset_id": "a1"},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["total_duration"] == pytest.approx(10.7, abs=0.01)
+
+
+# ── _fallback_recommend_clips 测试 ────────────────────────────────────────
+
+
+class TestFallbackRecommendClips:
+    """本地降级推荐方案测试."""
+
+    def test_fallback_returns_dict_with_clips(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2"], "one_take", 30.0)
+        assert "clips" in result
+        assert "config" in result
+        assert "total_duration" in result
+        assert "confidence" in result
+
+    def test_fallback_has_intro_and_outro(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2"], "one_take", 30.0)
+        clips = result["clips"]
+        assert clips[0]["clip_type"] == "intro"
+        assert clips[-1]["clip_type"] == "outro"
+
+    def test_fallback_showcase_count_matches_assets(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2", "a3"], "one_take", 30.0)
+        showcase_clips = [c for c in result["clips"] if c["clip_type"] == "showcase"]
+        assert len(showcase_clips) == 3
+
+    def test_fallback_no_assets_still_works(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", [], "one_take", 30.0)
+        assert len(result["clips"]) >= 2  # 至少有intro和outro
+
+    def test_fallback_intro_uses_first_asset(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2"], "one_take", 30.0)
+        assert result["clips"][0]["asset_id"] == "a1"
+
+    def test_fallback_outro_has_empty_asset(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1"], "one_take", 30.0)
+        assert result["clips"][-1]["asset_id"] == ""
+
+    def test_fallback_confidence_in_range(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1"], "one_take", 30.0)
+        assert 0.75 <= result["confidence"] <= 0.95
+
+    def test_fallback_title_contains_asset_count(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2", "a3"], "one_take", 30.0)
+        assert "3" in result["config"]["title"]["text"]
+        assert result["config"]["title"]["ai_auto"] is True
+
+    def test_fallback_total_duration_matches(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2"], "one_take", 30.0)
+        total = sum(c["duration"] for c in result["clips"])
+        assert result["total_duration"] == round(total, 1)
+
+    def test_fallback_orders_are_sequential(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2", "a3"], "one_take", 30.0)
+        orders = [c["order"] for c in result["clips"]]
+        assert orders == list(range(len(result["clips"])))
+
+
+# ── _call_ai_cover_service 测试 ───────────────────────────────────────────
+
+
+class TestAiCoverService:
+    """AI封面生成服务测试."""
+
+    def test_cover_type_upload(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _call_ai_cover_service("plan1", ["a1"], "upload")
+        assert result["type"] == "upload"
+        assert result["image_url"] == ""
+
+    def test_cover_type_manual_with_frame_time(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _call_ai_cover_service("plan1", ["a1"], "manual", frame_time=5.5)
+        assert result["type"] == "manual"
+        assert result["frame_time"] == 5.5
+        assert "5.5" in result["image_url"]
+
+    def test_cover_type_ai_frame(self):
+        with patch("shared.ai_service.time.sleep"):
+            with patch("shared.ai_service.random.uniform", side_effect=[5.0, 0.9]):
+                result = _call_ai_cover_service("plan1", ["a1"], "ai_frame")
+        assert result["type"] == "ai_frame"
+        assert result["frame_time"] == 5.0
+        assert result["confidence"] == 0.9
+        assert "plan1" in result["image_url"]
+
+    def test_cover_type_ai_regenerate(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _call_ai_cover_service("plan1", ["a1"], "ai_regenerate")
+        assert result["type"] == "ai_frame"
+
+    def test_cover_frame_time_in_range(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _call_ai_cover_service("plan1", ["a1"], "ai_frame")
+        assert 1.0 <= result["frame_time"] <= 10.0