fix: harden story-system runtime and add skill-flow integration test

webnovel-writer/references/csv/题材与调性推理.csv

@@ -0,0 +1,4 @@
+编号,适用技能,分类,层级,关键词,意图与同义词,适用题材,大模型指令,核心摘要,详细展开,题材/流派,题材别名,核心调性,节奏策略,强制禁忌/毒点,推荐基础检索表,推荐动态检索表,默认查询词
+GR-001,write|plan,题材路由,知识补充,玄幻退婚流|退婚流|废材逆袭,退婚打脸怎么写|莫欺少年穷|三年之约怎么立,玄幻|仙侠,先压后爆，耻辱必须转成长线兑现。,"退婚或逐出型起手必须先把尊严踩到底，再把反击延迟兑现为长线承诺。","退婚流不靠瞬间翻盘，而靠压抑、立誓、补刀和后续兑现形成持续追读欲。玄幻退婚流要先给耻辱和压抑，再给立誓与首轮反打，禁止一章内把压抑和兑现写成流水账。",玄幻退婚流,退婚流|废材逆袭,先压后爆,三章内必须有首次有效反打,打脸不能软收尾|主角还没兑现就被配角代打,命名规则|人设与关系|金手指与设定,桥段套路|爽点与节奏|场景写法,退婚|打脸|废材逆袭
+GR-002,write|plan,题材路由,知识补充,规则动物园|规则怪谈动物园|规则怪谈,规则怪谈怎么写|动物园规则怎么写|违反规则会怎样,规则怪谈|悬疑|惊悚,规则先立死，再逐层揭示代价和漏洞。,"规则类故事必须先建立清晰规则，再让角色在遵守、试探、破坏之间持续换取信息。","读者爽点来自规则被一步步解密与反利用，而不是角色无缘无故乱闯。规则动物园要先把规则、异样征兆和试探成本写清，再给局部违反后的后果与解法。",规则动物园,规则怪谈动物园|规则怪谈,高压克制,先立规则后破局,规则写得像背景板|处罚没有代价|谜底提前透光,命名规则|场景写法|人设与关系,桥段套路|爽点与节奏|写作技法,规则|动物园|守则
+GR-003,write|plan,题材路由,知识补充,压抑后爆|先抑后扬|忍耐爆发,先压后爆怎么写|情绪爆发怎么写|压抑蓄力怎么排,全部,压抑必须具体，爆发必须改局。,"情绪爆发型章节要让限制、损失和退让持续累加，再在不可回避点集中兑现。","前段负责让读者和角色一起憋，后段负责一次性改写关系、局面或规则。压抑后爆题材要把前段损失写实，后段兑现写硬，不能只靠口号和情绪词。",压抑后爆,先抑后扬|忍耐爆发,压抑蓄力后强兑现,限制累加到临界点再爆发,前面没有真实压抑|爆发不改局面|爆发后立刻归零,写作技法|爽点与节奏|场景写法,桥段套路|爽点与节奏|场景写法,压抑|爆发|反打

webnovel-writer/scripts/data_modules/context_manager.py

@@ -9,6 +9,7 @@ import json
 import re
 import sys
 import logging
+import hashlib
 from pathlib import Path
 
 from runtime_compat import enable_windows_utf8_stdio
@@ -123,7 +124,16 @@ class ContextManager:
             return False
 
         required_sections = {"plot_structure", "long_term_memory", "story_contract", "prewrite_validation"}
-        return required_sections.issubset(set(sections.keys()))
+        if not required_sections.issubset(set(sections.keys())):
+            return False
+
+        chapter = int(cached.get("chapter") or (payload.get("meta") or {}).get("chapter") or 0)
+        if chapter <= 0:
+            return False
+
+        snapshot_signature = meta.get("story_contract_signature")
+        current_signature = self._story_contract_signature(chapter)
+        return snapshot_signature == current_signature
 
     def build_context(
         self,
@@ -154,7 +164,10 @@ class ContextManager:
         assembled = self.assemble_context(pack, template=template, max_chars=max_chars)
 
         if save_snapshot:
-            meta = {"template": template}
+            meta = {
+                "template": template,
+                "story_contract_signature": self._story_contract_signature(chapter),
+            }
             self.snapshot_manager.save_snapshot(chapter, assembled, meta=meta)
 
         return assembled
@@ -288,6 +301,7 @@ class ContextManager:
             chapter=chapter,
             review_contract=story_contract.get("review_contract") or {},
             plot_structure=plot_structure,
+            story_contract=story_contract,
         )
 
         return {
@@ -732,6 +746,25 @@ class ContextManager:
             "anti_patterns": read_json_if_exists(story_root / "anti_patterns.json") or [],
         }
 
+    def _story_contract_signature(self, chapter: int) -> Dict[str, str]:
+        story_root = self.config.story_system_dir
+        volume = volume_num_for_chapter_from_state(self.config.project_root, chapter) or 1
+        paths = {
+            "master_setting": story_root / "MASTER_SETTING.json",
+            "chapter_brief": story_root / "chapters" / f"chapter_{chapter:03d}.json",
+            "volume_brief": story_root / "volumes" / f"volume_{volume:03d}.json",
+            "review_contract": story_root / "reviews" / f"chapter_{chapter:03d}.review.json",
+            "anti_patterns": story_root / "anti_patterns.json",
+        }
+        signature: Dict[str, str] = {}
+        for name, path in paths.items():
+            if not path.is_file():
+                signature[name] = "missing"
+                continue
+            digest = hashlib.sha1(path.read_bytes()).hexdigest()
+            signature[name] = digest
+        return signature
+
     def _load_recent_summaries(self, chapter: int, window: int = 3) -> List[Dict[str, Any]]:
         summaries = []
         for ch in range(max(1, chapter - window), chapter):

webnovel-writer/scripts/data_modules/event_projection_router.py

@@ -10,13 +10,13 @@ class EventProjectionRouter:
         "character_state_changed": ["state", "memory"],
         "power_breakthrough": ["state", "memory"],
         "relationship_changed": ["index"],
-        "world_rule_revealed": ["index", "memory"],
-        "world_rule_broken": ["index", "memory"],
+        "world_rule_revealed": ["memory"],
+        "world_rule_broken": ["memory"],
         "open_loop_created": ["memory"],
         "open_loop_closed": ["memory"],
         "promise_created": ["memory"],
         "promise_paid_off": ["memory"],
-        "artifact_obtained": ["index", "memory"],
+        "artifact_obtained": ["index"],
     }
 
     def route(self, event: Dict) -> List[str]:

webnovel-writer/scripts/data_modules/index_projection_writer.py

@@ -55,4 +55,31 @@ class IndexProjectionWriter:
                             "chapter": chapter,
                         }
                     )
+            elif event_type == "artifact_obtained":
+                entity_id = str(
+                    payload.get("artifact_id")
+                    or payload.get("entity_id")
+                    or payload.get("id")
+                    or event.get("subject")
+                    or ""
+                ).strip()
+                if not entity_id:
+                    continue
+                current = {}
+                owner = str(payload.get("owner") or payload.get("holder") or "").strip()
+                location = str(payload.get("location") or "").strip()
+                if owner:
+                    current["holder"] = owner
+                if location:
+                    current["location"] = location
+                deltas.append(
+                    {
+                        "entity_id": entity_id,
+                        "canonical_name": str(payload.get("name") or event.get("subject") or entity_id).strip(),
+                        "type": str(payload.get("type") or "物品").strip() or "物品",
+                        "current": current,
+                        "desc": str(payload.get("description") or "").strip(),
+                        "chapter": chapter,
+                    }
+                )
         return deltas

webnovel-writer/scripts/data_modules/prewrite_validator.py

@@ -16,16 +16,34 @@ class PrewriteValidator:
         chapter: int,
         review_contract: Dict[str, Any],
         plot_structure: Dict[str, Any],
+        story_contract: Dict[str, Any] | None = None,
     ) -> Dict[str, Any]:
         state = json.loads(
             (self.project_root / ".webnovel" / "state.json").read_text(encoding="utf-8")
         )
         pending = state.get("disambiguation_pending") or []
         warnings = state.get("disambiguation_warnings") or []
+        contract_provided = story_contract is not None
+        story_contract = story_contract or {}
+        missing_contracts = []
+        if contract_provided:
+            missing_contracts = [
+                name
+                for name in ("master_setting", "chapter_brief", "volume_brief", "review_contract")
+                if not story_contract.get(name)
+            ]
+        blocking_reasons = []
+        if pending:
+            blocking_reasons.append("存在高优先级 disambiguation_pending")
+        if missing_contracts:
+            blocking_reasons.append(
+                "缺少 Story System 合同: " + ", ".join(missing_contracts)
+            )
         return {
             "chapter": chapter,
-            "blocking": bool(pending),
-            "blocking_reasons": ["存在高优先级 disambiguation_pending"] if pending else [],
+            "blocking": bool(pending) or bool(missing_contracts),
+            "blocking_reasons": blocking_reasons,
+            "missing_contracts": missing_contracts,
             "forbidden_zones": list(review_contract.get("blocking_rules") or []),
             "disambiguation_domain": {
                 "pending_count": len(pending),

webnovel-writer/scripts/data_modules/tests/test_context_manager.py

@@ -278,6 +278,128 @@ def test_context_manager_includes_story_contract_and_prewrite_validation(temp_pr
     assert list(sections.keys()).index("story_contract") < list(sections.keys()).index("scene")
 
 
+def test_context_manager_invalidates_snapshot_when_story_contract_changes(temp_project):
+    state = {
+        "progress": {"volumes_planned": [{"volume": 1, "chapters_range": "1-10"}]},
+        "protagonist_state": {"name": "萧炎"},
+        "chapter_meta": {},
+        "disambiguation_warnings": [],
+        "disambiguation_pending": [],
+    }
+    temp_project.state_file.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
+
+    story_root = temp_project.story_system_dir
+    (story_root / "chapters").mkdir(parents=True, exist_ok=True)
+    (story_root / "volumes").mkdir(parents=True, exist_ok=True)
+    (story_root / "reviews").mkdir(parents=True, exist_ok=True)
+    (story_root / "MASTER_SETTING.json").write_text(
+        json.dumps(
+            {
+                "meta": {"schema_version": "story-system/v1", "contract_type": "MASTER_SETTING"},
+                "route": {"primary_genre": "玄幻退婚流"},
+                "master_constraints": {"core_tone": "先压后爆"},
+                "base_context": [],
+                "source_trace": [],
+                "override_policy": {},
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+    (story_root / "chapters" / "chapter_003.json").write_text(
+        json.dumps(
+            {
+                "meta": {"schema_version": "story-system/v1", "contract_type": "CHAPTER_BRIEF", "chapter": 3},
+                "override_allowed": {"chapter_focus": "旧焦点"},
+                "dynamic_context": [],
+                "source_trace": [],
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+    (story_root / "volumes" / "volume_001.json").write_text(
+        json.dumps(
+            {
+                "meta": {"schema_version": "story-system/v1", "contract_type": "VOLUME_BRIEF"},
+                "volume_goal": {"summary": "旧卷目标"},
+                "selected_tropes": [],
+                "selected_pacing": {},
+                "selected_scenes": [],
+                "anti_patterns": [],
+                "system_constraints": [],
+                "overrides": {"locked": {}, "append_only": {}, "override_allowed": {}},
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+    review_path = story_root / "reviews" / "chapter_003.review.json"
+    review_path.write_text(
+        json.dumps(
+            {
+                "meta": {"schema_version": "story-system/v1", "contract_type": "REVIEW_CONTRACT"},
+                "must_check": ["旧节点"],
+                "blocking_rules": ["旧禁区"],
+                "genre_specific_risks": [],
+                "anti_patterns": [],
+                "system_constraints": [],
+                "review_thresholds": {},
+                "overrides": {"locked": {}, "append_only": {}, "override_allowed": {}},
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+    temp_project.outline_dir.mkdir(parents=True, exist_ok=True)
+    (temp_project.outline_dir / "第1卷-详细大纲.md").write_text(
+        "### 第3章：试炼\n必须覆盖节点：旧节点\n本章禁区：旧禁区",
+        encoding="utf-8",
+    )
+
+    manager = ContextManager(temp_project)
+    first = manager.build_context(3, template="plot", use_snapshot=True, save_snapshot=True)
+    assert first["sections"]["prewrite_validation"]["content"]["forbidden_zones"] == ["旧禁区"]
+
+    review_path.write_text(
+        json.dumps(
+            {
+                "meta": {"schema_version": "story-system/v1", "contract_type": "REVIEW_CONTRACT"},
+                "must_check": ["新节点"],
+                "blocking_rules": ["新禁区"],
+                "genre_specific_risks": [],
+                "anti_patterns": [],
+                "system_constraints": [],
+                "review_thresholds": {},
+                "overrides": {"locked": {}, "append_only": {}, "override_allowed": {}},
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+
+    second = manager.build_context(3, template="plot", use_snapshot=True, save_snapshot=False)
+    assert second["sections"]["story_contract"]["content"]["review_contract"]["blocking_rules"] == ["新禁区"]
+    assert second["sections"]["prewrite_validation"]["content"]["forbidden_zones"] == ["新禁区"]
+
+
+def test_context_manager_blocks_when_story_contract_missing(temp_project):
+    state = {
+        "protagonist_state": {"name": "萧炎"},
+        "chapter_meta": {},
+        "disambiguation_warnings": [],
+        "disambiguation_pending": [],
+    }
+    temp_project.state_file.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
+
+    manager = ContextManager(temp_project)
+    payload = manager.build_context(3, use_snapshot=False, save_snapshot=False)
+
+    prewrite = payload["sections"]["prewrite_validation"]["content"]
+    assert prewrite["blocking"] is True
+    assert any("合同" in reason for reason in prewrite["blocking_reasons"])
+
+
 def test_query_router():
     router = QueryRouter()
     assert router.route("角色是谁") == "entity"

webnovel-writer/scripts/data_modules/tests/test_event_projection_router.py

@@ -24,6 +24,18 @@ def test_router_maps_relationship_changed_to_index():
     assert "index" in targets
 
 
+def test_router_maps_world_rule_broken_to_memory_only():
+    router = EventProjectionRouter()
+    targets = router.route(
+        {
+            "event_type": "world_rule_broken",
+            "subject": "金手指",
+            "payload": {"field": "world_rule"},
+        }
+    )
+    assert targets == ["memory"]
+
+
 def test_router_collects_required_writers_from_commit_payload():
     router = EventProjectionRouter()
     targets = router.required_writers(

webnovel-writer/scripts/data_modules/tests/test_prewrite_validator.py

@@ -32,3 +32,35 @@ def test_prewrite_validator_builds_disambiguation_domain_and_fulfillment_seed(tm
     assert payload["blocking"] is False
     assert payload["fulfillment_seed"]["planned_nodes"] == ["发现陷阱"]
     assert payload["disambiguation_domain"]["pending_count"] == 0
+
+
+def test_prewrite_validator_blocks_when_required_contracts_missing(tmp_path):
+    project_root = tmp_path
+    (project_root / ".webnovel").mkdir(parents=True, exist_ok=True)
+    (project_root / ".webnovel" / "state.json").write_text(
+        json.dumps(
+            {
+                "disambiguation_pending": [],
+                "disambiguation_warnings": [],
+                "chapter_meta": {},
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+
+    payload = PrewriteValidator(project_root).build(
+        chapter=3,
+        review_contract={},
+        plot_structure={},
+        story_contract={
+            "master_setting": {},
+            "chapter_brief": {},
+            "volume_brief": {},
+            "review_contract": {},
+        },
+    )
+
+    assert payload["blocking"] is True
+    assert "missing_contracts" in payload
+    assert set(payload["missing_contracts"]) >= {"master_setting", "review_contract"}

webnovel-writer/scripts/data_modules/tests/test_projection_writers.py

@@ -134,6 +134,37 @@ def test_index_projection_writer_derives_relationship_from_event(tmp_path):
     assert rels[0]["type"] == "师徒"
 
 
+def test_index_projection_writer_derives_artifact_entity_from_event(tmp_path):
+    cfg = DataModulesConfig.from_project_root(tmp_path)
+    cfg.ensure_dirs()
+    writer = IndexProjectionWriter(tmp_path)
+
+    result = writer.apply(
+        {
+            "meta": {"status": "accepted", "chapter": 3},
+            "entity_deltas": [],
+            "accepted_events": [
+                {
+                    "event_id": "evt-002",
+                    "chapter": 3,
+                    "event_type": "artifact_obtained",
+                    "subject": "黑戒",
+                    "payload": {
+                        "artifact_id": "black_ring",
+                        "name": "黑戒",
+                        "owner": "xiaoyan",
+                    },
+                }
+            ],
+        }
+    )
+
+    entity = IndexManager(cfg).get_entity("black_ring")
+    assert result["applied"] is True
+    assert entity["canonical_name"] == "黑戒"
+    assert entity["current_json"]["holder"] == "xiaoyan"
+
+
 def test_summary_projection_writer_writes_summary_markdown(tmp_path):
     cfg = DataModulesConfig.from_project_root(tmp_path)
     cfg.ensure_dirs()

webnovel-writer/scripts/data_modules/tests/test_story_system_cli.py

@@ -101,3 +101,29 @@ def test_markdown_writer_preserves_manual_notes_outside_markers(tmp_path):
     assert "手工备注" in text
     assert "## Auto" in text
     assert "旧内容" not in text
+
+
+def test_story_system_default_csv_dir_routes_real_genre_seed(tmp_path, monkeypatch, capsys):
+    project_root = tmp_path / "book"
+    (project_root / ".webnovel").mkdir(parents=True, exist_ok=True)
+    (project_root / ".webnovel" / "state.json").write_text("{}", encoding="utf-8")
+
+    from story_system import main
+
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "story_system",
+            "玄幻退婚流",
+            "--project-root",
+            str(project_root),
+            "--format",
+            "json",
+        ],
+    )
+    main()
+
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["master_setting"]["route"]["primary_genre"] == "玄幻退婚流"
+    assert payload["master_setting"]["route"]["route_source"] != "empty_csv_fallback"

webnovel-writer/scripts/data_modules/tests/test_webnovel_unified_cli.py

@@ -1,6 +1,8 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
+import asyncio
+import importlib
 import json
 import sys
 from pathlib import Path
@@ -454,3 +456,221 @@ def test_review_pipeline_main_creates_output_directories(tmp_path):
         sys.argv = old_argv
 
     assert metrics_out.is_file()
+
+
+def test_webnovel_skill_flow_runs_story_contract_context_and_review_pipeline_with_stubbed_vector_model(
+    monkeypatch, tmp_path, capsys
+):
+    _ensure_scripts_on_path()
+    module = _load_webnovel_module()
+    import data_modules.rag_adapter as rag_module
+    from data_modules.config import DataModulesConfig
+
+    project_root = (tmp_path / "book").resolve()
+    cfg = DataModulesConfig.from_project_root(project_root)
+    cfg.ensure_dirs()
+
+    cfg.state_file.write_text(
+        json.dumps(
+            {
+                "project": {"genre": "xuanhuan"},
+                "progress": {
+                    "current_chapter": 3,
+                    "total_words": 9000,
+                    "volumes_planned": [{"volume": 1, "chapters_range": "1-20"}],
+                },
+                "protagonist_state": {
+                    "name": "萧炎",
+                    "location": {"current": "天云宗外院"},
+                    "power": {"realm": "斗者", "layer": 9},
+                },
+                "chapter_meta": {},
+                "disambiguation_warnings": [],
+                "disambiguation_pending": [],
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+
+    outline_dir = project_root / "大纲"
+    outline_dir.mkdir(parents=True, exist_ok=True)
+    (outline_dir / "第1卷-详细大纲.md").write_text(
+        "\n".join(
+            [
+                "### 第3章：试炼冲突",
+                "本章将聚焦萧炎与药老关系冲突，并回收旧线索真相。",
+                "CBN：萧炎进入试炼场",
+                "CPNs：",
+                "- 药老提醒规则异常",
+                "- 萧炎发现师徒分歧",
+                "CEN：萧炎决定暂缓冲突",
+                "必须覆盖节点：发现规则异常",
+                "本章禁区：不可提前摊牌",
+            ]
+        ),
+        encoding="utf-8",
+    )
+
+    refs_dir = project_root / ".claude" / "references"
+    refs_dir.mkdir(parents=True, exist_ok=True)
+    (refs_dir / "genre-profiles.md").write_text("## xuanhuan\n- 升级线清晰", encoding="utf-8")
+    (refs_dir / "reading-power-taxonomy.md").write_text("## xuanhuan\n- 冲突钩优先", encoding="utf-8")
+
+    calls = {"embed": 0, "embed_batch": 0, "rerank": 0}
+
+    class _StubVectorClient:
+        async def embed(self, texts):
+            calls["embed"] += 1
+            return [[1.0, 0.0] for _ in texts]
+
+        async def embed_batch(self, texts, skip_failures=True):
+            calls["embed_batch"] += 1
+            return [[1.0, 0.0] for _ in texts]
+
+        async def rerank(self, query, documents, top_n=None):
+            calls["rerank"] += 1
+            limit = top_n or len(documents)
+            return [
+                {"index": i, "relevance_score": 1.0 / (i + 1)}
+                for i in range(min(limit, len(documents)))
+            ]
+
+    monkeypatch.setenv("EMBED_API_KEY", "fake-embed-key")
+    monkeypatch.setattr(rag_module, "get_client", lambda config: _StubVectorClient())
+
+    adapter = rag_module.RAGAdapter(cfg)
+    asyncio.run(
+        adapter.store_chunks(
+            [
+                {
+                    "chapter": 2,
+                    "scene_index": 1,
+                    "content": "萧炎与药老关系紧张，线索逐步浮现，冲突升级。",
+                }
+            ]
+        )
+    )
+
+    script_to_module = {
+        "story_system.py": "story_system",
+        "extract_chapter_context.py": "extract_chapter_context",
+        "review_pipeline.py": "review_pipeline",
+    }
+
+    def _run_script_inproc(script_name, argv):
+        module_name = script_to_module.get(script_name)
+        if not module_name:
+            raise AssertionError(f"unexpected script call: {script_name}")
+        script_module = importlib.import_module(module_name)
+        old_argv = sys.argv
+        try:
+            sys.argv = [module_name, *argv]
+            script_module.main()
+            return 0
+        except SystemExit as exc:
+            return int(exc.code or 0)
+        finally:
+            sys.argv = old_argv
+
+    monkeypatch.setattr(module, "_run_script", _run_script_inproc)
+
+    def _run_webnovel(argv):
+        monkeypatch.setattr(sys, "argv", ["webnovel", *argv])
+        with pytest.raises(SystemExit) as exc:
+            module.main()
+        return int(exc.value.code or 0)
+
+    assert (
+        _run_webnovel(
+            [
+                "--project-root",
+                str(project_root),
+                "story-system",
+                "玄幻退婚流",
+                "--chapter",
+                "3",
+                "--persist",
+                "--emit-runtime-contracts",
+                "--format",
+                "json",
+            ]
+        )
+        == 0
+    )
+    capsys.readouterr()
+
+    story_root = project_root / ".story-system"
+    assert (story_root / "MASTER_SETTING.json").is_file()
+    assert (story_root / "volumes" / "volume_001.json").is_file()
+    assert (story_root / "reviews" / "chapter_003.review.json").is_file()
+
+    assert (
+        _run_webnovel(
+            [
+                "--project-root",
+                str(project_root),
+                "extract-context",
+                "--chapter",
+                "3",
+                "--format",
+                "json",
+            ]
+        )
+        == 0
+    )
+    context_payload = json.loads(capsys.readouterr().out)
+    assert (
+        context_payload["story_contract"]["review_contract"]["meta"]["contract_type"]
+        == "REVIEW_CONTRACT"
+    )
+    assert context_payload["prewrite_validation"]["blocking"] is False
+    assert context_payload["rag_assist"]["invoked"] is True
+    assert context_payload["rag_assist"]["hits"]
+    assert calls["embed_batch"] >= 1
+    assert calls["embed"] >= 1
+    assert calls["rerank"] >= 1
+
+    review_results_path = project_root / ".webnovel" / "tmp" / "review_results.json"
+    review_results_path.parent.mkdir(parents=True, exist_ok=True)
+    review_results_path.write_text(
+        json.dumps(
+            {
+                "issues": [
+                    {
+                        "severity": "medium",
+                        "category": "continuity",
+                        "location": "第3段",
+                        "description": "衔接略弱",
+                        "evidence": "上章钩子未明确承接",
+                        "fix_hint": "补衔接句",
+                    }
+                ],
+                "summary": "1个中优问题",
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+    metrics_out = project_root / ".webnovel" / "tmp" / "review_metrics.json"
+    assert (
+        _run_webnovel(
+            [
+                "--project-root",
+                str(project_root),
+                "review-pipeline",
+                "--chapter",
+                "3",
+                "--review-results",
+                str(review_results_path),
+                "--metrics-out",
+                str(metrics_out),
+                "--report-file",
+                "审查报告/第3章.md",
+            ]
+        )
+        == 0
+    )
+    assert metrics_out.is_file()
+    metrics_payload = json.loads(metrics_out.read_text(encoding="utf-8"))
+    assert metrics_payload["issues_count"] == 1