feat: implement story system phases 1-4

README.md

@@ -81,6 +81,23 @@ RERANK_API_KEY=your_rerank_api_key
 python -X utf8 "<CLAUDE_PLUGIN_ROOT>/scripts/webnovel.py" --project-root "<WORKSPACE_ROOT>" preflight
 ```
 
+### Story System Phase 1-4
+
+当前已经补齐合同种子、合同优先运行时、章节提交主链、统一事件主链四段：
+
+- `.story-system/MASTER_SETTING.json` / `chapters/` / `volumes/` / `reviews/`
+- `.story-system/commits/chapter_XXX.commit.json`
+- `.story-system/events/chapter_XXX.events.json`
+
+常用统一 CLI：
+
+```bash
+python -X utf8 "<CLAUDE_PLUGIN_ROOT>/scripts/webnovel.py" --project-root "<PROJECT_ROOT>" story-system "玄幻退婚流" --persist
+python -X utf8 "<CLAUDE_PLUGIN_ROOT>/scripts/webnovel.py" --project-root "<PROJECT_ROOT>" story-system "玄幻退婚流" --emit-runtime-contracts --chapter 12
+python -X utf8 "<CLAUDE_PLUGIN_ROOT>/scripts/webnovel.py" --project-root "<PROJECT_ROOT>" chapter-commit --chapter 12 --review-result .webnovel/tmp/review.json --fulfillment-result .webnovel/tmp/fulfillment.json --disambiguation-result .webnovel/tmp/disambiguation.json --extraction-result .webnovel/tmp/extraction.json
+python -X utf8 "<CLAUDE_PLUGIN_ROOT>/scripts/webnovel.py" --project-root "<PROJECT_ROOT>" story-events --health
+```
+
 ### 6) 启动可视化面板（可选）
 
 ```bash

docs/architecture/overview.md

@@ -58,3 +58,28 @@
 | OOC Checker | 人物行为是否偏离人设 |
 | Continuity Checker | 场景与叙事连贯性 |
 | Reader-pull Checker | 钩子强度、期待管理、追读力 |
+
+## Story System Phase 1-4
+
+Story System 现在以 `.story-system/` 为独立运行面，分四段递进：
+
+1. Phase 1：`MASTER_SETTING / chapter brief / anti_patterns`
+2. Phase 2：`VOLUME_BRIEF / REVIEW_CONTRACT / prewrite validation`
+3. Phase 3：`CHAPTER_COMMIT + state/index/summary/memory` 投影
+4. Phase 4：`story_events + amend proposal + override ledger`
+
+核心链路：
+
+```text
+story-system --persist
+    -> 合同种子
+story-system --emit-runtime-contracts --chapter N
+    -> runtime contracts + prewrite validation
+chapter-commit --chapter N
+    -> commit + projections
+story-events --chapter N / --health
+    -> event 审计与健康检查
+```
+
+其中 Phase 4 不再起第二套投影循环，事件路由仅负责声明式激活 writer，
+实际执行入口仍是 `ChapterCommitService.apply_projections()`。

docs/architecture/story-system-phase4.md

@@ -0,0 +1,72 @@
+# Story System Phase 4
+
+## 目标
+
+Phase 4 把 Phase 3 的 `accepted_events` 升级为正式事件主链，并把
+`override_contracts` 扩成统一 override ledger。
+
+本阶段新增两条稳定链路：
+
+1. `CHAPTER_COMMIT.accepted_events -> .story-system/events/*.events.json`
+2. `world_rule_broken -> amend_proposal -> override_contracts`
+
+## 产物
+
+运行后会出现这些核心文件或表：
+
+- `.story-system/events/chapter_XXX.events.json`
+- `.webnovel/index.db.story_events`
+- `.webnovel/index.db.override_contracts.record_type=*`
+
+`story_events` 是 canonical 审计镜像，`override_contracts` 继续保留旧
+Override Contract / 债务链，同时新增：
+
+- `soft_deviation`
+- `contract_override`
+- `amend_proposal`
+
+## 执行关系
+
+```text
+review / fulfillment / disambiguation / extraction
+                │
+                ▼
+         CHAPTER_COMMIT.accepted
+                │
+                ├── apply_projections()
+                │     ├── state/index/summary/memory
+                │     └── EventProjectionRouter 决定激活哪些 writer
+                │
+                ├── EventLogStore.write_events()
+                │     ├── JSON 文件
+                │     └── SQLite story_events 镜像
+                │
+                └── AmendProposalTrigger.check()
+                      └── persist_amend_proposals() -> override_contracts
+```
+
+## CLI
+
+统一入口仍然是 `webnovel.py`：
+
+```bash
+python -X utf8 "<CLAUDE_PLUGIN_ROOT>/scripts/webnovel.py" --project-root "<PROJECT_ROOT>" story-system "玄幻退婚流" --persist
+python -X utf8 "<CLAUDE_PLUGIN_ROOT>/scripts/webnovel.py" --project-root "<PROJECT_ROOT>" story-system "玄幻退婚流" --emit-runtime-contracts --chapter 12
+python -X utf8 "<CLAUDE_PLUGIN_ROOT>/scripts/webnovel.py" --project-root "<PROJECT_ROOT>" chapter-commit --chapter 12 --review-result .webnovel/tmp/review.json --fulfillment-result .webnovel/tmp/fulfillment.json --disambiguation-result .webnovel/tmp/disambiguation.json --extraction-result .webnovel/tmp/extraction.json
+python -X utf8 "<CLAUDE_PLUGIN_ROOT>/scripts/webnovel.py" --project-root "<PROJECT_ROOT>" story-events --chapter 12
+python -X utf8 "<CLAUDE_PLUGIN_ROOT>/scripts/webnovel.py" --project-root "<PROJECT_ROOT>" story-events --health
+```
+
+## 最小运维检查
+
+- `story-system --persist` 后应存在 `MASTER_SETTING.json`
+- `story-system --emit-runtime-contracts --chapter N` 后应存在 `volume_XXX.json`
+  与 `chapter_XXX.review.json`
+- `chapter-commit` accepted 后应存在 `chapter_XXX.commit.json`
+- `story-events --health` 应返回 `sqlite_rows` 与 `event_files`
+
+## 当前边界
+
+- 事件路由仍由 `ChapterCommitService.apply_projections()` 统一调度
+- Phase 4 不新增第二套独立投影循环
+- 旧链路降级与完全切换留到 Phase 5

docs/guides/commands.md

@@ -123,3 +123,24 @@ python -X utf8 "<CLAUDE_PLUGIN_ROOT>/scripts/webnovel.py" --project-root "<PROJE
 - `memory conflicts`：查看同主键 active 冲突项
 - `memory bootstrap`：从 `index.db` 与 `summaries` 回填初始长期记忆
 - `memory update`：对指定章节结果执行一次手动映射写入
+
+## Story System 统一 CLI
+
+用途：管理合同、提交链与事件审计。
+
+示例：
+
+```bash
+python -X utf8 "<CLAUDE_PLUGIN_ROOT>/scripts/webnovel.py" --project-root "<PROJECT_ROOT>" story-system "玄幻退婚流" --persist
+python -X utf8 "<CLAUDE_PLUGIN_ROOT>/scripts/webnovel.py" --project-root "<PROJECT_ROOT>" story-system "玄幻退婚流" --emit-runtime-contracts --chapter 12
+python -X utf8 "<CLAUDE_PLUGIN_ROOT>/scripts/webnovel.py" --project-root "<PROJECT_ROOT>" chapter-commit --chapter 12 --review-result .webnovel/tmp/review.json --fulfillment-result .webnovel/tmp/fulfillment.json --disambiguation-result .webnovel/tmp/disambiguation.json --extraction-result .webnovel/tmp/extraction.json
+python -X utf8 "<CLAUDE_PLUGIN_ROOT>/scripts/webnovel.py" --project-root "<PROJECT_ROOT>" story-events --chapter 12
+python -X utf8 "<CLAUDE_PLUGIN_ROOT>/scripts/webnovel.py" --project-root "<PROJECT_ROOT>" story-events --health
+```
+
+产物：
+
+- `story-system --persist`：写入 `.story-system/MASTER_SETTING.json`
+- `--emit-runtime-contracts`：写入 `volumes/*.json` 与 `reviews/*.review.json`
+- `chapter-commit`：写入 `commits/*.commit.json`
+- `story-events`：读取 `events/*.events.json` 或 `index.db.story_events`

docs/operations/operations.md

@@ -97,3 +97,45 @@ python "${SCRIPTS_DIR}/webnovel.py" --project-root "${PROJECT_ROOT}" rag stats
 pwsh "${CLAUDE_PLUGIN_ROOT}/scripts/run_tests.ps1" -Mode smoke
 pwsh "${CLAUDE_PLUGIN_ROOT}/scripts/run_tests.ps1" -Mode full
 ```
+
+## Story System 运维
+
+### preflight
+
+检查统一入口与事件链目录是否可用：
+
+```bash
+python -X utf8 "${CLAUDE_PLUGIN_ROOT}/scripts/webnovel.py" --project-root "${PROJECT_ROOT}" preflight
+python -X utf8 "${CLAUDE_PLUGIN_ROOT}/scripts/webnovel.py" --project-root "${PROJECT_ROOT}" story-events --health
+```
+
+重点看三项：
+
+- `.story-system/events/` 是否可读
+- `.webnovel/index.db` 中 `story_events` 是否可查
+- `override_contracts` 是否能统计 `amend_proposal`
+
+### health
+
+最小健康检查命令：
+
+```bash
+python -X utf8 "${CLAUDE_PLUGIN_ROOT}/scripts/webnovel.py" --project-root "${PROJECT_ROOT}" story-events --health
+```
+
+返回字段：
+
+- `sqlite_rows`
+- `event_files`
+- `ok`
+
+### backup
+
+做 Story System 相关备份时，至少同时备这两块：
+
+```bash
+.story-system/
+.webnovel/index.db
+```
+
+如果要做章节级回溯，建议连同 `.webnovel/summaries/` 一起备份。

docs/superpowers/README.md

@@ -17,6 +17,7 @@
 - [`plans/2026-04-12-story-system-phase2-contract-first-runtime.md`](./plans/2026-04-12-story-system-phase2-contract-first-runtime.md)：Story System Phase 2 合同优先运行时实施计划
 - [`plans/2026-04-12-story-system-phase3-chapter-commit-chain.md`](./plans/2026-04-12-story-system-phase3-chapter-commit-chain.md)：Story System Phase 3 章节提交主链实施计划
 - [`plans/2026-04-12-story-system-phase4-event-log-and-override-ledger.md`](./plans/2026-04-12-story-system-phase4-event-log-and-override-ledger.md)：Story System Phase 4 统一事件主链与 Override Ledger 实施计划
+- [`../architecture/story-system-phase4.md`](../architecture/story-system-phase4.md)：Phase 4 落地后的事件主链与 override ledger 运行说明
 
 ## 使用约定
 

webnovel-writer/dashboard/app.py

@@ -38,6 +38,10 @@ def _webnovel_dir() -> Path:
     return _get_project_root() / ".webnovel"
 
 
+def _story_system_dir() -> Path:
+    return _get_project_root() / ".story-system"
+
+
 # ---------------------------------------------------------------------------
 # 应用工厂
 # ---------------------------------------------------------------------------
@@ -97,7 +101,7 @@ def create_app(project_root: str | Path | None = None) -> FastAPI:
             rows = conn.execute(query, params).fetchall()
             return [dict(r) for r in rows]
         except sqlite3.OperationalError as exc:
-            if "no such table" in str(exc).lower():
+            if "no such table" in str(exc).lower() or "no such column" in str(exc).lower():
                 return []
             raise HTTPException(status_code=500, detail=f"数据库查询失败: {exc}") from exc
 
@@ -335,6 +339,64 @@ def create_app(project_root: str | Path | None = None) -> FastAPI:
                 (limit,),
             )
 
+    @app.get("/api/story-events")
+    def list_story_events(chapter: Optional[int] = None, limit: int = 200):
+        with closing(_get_db()) as conn:
+            if chapter is not None:
+                rows = _fetchall_safe(
+                    conn,
+                    """
+                    SELECT event_id, chapter, event_type, subject, payload_json, created_at
+                    FROM story_events
+                    WHERE chapter = ?
+                    ORDER BY id DESC
+                    LIMIT ?
+                    """,
+                    (chapter, limit),
+                )
+            else:
+                rows = _fetchall_safe(
+                    conn,
+                    """
+                    SELECT event_id, chapter, event_type, subject, payload_json, created_at
+                    FROM story_events
+                    ORDER BY chapter DESC, id DESC
+                    LIMIT ?
+                    """,
+                    (limit,),
+                )
+
+        normalized = []
+        for row in rows:
+            payload = {}
+            try:
+                payload = json.loads(row.get("payload_json") or "{}")
+            except json.JSONDecodeError:
+                payload = {}
+            normalized.append({**row, "payload": payload})
+        return normalized
+
+    @app.get("/api/story-events/health")
+    def story_event_health():
+        with closing(_get_db()) as conn:
+            event_rows = _fetchall_safe(conn, "SELECT COUNT(*) AS count FROM story_events")
+            proposal_rows = _fetchall_safe(
+                conn,
+                """
+                SELECT COUNT(*) AS count
+                FROM override_contracts
+                WHERE record_type = 'amend_proposal' AND status = 'pending'
+                """,
+            )
+
+        events_dir = _story_system_dir() / "events"
+        file_count = len(list(events_dir.glob("chapter_*.events.json"))) if events_dir.is_dir() else 0
+        return {
+            "story_events": event_rows[0]["count"] if event_rows else 0,
+            "pending_amend_proposals": proposal_rows[0]["count"] if proposal_rows else 0,
+            "event_files": file_count,
+        }
+
     # ===========================================================
     # API：文档浏览（正文/大纲/设定集 —— 只读）
     # ===========================================================

webnovel-writer/scripts/chapter_commit.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+from runtime_compat import enable_windows_utf8_stdio
+
+from data_modules.chapter_commit_service import ChapterCommitService
+
+
+def _read_json(path: str) -> dict:
+    return json.loads(Path(path).read_text(encoding="utf-8"))
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Chapter commit CLI")
+    parser.add_argument("--project-root", required=True)
+    parser.add_argument("--chapter", type=int, required=True)
+    parser.add_argument("--review-result", required=True)
+    parser.add_argument("--fulfillment-result", required=True)
+    parser.add_argument("--disambiguation-result", required=True)
+    parser.add_argument("--extraction-result", required=True)
+    args = parser.parse_args()
+
+    service = ChapterCommitService(Path(args.project_root))
+    payload = service.build_commit(
+        chapter=args.chapter,
+        review_result=_read_json(args.review_result),
+        fulfillment_result=_read_json(args.fulfillment_result),
+        disambiguation_result=_read_json(args.disambiguation_result),
+        extraction_result=_read_json(args.extraction_result),
+    )
+    service.persist_commit(payload)
+    if payload["meta"]["status"] == "accepted":
+        payload = service.apply_projections(payload)
+    print(json.dumps(payload, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    if sys.platform == "win32":
+        enable_windows_utf8_stdio()
+    main()

webnovel-writer/scripts/data_modules/amend_proposal_schema.py

@@ -0,0 +1,15 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class AmendProposal(BaseModel):
+    proposal_id: str
+    chapter: int = Field(ge=1)
+    target_level: str
+    field: str
+    base_value: str = ""
+    proposed_value: str = ""
+    reason_tag: str

webnovel-writer/scripts/data_modules/chapter_commit_service.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any, Dict
+
+from .config import DataModulesConfig
+from .event_log_store import EventLogStore
+from .event_projection_router import EventProjectionRouter
+from .index_manager import IndexManager
+from .override_ledger_service import (
+    AmendProposalTrigger,
+    ensure_override_ledger_columns,
+    persist_amend_proposals,
+)
+
+
+class ChapterCommitService:
+    def __init__(self, project_root: Path):
+        self.project_root = Path(project_root)
+
+    def build_commit(
+        self,
+        chapter: int,
+        review_result: Dict[str, Any],
+        fulfillment_result: Dict[str, Any],
+        disambiguation_result: Dict[str, Any],
+        extraction_result: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        rejected = bool(review_result.get("blocking_count")) or bool(
+            fulfillment_result.get("missed_nodes")
+        ) or bool(disambiguation_result.get("pending"))
+        status = "rejected" if rejected else "accepted"
+        return {
+            "meta": {
+                "schema_version": "story-system/v1",
+                "chapter": chapter,
+                "status": status,
+            },
+            "contract_refs": {
+                "master": "MASTER_SETTING.json",
+                "chapter": f"chapter_{chapter:03d}.json",
+                "review": f"chapter_{chapter:03d}.review.json",
+            },
+            "outline_snapshot": {
+                "planned_nodes": fulfillment_result.get("planned_nodes", []),
+                "covered_nodes": fulfillment_result.get("covered_nodes", []),
+                "missed_nodes": fulfillment_result.get("missed_nodes", []),
+                "extra_nodes": fulfillment_result.get("extra_nodes", []),
+            },
+            "review_result": review_result,
+            "fulfillment_result": fulfillment_result,
+            "disambiguation_result": disambiguation_result,
+            "accepted_events": extraction_result.get("accepted_events", []),
+            "state_deltas": extraction_result.get("state_deltas", []),
+            "entity_deltas": extraction_result.get("entity_deltas", []),
+            "summary_text": extraction_result.get("summary_text", ""),
+            "projection_status": {
+                "state": "pending",
+                "index": "pending",
+                "summary": "pending",
+                "memory": "pending",
+            },
+        }
+
+    def persist_commit(self, payload: Dict[str, Any]) -> Path:
+        target = self.project_root / ".story-system" / "commits"
+        target.mkdir(parents=True, exist_ok=True)
+        path = target / f"chapter_{int(payload['meta']['chapter']):03d}.commit.json"
+        path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+        return path
+
+    def apply_projections(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        if payload["meta"]["status"] != "accepted":
+            return payload
+
+        chapter = int((payload.get("meta") or {}).get("chapter") or 0)
+        EventLogStore(self.project_root).write_events(chapter, payload.get("accepted_events", []))
+
+        proposals = AmendProposalTrigger().check(chapter, payload.get("accepted_events", []))
+        if proposals:
+            manager = IndexManager(DataModulesConfig.from_project_root(self.project_root))
+            with manager._get_conn() as conn:
+                ensure_override_ledger_columns(conn)
+                persist_amend_proposals(conn, chapter, proposals)
+                conn.commit()
+
+        from .index_projection_writer import IndexProjectionWriter
+        from .memory_projection_writer import MemoryProjectionWriter
+        from .state_projection_writer import StateProjectionWriter
+        from .summary_projection_writer import SummaryProjectionWriter
+
+        writers = {
+            "state": StateProjectionWriter(self.project_root),
+            "index": IndexProjectionWriter(self.project_root),
+            "summary": SummaryProjectionWriter(self.project_root),
+            "memory": MemoryProjectionWriter(self.project_root),
+        }
+        required_writers = set(EventProjectionRouter().required_writers(payload))
+        for name, writer in writers.items():
+            if name not in required_writers:
+                payload["projection_status"][name] = "skipped"
+                continue
+            try:
+                result = writer.apply(payload)
+                payload["projection_status"][name] = "done" if result.get("applied") else "skipped"
+            except Exception as exc:
+                payload["projection_status"][name] = f"failed:{exc}"
+        self.persist_commit(payload)
+        return payload

webnovel-writer/scripts/data_modules/config.py

@@ -124,6 +124,22 @@ class DataModulesConfig:
     def outline_dir(self) -> Path:
         return self.project_root / "大纲"
 
+    @property
+    def story_system_dir(self) -> Path:
+        return self.project_root / ".story-system"
+
+    @property
+    def story_system_chapters_dir(self) -> Path:
+        return self.story_system_dir / "chapters"
+
+    @property
+    def story_system_master_json(self) -> Path:
+        return self.story_system_dir / "MASTER_SETTING.json"
+
+    @property
+    def story_system_anti_patterns_json(self) -> Path:
+        return self.story_system_dir / "anti_patterns.json"
+
 
     # ================= Embedding API 配置 =================
     embed_api_type: str = "openai"

webnovel-writer/scripts/data_modules/context_manager.py

@@ -15,14 +15,24 @@ from runtime_compat import enable_windows_utf8_stdio
 from typing import Any, Dict, List, Optional
 
 try:
-    from chapter_outline_loader import load_chapter_outline, load_chapter_plot_structure
+    from chapter_outline_loader import (
+        load_chapter_outline,
+        load_chapter_plot_structure,
+        volume_num_for_chapter_from_state,
+    )
 except ImportError:  # pragma: no cover
-    from scripts.chapter_outline_loader import load_chapter_outline, load_chapter_plot_structure
+    from scripts.chapter_outline_loader import (
+        load_chapter_outline,
+        load_chapter_plot_structure,
+        volume_num_for_chapter_from_state,
+    )
 
 from .config import get_config
 from .index_manager import IndexManager, WritingChecklistScoreMeta
 from .context_ranker import ContextRanker
+from .prewrite_validator import PrewriteValidator
 from .snapshot_manager import SnapshotManager, SnapshotVersionMismatch
+from .story_contracts import read_json_if_exists
 from .context_weights import (
     DEFAULT_TEMPLATE as CONTEXT_DEFAULT_TEMPLATE,
     TEMPLATE_WEIGHTS as CONTEXT_TEMPLATE_WEIGHTS,
@@ -61,9 +71,13 @@ class ContextManager:
         "genre_profile",
         "writing_guidance",
         "plot_structure",
+        "story_contract",
+        "prewrite_validation",
     }
     SECTION_ORDER = [
         "core",
+        "story_contract",
+        "prewrite_validation",
         "scene",
         "global",
         "reader_signal",
@@ -108,7 +122,7 @@ class ContextManager:
         if not isinstance(sections, dict):
             return False
 
-        required_sections = {"plot_structure", "long_term_memory"}
+        required_sections = {"plot_structure", "long_term_memory", "story_contract", "prewrite_validation"}
         return required_sections.issubset(set(sections.keys()))
 
     def build_context(
@@ -253,6 +267,7 @@ class ContextManager:
         scene["appearing_characters"] = self.filter_invalid_items(
             scene["appearing_characters"], source_type="entity", id_key="entity_id"
         )
+        story_contract = self._load_story_contract(chapter)
 
         global_ctx = {
             "worldview_skeleton": self._load_setting("世界观"),
@@ -269,10 +284,17 @@ class ContextManager:
         genre_profile = self._load_genre_profile(state)
         writing_guidance = self._build_writing_guidance(chapter, reader_signal, genre_profile)
         plot_structure = self._load_plot_structure(chapter)
+        prewrite_validation = PrewriteValidator(self.config.project_root).build(
+            chapter=chapter,
+            review_contract=story_contract.get("review_contract") or {},
+            plot_structure=plot_structure,
+        )
 
         return {
             "meta": {"chapter": chapter},
             "core": core,
+            "story_contract": story_contract,
+            "prewrite_validation": prewrite_validation,
             "scene": scene,
             "global": global_ctx,
             "reader_signal": reader_signal,
@@ -693,6 +715,23 @@ class ContextManager:
     def _load_plot_structure(self, chapter: int) -> Dict[str, Any]:
         return load_chapter_plot_structure(self.config.project_root, chapter)
 
+    def _load_story_contract(self, chapter: int) -> Dict[str, Any]:
+        story_root = self.config.story_system_dir
+        volume = volume_num_for_chapter_from_state(self.config.project_root, chapter) or 1
+        return {
+            "master_setting": read_json_if_exists(story_root / "MASTER_SETTING.json") or {},
+            "chapter_brief": read_json_if_exists(
+                story_root / "chapters" / f"chapter_{chapter:03d}.json"
+            ) or {},
+            "volume_brief": read_json_if_exists(
+                story_root / "volumes" / f"volume_{volume:03d}.json"
+            ) or {},
+            "review_contract": read_json_if_exists(
+                story_root / "reviews" / f"chapter_{chapter:03d}.review.json"
+            ) or {},
+            "anti_patterns": read_json_if_exists(story_root / "anti_patterns.json") or [],
+        }
+
     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_log_store.py

@@ -0,0 +1,149 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import json
+import sqlite3
+from pathlib import Path
+from typing import Any, Dict, List
+
+from .story_contracts import StoryContractPaths, read_json_if_exists, write_json
+from .story_event_schema import StoryEvent
+
+
+class EventLogStore:
+    def __init__(self, project_root: Path):
+        self.project_root = Path(project_root).expanduser().resolve()
+        self.paths = StoryContractPaths.from_project_root(self.project_root)
+
+    def write_events(self, chapter: int, events: List[dict]) -> Path:
+        normalized = self._normalize_events(chapter, events)
+        path = self.paths.event_json(chapter)
+        write_json(path, normalized)
+        self._write_sqlite_mirror(normalized)
+        return path
+
+    def read_events(self, chapter: int) -> List[Dict[str, Any]]:
+        return list(read_json_if_exists(self.paths.event_json(chapter)) or [])
+
+    def list_recent(self, chapter: int | None = None, limit: int = 200) -> List[Dict[str, Any]]:
+        db_path = self.project_root / ".webnovel" / "index.db"
+        if not db_path.is_file():
+            return []
+        conn = sqlite3.connect(str(db_path))
+        conn.row_factory = sqlite3.Row
+        try:
+            if chapter is not None:
+                rows = conn.execute(
+                    """
+                    SELECT event_id, chapter, event_type, subject, payload_json
+                    FROM story_events
+                    WHERE chapter = ?
+                    ORDER BY id DESC
+                    LIMIT ?
+                    """,
+                    (chapter, limit),
+                ).fetchall()
+            else:
+                rows = conn.execute(
+                    """
+                    SELECT event_id, chapter, event_type, subject, payload_json
+                    FROM story_events
+                    ORDER BY chapter DESC, id DESC
+                    LIMIT ?
+                    """,
+                    (limit,),
+                ).fetchall()
+        except sqlite3.OperationalError:
+            return []
+        finally:
+            conn.close()
+
+        result: List[Dict[str, Any]] = []
+        for row in rows:
+            payload = {}
+            try:
+                payload = json.loads(row["payload_json"] or "{}")
+            except json.JSONDecodeError:
+                payload = {}
+            result.append(
+                {
+                    "event_id": row["event_id"],
+                    "chapter": row["chapter"],
+                    "event_type": row["event_type"],
+                    "subject": row["subject"],
+                    "payload": payload,
+                }
+            )
+        return result
+
+    def health(self) -> Dict[str, Any]:
+        db_path = self.project_root / ".webnovel" / "index.db"
+        file_count = len(list(self.paths.events_dir.glob("chapter_*.events.json")))
+        sqlite_rows = 0
+        if db_path.is_file():
+            conn = sqlite3.connect(str(db_path))
+            try:
+                try:
+                    sqlite_rows = int(
+                        conn.execute("SELECT COUNT(*) FROM story_events").fetchone()[0]
+                    )
+                except sqlite3.OperationalError:
+                    sqlite_rows = 0
+            finally:
+                conn.close()
+        return {"ok": True, "sqlite_rows": sqlite_rows, "event_files": file_count}
+
+    def _normalize_events(self, chapter: int, events: List[dict]) -> List[Dict[str, Any]]:
+        normalized: List[Dict[str, Any]] = []
+        for event in events or []:
+            if not isinstance(event, dict):
+                continue
+            payload = dict(event)
+            payload["chapter"] = int(payload.get("chapter") or chapter)
+            normalized.append(StoryEvent.model_validate(payload).model_dump())
+        return normalized
+
+    def _write_sqlite_mirror(self, events: List[Dict[str, Any]]) -> None:
+        db_path = self.project_root / ".webnovel" / "index.db"
+        db_path.parent.mkdir(parents=True, exist_ok=True)
+        conn = sqlite3.connect(str(db_path))
+        try:
+            conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS story_events (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    event_id TEXT NOT NULL UNIQUE,
+                    chapter INTEGER NOT NULL,
+                    event_type TEXT NOT NULL,
+                    subject TEXT NOT NULL,
+                    payload_json TEXT NOT NULL,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                )
+                """
+            )
+            conn.execute(
+                "CREATE INDEX IF NOT EXISTS idx_story_events_chapter ON story_events(chapter)"
+            )
+            conn.execute(
+                "CREATE INDEX IF NOT EXISTS idx_story_events_type ON story_events(event_type)"
+            )
+            conn.executemany(
+                """
+                INSERT OR IGNORE INTO story_events(event_id, chapter, event_type, subject, payload_json)
+                VALUES (?, ?, ?, ?, ?)
+                """,
+                [
+                    (
+                        event["event_id"],
+                        int(event["chapter"]),
+                        event["event_type"],
+                        event["subject"],
+                        json.dumps(event.get("payload") or {}, ensure_ascii=False),
+                    )
+                    for event in events
+                ],
+            )
+            conn.commit()
+        finally:
+            conn.close()

webnovel-writer/scripts/data_modules/event_projection_router.py

@@ -0,0 +1,37 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+from typing import Dict, List, Set
+
+
+class EventProjectionRouter:
+    TABLE = {
+        "character_state_changed": ["state", "memory"],
+        "power_breakthrough": ["state", "memory"],
+        "relationship_changed": ["index"],
+        "world_rule_revealed": ["index", "memory"],
+        "world_rule_broken": ["index", "memory"],
+        "open_loop_created": ["memory"],
+        "open_loop_closed": ["memory"],
+        "promise_created": ["memory"],
+        "promise_paid_off": ["memory"],
+        "artifact_obtained": ["index", "memory"],
+    }
+
+    def route(self, event: Dict) -> List[str]:
+        return list(self.TABLE.get(str(event.get("event_type") or "").strip(), []))
+
+    def required_writers(self, commit_payload: Dict) -> List[str]:
+        writers: Set[str] = set()
+        if str((commit_payload.get("meta") or {}).get("status") or "") == "accepted":
+            writers.add("state")
+        if commit_payload.get("entity_deltas"):
+            writers.add("index")
+        if str(commit_payload.get("summary_text") or "").strip():
+            writers.add("summary")
+        for event in commit_payload.get("accepted_events") or []:
+            if not isinstance(event, dict):
+                continue
+            writers.update(self.route(event))
+        return sorted(writers)

webnovel-writer/scripts/data_modules/index_manager.py

@@ -50,6 +50,7 @@ from .index_debt_mixin import IndexDebtMixin
 from .index_reading_mixin import IndexReadingMixin
 from .index_observability_mixin import IndexObservabilityMixin
 from .observability import safe_append_perf_timing, safe_log_tool_call
+from .override_ledger_service import ensure_override_ledger_columns
 
 
 @dataclass
@@ -494,6 +495,7 @@ class IndexManager(IndexChapterMixin, IndexEntityMixin, IndexDebtMixin, IndexRea
             cursor.execute(
                 "CREATE INDEX IF NOT EXISTS idx_override_contracts_due ON override_contracts(due_chapter)"
             )
+            ensure_override_ledger_columns(conn)
             cursor.execute(
                 "CREATE INDEX IF NOT EXISTS idx_chase_debt_status ON chase_debt(status)"
             )
@@ -631,6 +633,52 @@ class IndexManager(IndexChapterMixin, IndexEntityMixin, IndexDebtMixin, IndexRea
         finally:
             conn.close()
 
+    def apply_entity_delta(self, delta: Dict[str, Any]) -> bool:
+        """将 commit/entity 提取产物映射为实体或关系索引更新。"""
+        if not isinstance(delta, dict):
+            return False
+
+        from_entity = str(delta.get("from_entity") or delta.get("from") or "").strip()
+        to_entity = str(delta.get("to_entity") or delta.get("to") or "").strip()
+        rel_type = str(delta.get("relation_type") or delta.get("relationship_type") or delta.get("type") or "").strip()
+        chapter = int(delta.get("chapter") or 0)
+        if from_entity and to_entity and rel_type:
+            self.upsert_relationship(
+                RelationshipMeta(
+                    from_entity=from_entity,
+                    to_entity=to_entity,
+                    type=rel_type,
+                    description=str(delta.get("description") or "").strip(),
+                    chapter=chapter,
+                )
+            )
+            return True
+
+        entity_id = str(delta.get("entity_id") or delta.get("id") or "").strip()
+        if not entity_id:
+            return False
+
+        current = dict(delta.get("current") or {})
+        field = str(delta.get("field") or "").strip()
+        if field and "new" in delta and field not in current:
+            current[field] = delta.get("new")
+
+        canonical_name = str(delta.get("canonical_name") or delta.get("name") or entity_id).strip()
+        entity = EntityMeta(
+            id=entity_id,
+            type=str(delta.get("type") or "角色").strip() or "角色",
+            canonical_name=canonical_name,
+            tier=str(delta.get("tier") or "装饰").strip() or "装饰",
+            desc=str(delta.get("desc") or "").strip(),
+            current=current,
+            first_appearance=chapter,
+            last_appearance=chapter,
+            is_protagonist=bool(delta.get("is_protagonist")),
+            is_archived=bool(delta.get("is_archived")),
+        )
+        self.upsert_entity(entity, update_metadata=True)
+        return True
+
     # ==================== 章节操作 ====================
 
 # ==================== CLI 接口 ====================

webnovel-writer/scripts/data_modules/index_projection_writer.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+from pathlib import Path
+
+from .config import DataModulesConfig
+from .index_manager import IndexManager
+
+
+class IndexProjectionWriter:
+    def __init__(self, project_root: Path):
+        self.project_root = Path(project_root)
+
+    def apply(self, commit_payload: dict) -> dict:
+        if commit_payload["meta"]["status"] != "accepted":
+            return {"applied": False, "writer": "index", "reason": "commit_rejected"}
+
+        manager = IndexManager(DataModulesConfig.from_project_root(self.project_root))
+        applied_count = 0
+        for delta in self._collect_entity_deltas(commit_payload):
+            result = manager.apply_entity_delta(delta)
+            if result:
+                applied_count += 1
+        return {
+            "applied": applied_count > 0,
+            "writer": "index",
+            "applied_count": applied_count,
+        }
+
+    def _collect_entity_deltas(self, commit_payload: dict) -> list[dict]:
+        deltas = [dict(delta) for delta in (commit_payload.get("entity_deltas") or []) if isinstance(delta, dict)]
+        for event in commit_payload.get("accepted_events") or []:
+            if not isinstance(event, dict):
+                continue
+            event_type = str(event.get("event_type") or "").strip()
+            payload = dict(event.get("payload") or {})
+            chapter = int(event.get("chapter") or commit_payload.get("meta", {}).get("chapter") or 0)
+            if event_type == "relationship_changed":
+                from_entity = str(payload.get("from_entity") or event.get("subject") or "").strip()
+                to_entity = str(payload.get("to_entity") or payload.get("to") or "").strip()
+                rel_type = str(
+                    payload.get("relationship_type")
+                    or payload.get("relation_type")
+                    or payload.get("type")
+                    or ""
+                ).strip()
+                if from_entity and to_entity and rel_type:
+                    deltas.append(
+                        {
+                            "from_entity": from_entity,
+                            "to_entity": to_entity,
+                            "relationship_type": rel_type,
+                            "description": str(payload.get("description") or "").strip(),
+                            "chapter": chapter,
+                        }
+                    )
+        return deltas

webnovel-writer/scripts/data_modules/memory/writer.py

@@ -161,7 +161,11 @@ class MemoryWriter:
             rule = str(row.get("rule", "") or "").strip()
             if not rule:
                 continue
-            subject = str(row.get("domain", "") or "").strip() or str(row.get("scope", "") or "").strip() or "global"
+            subject = (
+                str(row.get("domain", "") or "").strip()
+                or str(row.get("scope", "") or "").strip()
+                or "global"
+            )
             field = str(row.get("field", "") or "").strip() or rule[:32]
             item = MemoryItem(
                 id=self._item_id("world_rule", subject, field, chapter),
@@ -221,3 +225,81 @@ class MemoryWriter:
             )
             self._upsert(item, stats)
 
+    def apply_commit_projection(self, commit_payload: Dict[str, Any]) -> Dict[str, Any]:
+        chapter = int((commit_payload.get("meta") or {}).get("chapter") or 0)
+        entity_deltas = list(commit_payload.get("entity_deltas") or [])
+        accepted_events = list(commit_payload.get("accepted_events") or [])
+
+        memory_facts: Dict[str, Any] = {
+            "timeline_events": [],
+            "world_rules": [],
+            "open_loops": [],
+            "reader_promises": [],
+        }
+        for event in accepted_events:
+            if not isinstance(event, dict):
+                continue
+            event_type = str(event.get("event_type") or "").strip()
+            payload = event.get("payload") or {}
+            if event_type in {"world_rule_revealed", "world_rule_broken"}:
+                rule_text = str(payload.get("proposed_value") or payload.get("rule") or payload.get("base_value") or "").strip()
+                if rule_text:
+                    memory_facts["world_rules"].append(
+                        {
+                            "rule": rule_text,
+                            "scope": payload.get("scope") or "global",
+                            "domain": payload.get("domain") or event.get("subject") or "global",
+                            "field": payload.get("field") or event_type,
+                        }
+                    )
+            elif event_type == "open_loop_created":
+                content = str(payload.get("content") or event.get("subject") or "").strip()
+                if content:
+                    memory_facts["open_loops"].append(
+                        {
+                            "content": content,
+                            "status": payload.get("status") or "active",
+                            "urgency": payload.get("urgency") or 0,
+                        }
+                    )
+            elif event_type in {"promise_created", "promise_paid_off"}:
+                content = str(payload.get("content") or event.get("subject") or "").strip()
+                if content:
+                    memory_facts["reader_promises"].append(
+                        {
+                            "content": content,
+                            "type": payload.get("type") or event_type,
+                            "target": payload.get("target") or event.get("subject") or "",
+                        }
+                    )
+
+        result = {
+            "entities_new": [
+                {
+                    "suggested_id": row.get("entity_id") or row.get("id"),
+                    "name": row.get("canonical_name") or row.get("name") or row.get("entity_id") or row.get("id"),
+                    "type": row.get("type") or "角色",
+                    "tier": row.get("tier") or "装饰",
+                }
+                for row in entity_deltas
+                if isinstance(row, dict)
+                and str(row.get("entity_id") or row.get("id") or "").strip()
+                and not (row.get("from_entity") or row.get("from"))
+            ],
+            "state_changes": list(commit_payload.get("state_deltas") or []),
+            "relationships_new": [
+                {
+                    "from": row.get("from_entity") or row.get("from"),
+                    "to": row.get("to_entity") or row.get("to"),
+                    "type": row.get("relation_type") or row.get("relationship_type") or row.get("type"),
+                    "description": row.get("description") or "",
+                }
+                for row in entity_deltas
+                if isinstance(row, dict)
+                and str(row.get("from_entity") or row.get("from") or "").strip()
+                and str(row.get("to_entity") or row.get("to") or "").strip()
+            ],
+            "memory_facts": memory_facts,
+        }
+        return self.update_from_chapter_result(chapter, result)
+

webnovel-writer/scripts/data_modules/memory_projection_writer.py

@@ -0,0 +1,25 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+from pathlib import Path
+
+from .config import DataModulesConfig
+from .memory.writer import MemoryWriter
+
+
+class MemoryProjectionWriter:
+    def __init__(self, project_root: Path):
+        self.project_root = Path(project_root)
+
+    def apply(self, commit_payload: dict) -> dict:
+        if commit_payload["meta"]["status"] != "accepted":
+            return {"applied": False, "writer": "memory", "reason": "commit_rejected"}
+        result = MemoryWriter(DataModulesConfig.from_project_root(self.project_root)).apply_commit_projection(
+            commit_payload
+        )
+        return {
+            "applied": bool((result or {}).get("items_added") or (result or {}).get("items_updated")),
+            "writer": "memory",
+            **(result or {}),
+        }

webnovel-writer/scripts/data_modules/override_ledger_service.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import sqlite3
+from typing import Dict, List
+
+from .amend_proposal_schema import AmendProposal
+
+
+def normalize_override_record(
+    *,
+    record_type: str,
+    field: str,
+    base_value: str,
+    override_value: str,
+    source_level: str,
+) -> Dict[str, str]:
+    return {
+        "record_type": str(record_type or "").strip(),
+        "field": str(field or "").strip(),
+        "base_value": str(base_value or "").strip(),
+        "override_value": str(override_value or "").strip(),
+        "source_level": str(source_level or "").strip(),
+    }
+
+
+def ensure_override_ledger_columns(conn: sqlite3.Connection) -> None:
+    existing = {
+        row[1] for row in conn.execute("PRAGMA table_info(override_contracts)").fetchall()
+    }
+    wanted = {
+        "record_type": "TEXT DEFAULT 'soft_deviation'",
+        "field": "TEXT DEFAULT ''",
+        "base_value": "TEXT DEFAULT ''",
+        "override_value": "TEXT DEFAULT ''",
+        "source_level": "TEXT DEFAULT ''",
+        "reason_tag": "TEXT DEFAULT ''",
+    }
+    for name, ddl in wanted.items():
+        if name not in existing:
+            conn.execute(f"ALTER TABLE override_contracts ADD COLUMN {name} {ddl}")
+    conn.execute(
+        "CREATE INDEX IF NOT EXISTS idx_override_contracts_record_type ON override_contracts(record_type)"
+    )
+
+
+class AmendProposalTrigger:
+    RULES = {
+        "world_rule_broken": {"target_level": "master", "reason_tag": "world_rule_broken"},
+        "relationship_changed": None,
+        "power_breakthrough": None,
+        "artifact_obtained": None,
+        "character_state_changed": None,
+        "world_rule_revealed": None,
+        "open_loop_created": None,
+        "open_loop_closed": None,
+        "promise_created": None,
+        "promise_paid_off": None,
+    }
+
+    def check(self, chapter: int, events: List[dict]) -> List[Dict[str, str | int]]:
+        proposals: List[Dict[str, str | int]] = []
+        for event in events or []:
+            if not isinstance(event, dict):
+                continue
+            rule = self.RULES.get(str(event.get("event_type") or "").strip())
+            if not rule:
+                continue
+            payload = dict(event.get("payload") or {})
+            proposal = AmendProposal(
+                proposal_id=f"amend-{chapter}-{event.get('event_id')}",
+                chapter=chapter,
+                target_level=rule["target_level"],
+                field=str(payload.get("field") or "").strip(),
+                base_value=str(payload.get("base_value") or "").strip(),
+                proposed_value=str(payload.get("proposed_value") or "").strip(),
+                reason_tag=rule["reason_tag"],
+            )
+            proposals.append(proposal.model_dump())
+        return proposals
+
+
+def persist_amend_proposals(
+    conn: sqlite3.Connection, chapter: int, proposals: List[dict]
+) -> int:
+    inserted = 0
+    for proposal in proposals or []:
+        row = normalize_override_record(
+            record_type="amend_proposal",
+            field=str(proposal.get("field") or ""),
+            base_value=str(proposal.get("base_value") or ""),
+            override_value=str(proposal.get("proposed_value") or ""),
+            source_level=str(proposal.get("target_level") or ""),
+        )
+        cursor = conn.execute(
+            """
+            INSERT OR IGNORE INTO override_contracts (
+                chapter,
+                constraint_type,
+                constraint_id,
+                rationale_type,
+                rationale_text,
+                payback_plan,
+                due_chapter,
+                status,
+                record_type,
+                field,
+                base_value,
+                override_value,
+                source_level,
+                reason_tag
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                chapter,
+                "AMEND_PROPOSAL",
+                str(proposal.get("proposal_id") or ""),
+                "story_amend_proposal",
+                f"事件触发合同修订提案: {proposal.get('proposal_id')}",
+                "",
+                chapter,
+                "pending",
+                row["record_type"],
+                row["field"],
+                row["base_value"],
+                row["override_value"],
+                row["source_level"],
+                str(proposal.get("reason_tag") or ""),
+            ),
+        )
+        inserted += max(int(cursor.rowcount), 0)
+    return inserted

webnovel-writer/scripts/data_modules/prewrite_validator.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any, Dict
+
+
+class PrewriteValidator:
+    def __init__(self, project_root: Path):
+        self.project_root = Path(project_root)
+
+    def build(
+        self,
+        chapter: int,
+        review_contract: Dict[str, Any],
+        plot_structure: Dict[str, Any],
+    ) -> 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 []
+        return {
+            "chapter": chapter,
+            "blocking": bool(pending),
+            "blocking_reasons": ["存在高优先级 disambiguation_pending"] if pending else [],
+            "forbidden_zones": list(review_contract.get("blocking_rules") or []),
+            "disambiguation_domain": {
+                "pending_count": len(pending),
+                "warning_count": len(warnings),
+                "allowed_mentions": [
+                    item.get("mention", "")
+                    for item in warnings
+                    if isinstance(item, dict) and item.get("mention")
+                ],
+            },
+            "fulfillment_seed": {
+                "planned_nodes": list(plot_structure.get("mandatory_nodes") or []),
+                "prohibitions": list(plot_structure.get("prohibitions") or []),
+            },
+        }

webnovel-writer/scripts/data_modules/runtime_contract_builder.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Dict, Tuple
+
+from chapter_outline_loader import load_chapter_plot_structure, volume_num_for_chapter_from_state
+
+from .story_contract_schema import MasterSetting, ReviewContract, VolumeBrief
+from .story_contracts import read_json_if_exists
+
+
+class RuntimeContractBuilder:
+    def __init__(self, project_root: Path):
+        self.project_root = Path(project_root)
+
+    def build_for_chapter(self, chapter: int) -> Tuple[Dict[str, Any], Dict[str, Any]]:
+        master = self._load_master_setting()
+        anti_patterns = self._load_anti_patterns()
+        plot = self._load_plot_structure(chapter)
+        volume = self._resolve_volume(chapter)
+
+        volume_brief = VolumeBrief.model_validate(
+            {
+                "meta": {"schema_version": "story-system/v1", "contract_type": "VOLUME_BRIEF"},
+                "volume_goal": {"summary": f"第{volume}卷延续 {master.route.get('primary_genre', '')} 的主冲突"},
+                "selected_tropes": [master.route.get("primary_genre", "")],
+                "selected_pacing": {"wave": master.master_constraints.get("pacing_strategy", "")},
+                "selected_scenes": list(plot.get("cpns") or []),
+                "anti_patterns": [row.get("text", "") for row in anti_patterns if row.get("text")],
+                "system_constraints": [master.master_constraints.get("core_tone", "")] if master.master_constraints.get("core_tone") else [],
+                "overrides": {"locked": {}, "append_only": {}, "override_allowed": {}},
+            }
+        ).model_dump()
+        review_contract = ReviewContract.model_validate(
+            {
+                "meta": {"schema_version": "story-system/v1", "contract_type": "REVIEW_CONTRACT"},
+                "must_check": list(plot.get("mandatory_nodes") or []),
+                "blocking_rules": list(plot.get("prohibitions") or []),
+                "genre_specific_risks": [master.route.get("primary_genre", "")] if master.route.get("primary_genre") else [],
+                "anti_patterns": volume_brief["anti_patterns"],
+                "system_constraints": volume_brief["system_constraints"],
+                "review_thresholds": {"blocking_count": 0, "missed_nodes": 0},
+                "overrides": {"locked": {}, "append_only": {}, "override_allowed": {}},
+            }
+        ).model_dump()
+        return volume_brief, review_contract
+
+    def _load_master_setting(self) -> MasterSetting:
+        raw = read_json_if_exists(self.project_root / ".story-system" / "MASTER_SETTING.json") or {}
+        return MasterSetting.model_validate(raw)
+
+    def _load_anti_patterns(self) -> list[Dict[str, Any]]:
+        raw = read_json_if_exists(self.project_root / ".story-system" / "anti_patterns.json") or []
+        return list(raw)
+
+    def _load_plot_structure(self, chapter: int) -> Dict[str, Any]:
+        raw = load_chapter_plot_structure(self.project_root, chapter) or {}
+        return {
+            "mandatory_nodes": list(raw.get("mandatory_nodes") or []),
+            "prohibitions": list(raw.get("prohibitions") or []),
+            "cpns": list(raw.get("cpns") or []),
+        }
+
+    def _resolve_volume(self, chapter: int) -> int:
+        return volume_num_for_chapter_from_state(self.project_root, chapter) or 1

webnovel-writer/scripts/data_modules/state_projection_writer.py

@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+from pathlib import Path
+
+from .story_contracts import read_json_if_exists, write_json
+
+
+class StateProjectionWriter:
+    def __init__(self, project_root: Path):
+        self.project_root = Path(project_root)
+
+    def apply(self, commit_payload: dict) -> dict:
+        if commit_payload["meta"]["status"] != "accepted":
+            return {"applied": False, "writer": "state", "reason": "commit_rejected"}
+
+        chapter = int(commit_payload.get("meta", {}).get("chapter") or 0)
+        state_path = self.project_root / ".webnovel" / "state.json"
+        state = read_json_if_exists(state_path) or {}
+        entity_state = state.setdefault("entity_state", {})
+        progress = state.setdefault("progress", {})
+        chapter_status = progress.setdefault("chapter_status", {})
+
+        applied_count = 0
+        for delta in self._collect_state_deltas(commit_payload):
+            entity_id = str(delta.get("entity_id") or "").strip()
+            field = str(delta.get("field") or "").strip()
+            if not entity_id or not field:
+                continue
+            entity_state.setdefault(entity_id, {})[field] = delta.get("new")
+            applied_count += 1
+
+        if chapter > 0:
+            chapter_status[str(chapter)] = "chapter_committed"
+
+        write_json(state_path, state)
+        return {
+            "applied": applied_count > 0 or chapter > 0,
+            "writer": "state",
+            "applied_count": applied_count,
+        }
+
+    def _collect_state_deltas(self, commit_payload: dict) -> list[dict]:
+        deltas = [dict(delta) for delta in (commit_payload.get("state_deltas") or []) if isinstance(delta, dict)]
+        seen = {
+            (
+                str(delta.get("entity_id") or "").strip(),
+                str(delta.get("field") or "").strip(),
+            )
+            for delta in deltas
+        }
+
+        for event in commit_payload.get("accepted_events") or []:
+            if not isinstance(event, dict):
+                continue
+            event_type = str(event.get("event_type") or "").strip()
+            payload = dict(event.get("payload") or {})
+            entity_id = str(payload.get("entity_id") or event.get("subject") or "").strip()
+            if not entity_id:
+                continue
+
+            field = ""
+            if event_type == "power_breakthrough":
+                field = str(payload.get("field") or "realm").strip()
+            elif event_type == "character_state_changed":
+                field = str(payload.get("field") or "").strip()
+            else:
+                continue
+
+            key = (entity_id, field)
+            if not field or key in seen:
+                continue
+
+            seen.add(key)
+            deltas.append(
+                {
+                    "entity_id": entity_id,
+                    "field": field,
+                    "old": payload.get("old") if "old" in payload else payload.get("from"),
+                    "new": payload.get("new") if "new" in payload else payload.get("to"),
+                }
+            )
+        return deltas

webnovel-writer/scripts/data_modules/story_contract_schema.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+from typing import Any, Dict, List
+
+from pydantic import BaseModel, Field
+
+
+class ContractMeta(BaseModel):
+    schema_version: str = "story-system/v1"
+    contract_type: str
+    generator_version: str = "phase2"
+    source_trace: List[Dict[str, Any]] = Field(default_factory=list)
+
+
+class OverrideBundle(BaseModel):
+    locked: Dict[str, Any] = Field(default_factory=dict)
+    append_only: Dict[str, Any] = Field(default_factory=dict)
+    override_allowed: Dict[str, Any] = Field(default_factory=dict)
+
+
+class MasterSetting(BaseModel):
+    meta: ContractMeta
+    route: Dict[str, Any] = Field(default_factory=dict)
+    master_constraints: Dict[str, Any] = Field(default_factory=dict)
+    base_context: List[Dict[str, Any]] = Field(default_factory=list)
+    source_trace: List[Dict[str, Any]] = Field(default_factory=list)
+    override_policy: Dict[str, List[str]] = Field(default_factory=dict)
+
+
+class ChapterBrief(BaseModel):
+    meta: ContractMeta
+    override_allowed: Dict[str, Any] = Field(default_factory=dict)
+    dynamic_context: List[Dict[str, Any]] = Field(default_factory=list)
+    source_trace: List[Dict[str, Any]] = Field(default_factory=list)
+
+
+class VolumeBrief(BaseModel):
+    meta: ContractMeta
+    volume_goal: Dict[str, Any]
+    selected_tropes: List[str] = Field(default_factory=list)
+    selected_pacing: Dict[str, Any] = Field(default_factory=dict)
+    selected_scenes: List[str] = Field(default_factory=list)
+    anti_patterns: List[str] = Field(default_factory=list)
+    system_constraints: List[str] = Field(default_factory=list)
+    overrides: OverrideBundle = Field(default_factory=OverrideBundle)
+
+
+class ReviewContract(BaseModel):
+    meta: ContractMeta
+    must_check: List[str] = Field(default_factory=list)
+    blocking_rules: List[str] = Field(default_factory=list)
+    genre_specific_risks: List[str] = Field(default_factory=list)
+    anti_patterns: List[str] = Field(default_factory=list)
+    system_constraints: List[str] = Field(default_factory=list)
+    review_thresholds: Dict[str, Any] = Field(default_factory=dict)
+    overrides: OverrideBundle = Field(default_factory=OverrideBundle)

webnovel-writer/scripts/data_modules/story_contracts.py

@@ -0,0 +1,208 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Dict, Iterable, List
+
+from chapter_outline_loader import volume_num_for_chapter_from_state
+
+
+MARKER_BEGIN = "<!-- STORY-SYSTEM:BEGIN -->"
+MARKER_END = "<!-- STORY-SYSTEM:END -->"
+
+
+@dataclass(frozen=True)
+class StoryContractPaths:
+    project_root: Path
+
+    @classmethod
+    def from_project_root(cls, project_root: str | Path) -> "StoryContractPaths":
+        return cls(Path(project_root).expanduser().resolve())
+
+    @property
+    def root(self) -> Path:
+        return self.project_root / ".story-system"
+
+    @property
+    def chapters_dir(self) -> Path:
+        return self.root / "chapters"
+
+    @property
+    def volumes_dir(self) -> Path:
+        return self.root / "volumes"
+
+    @property
+    def reviews_dir(self) -> Path:
+        return self.root / "reviews"
+
+    @property
+    def commits_dir(self) -> Path:
+        return self.root / "commits"
+
+    @property
+    def events_dir(self) -> Path:
+        return self.root / "events"
+
+    @property
+    def master_json(self) -> Path:
+        return self.root / "MASTER_SETTING.json"
+
+    @property
+    def anti_patterns_json(self) -> Path:
+        return self.root / "anti_patterns.json"
+
+    def chapter_json(self, chapter: int) -> Path:
+        return self.chapters_dir / f"chapter_{chapter:03d}.json"
+
+    def volume_json(self, volume: int) -> Path:
+        return self.volumes_dir / f"volume_{volume:03d}.json"
+
+    def review_json(self, chapter: int) -> Path:
+        return self.reviews_dir / f"chapter_{chapter:03d}.review.json"
+
+    def commit_json(self, chapter: int) -> Path:
+        return self.commits_dir / f"chapter_{chapter:03d}.commit.json"
+
+    def event_json(self, chapter: int) -> Path:
+        return self.events_dir / f"chapter_{chapter:03d}.events.json"
+
+
+def _merge_append_only(master: Dict[str, Any], chapter: Dict[str, Any]) -> Dict[str, List[Any]]:
+    merged: Dict[str, List[Any]] = {}
+    for key in set(master) | set(chapter):
+        seen: List[Any] = []
+        for source_list in (master.get(key) or [], chapter.get(key) or []):
+            for item in source_list:
+                if item not in seen:
+                    seen.append(item)
+        merged[key] = seen
+    return merged
+
+
+def merge_contract_layers(master: Dict[str, Any], chapter: Dict[str, Any] | None) -> Dict[str, Any]:
+    chapter = chapter or {}
+    return {
+        "locked": dict(master.get("locked") or {}),
+        "append_only": _merge_append_only(
+            master.get("append_only") or {},
+            chapter.get("append_only") or {},
+        ),
+        "override_allowed": {
+            **(master.get("override_allowed") or {}),
+            **(chapter.get("override_allowed") or {}),
+        },
+    }
+
+
+def merge_anti_patterns(*groups: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    seen: set[str] = set()
+    merged: List[Dict[str, Any]] = []
+    for group in groups:
+        for row in group:
+            text = str(row.get("text") or "").strip()
+            if not text or text in seen:
+                continue
+            seen.add(text)
+            merged.append(dict(row))
+    return merged
+
+
+def read_json_if_exists(path: Path) -> Any | None:
+    if not path.is_file():
+        return None
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError as exc:
+        raise ValueError(f"Bad JSON in {path}") from exc
+
+
+def write_json(path: Path, payload: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def write_marked_markdown(path: Path, generated_block: str) -> None:
+    wrapped = f"{MARKER_BEGIN}\n{generated_block.rstrip()}\n{MARKER_END}\n"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    if path.exists():
+        current = path.read_text(encoding="utf-8")
+        if current.count(MARKER_BEGIN) > 1 or current.count(MARKER_END) > 1:
+            raise ValueError(f"{path} contains multiple STORY-SYSTEM markers")
+        if MARKER_BEGIN in current and MARKER_END in current:
+            before, _, rest = current.partition(MARKER_BEGIN)
+            _, _, after = rest.partition(MARKER_END)
+            path.write_text(f"{before}{wrapped}{after.lstrip()}", encoding="utf-8")
+            return
+    path.write_text(wrapped, encoding="utf-8")
+
+
+def render_master_markdown(master_payload: Dict[str, Any]) -> str:
+    route = master_payload.get("route") or {}
+    constraints = master_payload.get("master_constraints") or {}
+    return "\n".join(
+        [
+            "# MASTER_SETTING",
+            f"- 题材：{route.get('primary_genre', '')}",
+            f"- 调性：{constraints.get('core_tone', '')}",
+            f"- 节奏：{constraints.get('pacing_strategy', '')}",
+        ]
+    )
+
+
+def render_anti_patterns_markdown(anti_patterns: List[Dict[str, Any]]) -> str:
+    lines = ["# ANTI_PATTERNS"]
+    for row in anti_patterns:
+        lines.append(f"- {row.get('text', '')}")
+    return "\n".join(lines)
+
+
+def render_chapter_markdown(chapter_payload: Dict[str, Any]) -> str:
+    focus = (chapter_payload.get("override_allowed") or {}).get("chapter_focus", "")
+    return "\n".join(
+        [
+            f"# CHAPTER_{int(chapter_payload['meta']['chapter']):03d}",
+            f"- 章节焦点：{focus}",
+        ]
+    )
+
+
+def persist_story_seed(
+    project_root: Path,
+    master_payload: Dict[str, Any],
+    chapter_payload: Dict[str, Any] | None,
+    anti_patterns: List[Dict[str, Any]],
+) -> None:
+    paths = StoryContractPaths.from_project_root(project_root)
+    paths.root.mkdir(parents=True, exist_ok=True)
+    paths.chapters_dir.mkdir(parents=True, exist_ok=True)
+    write_json(paths.master_json, master_payload)
+    write_json(paths.anti_patterns_json, anti_patterns)
+    write_marked_markdown(paths.master_json.with_suffix(".md"), render_master_markdown(master_payload))
+    write_marked_markdown(
+        paths.anti_patterns_json.with_suffix(".md"),
+        render_anti_patterns_markdown(anti_patterns),
+    )
+    if chapter_payload is not None:
+        chapter_num = int(chapter_payload["meta"]["chapter"])
+        write_json(paths.chapter_json(chapter_num), chapter_payload)
+        write_marked_markdown(
+            paths.chapter_json(chapter_num).with_suffix(".md"),
+            render_chapter_markdown(chapter_payload),
+        )
+
+
+def persist_runtime_contracts(
+    project_root: Path,
+    chapter: int,
+    volume_brief: Dict[str, Any],
+    review_contract: Dict[str, Any],
+) -> None:
+    paths = StoryContractPaths.from_project_root(project_root)
+    volume = volume_num_for_chapter_from_state(paths.project_root, chapter) or 1
+    paths.volumes_dir.mkdir(parents=True, exist_ok=True)
+    paths.reviews_dir.mkdir(parents=True, exist_ok=True)
+    write_json(paths.volume_json(volume), volume_brief)
+    write_json(paths.review_json(chapter), review_contract)

webnovel-writer/scripts/data_modules/story_event_schema.py

@@ -0,0 +1,26 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+from typing import Any, Dict, Literal
+
+from pydantic import BaseModel, Field
+
+
+class StoryEvent(BaseModel):
+    event_id: str
+    chapter: int = Field(ge=1)
+    event_type: Literal[
+        "character_state_changed",
+        "relationship_changed",
+        "world_rule_revealed",
+        "world_rule_broken",
+        "power_breakthrough",
+        "artifact_obtained",
+        "promise_created",
+        "promise_paid_off",
+        "open_loop_created",
+        "open_loop_closed",
+    ]
+    subject: str
+    payload: Dict[str, Any] = Field(default_factory=dict)

webnovel-writer/scripts/data_modules/story_system_engine.py

@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import csv
+import re
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from reference_search import search as search_reference
+
+from .story_contracts import merge_anti_patterns
+
+
+ANTI_PATTERN_SOURCE_FIELDS = {
+    "场景写法": ["反面写法"],
+    "写作技法": ["常见误区"],
+    "爽点与节奏": ["常见崩盘误区"],
+    "人设与关系": ["忌讳写法"],
+    "桥段套路": ["忌讳写法"],
+    "题材与调性推理": ["强制禁忌/毒点"],
+}
+
+
+class StorySystemEngine:
+    def __init__(self, csv_dir: str | Path):
+        self.csv_dir = Path(csv_dir)
+
+    def build(self, query: str, genre: Optional[str], chapter: Optional[int]) -> Dict[str, Any]:
+        route = self._route(query=query, genre=genre)
+        search_query = self._expand_query(query, route.get("default_query", ""))
+        base_context = self._collect_tables(
+            search_query,
+            route["recommended_base_tables"],
+            genre=route["genre_filter"],
+            top_k=1,
+        )
+        dynamic_context = self._collect_tables(
+            search_query,
+            route["recommended_dynamic_tables"],
+            genre=route["genre_filter"],
+            top_k=2,
+        )
+        source_trace = route["source_trace"] + self._build_source_trace(base_context, dynamic_context)
+        anti_patterns = merge_anti_patterns(
+            route["route_anti_patterns"],
+            self._extract_anti_patterns(base_context),
+            self._extract_anti_patterns(dynamic_context),
+        )
+        return {
+            "meta": {"query": query, "chapter": chapter, "explicit_genre": genre or ""},
+            "master_setting": {
+                "meta": {
+                    "schema_version": "story-system/v1",
+                    "contract_type": "MASTER_SETTING",
+                    "generator_version": "phase1",
+                    "query": query,
+                },
+                "route": route["meta"],
+                "master_constraints": {
+                    "core_tone": route["core_tone"],
+                    "pacing_strategy": route["pacing_strategy"],
+                },
+                "base_context": base_context,
+                "source_trace": source_trace,
+                "override_policy": {
+                    "locked": ["route.primary_genre", "master_constraints.core_tone"],
+                    "append_only": ["anti_patterns"],
+                    "override_allowed": [],
+                },
+            },
+            "chapter_brief": (
+                {
+                    "meta": {
+                        "schema_version": "story-system/v1",
+                        "contract_type": "CHAPTER_BRIEF",
+                        "generator_version": "phase1",
+                        "chapter": chapter,
+                    },
+                    "override_allowed": {
+                        "chapter_focus": self._suggest_chapter_focus(query, dynamic_context),
+                    },
+                    "dynamic_context": dynamic_context,
+                    "source_trace": source_trace,
+                }
+                if chapter is not None
+                else None
+            ),
+            "anti_patterns": anti_patterns,
+        }
+
+    def _route(self, query: str, genre: Optional[str]) -> Dict[str, Any]:
+        route_rows = self._load_csv_rows("题材与调性推理")
+        query_text = self._normalize_text(" ".join([query or "", genre or ""]))
+
+        matched = None
+        route_source = "empty_csv_fallback"
+        for row in route_rows:
+            aliases = (
+                self._split_multi_value(row.get("关键词"))
+                + self._split_multi_value(row.get("意图与同义词"))
+                + self._split_multi_value(row.get("题材别名"))
+            )
+            if any(alias and self._normalize_text(alias) in query_text for alias in aliases):
+                matched = row
+                route_source = "keyword_or_alias_match"
+                break
+        if matched is None and genre:
+            matched = self._fallback_row_for_genre(route_rows, genre)
+            if matched is not None:
+                route_source = "explicit_genre_fallback"
+        if matched is None and route_rows:
+            matched = route_rows[0]
+            route_source = "default_seed_fallback"
+        if matched is None:
+            return self._empty_route(query=query, genre=genre)
+
+        primary_genre = str(matched.get("题材/流派") or genre or "").strip()
+        genre_filter = str(matched.get("适用题材") or genre or primary_genre).strip()
+        return {
+            "meta": {
+                "primary_genre": primary_genre,
+                "route_source": route_source,
+                "genre_filter": genre_filter,
+                "recommended_base_tables": self._split_multi_value(matched.get("推荐基础检索表")),
+                "recommended_dynamic_tables": self._split_multi_value(matched.get("推荐动态检索表")),
+            },
+            "core_tone": str(matched.get("核心调性") or "").strip(),
+            "pacing_strategy": str(matched.get("节奏策略") or "").strip(),
+            "route_anti_patterns": self._extract_route_anti_patterns(matched),
+            "recommended_base_tables": self._split_multi_value(matched.get("推荐基础检索表")),
+            "recommended_dynamic_tables": self._split_multi_value(matched.get("推荐动态检索表")),
+            "genre_filter": genre_filter,
+            "default_query": str(matched.get("默认查询词") or "").strip(),
+            "source_trace": [{"table": "题材与调性推理", "id": matched.get("编号", ""), "reason": route_source}],
+        }
+
+    def _collect_tables(self, query: str, tables: List[str], genre: str, top_k: int) -> List[Dict[str, Any]]:
+        rows: List[Dict[str, Any]] = []
+        for table_name in tables:
+            result = search_reference(
+                csv_dir=self.csv_dir,
+                skill="write",
+                query=query,
+                table=table_name,
+                genre=genre or None,
+                max_results=top_k,
+            )
+            raw_rows = {str(row.get("编号") or ""): row for row in self._load_csv_rows(table_name)}
+            for item in result.get("data", {}).get("results", []):
+                row_id = str(item.get("编号") or "")
+                full_row = dict(raw_rows.get(row_id) or {})
+                full_row["_table"] = str(item.get("表") or table_name)
+                full_row["编号"] = row_id
+                full_row["核心摘要"] = str(
+                    full_row.get("核心摘要")
+                    or item.get("内容摘要")
+                    or item.get("核心摘要")
+                    or ""
+                ).strip()
+                rows.append(full_row)
+        return rows
+
+    def _extract_anti_patterns(self, rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        extracted: List[Dict[str, Any]] = []
+        for row in rows:
+            table_name = str(row.get("_table") or "")
+            for field_name in ANTI_PATTERN_SOURCE_FIELDS.get(table_name, []):
+                for text in self._split_multi_value(row.get(field_name)):
+                    extracted.append(
+                        {
+                            "text": text,
+                            "source_table": table_name,
+                            "source_id": row.get("编号", ""),
+                        }
+                    )
+        return extracted
+
+    def _suggest_chapter_focus(self, query: str, dynamic_rows: List[Dict[str, Any]]) -> str:
+        for row in dynamic_rows:
+            summary = str(row.get("核心摘要") or "").strip()
+            if summary:
+                return summary
+        return query
+
+    def _build_source_trace(self, *groups: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        trace: List[Dict[str, Any]] = []
+        for group in groups:
+            for row in group:
+                trace.append(
+                    {
+                        "table": row.get("_table", ""),
+                        "id": row.get("编号", ""),
+                        "summary": row.get("核心摘要", ""),
+                    }
+                )
+        return trace
+
+    def _load_csv_rows(self, table_name: str) -> List[Dict[str, Any]]:
+        csv_path = self.csv_dir / f"{table_name}.csv"
+        if not csv_path.is_file():
+            return []
+        with csv_path.open("r", encoding="utf-8-sig", newline="") as f:
+            return list(csv.DictReader(f))
+
+    def _normalize_text(self, text: str) -> str:
+        return str(text or "").strip().lower()
+
+    def _split_multi_value(self, raw: Any) -> List[str]:
+        return [item.strip() for item in re.split(r"[|；;]+", str(raw or "")) if item.strip()]
+
+    def _expand_query(self, query: str, default_query: str) -> str:
+        items: List[str] = []
+        for candidate in [query, *self._split_multi_value(default_query)]:
+            text = str(candidate or "").strip()
+            if text and text not in items:
+                items.append(text)
+        return " ".join(items)
+
+    def _fallback_row_for_genre(self, rows: List[Dict[str, Any]], genre: str) -> Dict[str, Any] | None:
+        genre_text = self._normalize_text(genre)
+        for row in rows:
+            candidates = self._split_multi_value(row.get("适用题材")) + self._split_multi_value(row.get("题材/流派"))
+            if any(self._normalize_text(candidate) == genre_text for candidate in candidates):
+                return row
+        return None
+
+    def _extract_route_anti_patterns(self, row: Dict[str, Any]) -> List[Dict[str, Any]]:
+        return [
+            {"text": text, "source_table": "题材与调性推理", "source_id": row.get("编号", "")}
+            for text in self._split_multi_value(row.get("强制禁忌/毒点"))
+        ]
+
+    def _empty_route(self, query: str, genre: Optional[str]) -> Dict[str, Any]:
+        fallback_genre = str(genre or "未命中题材").strip()
+        route_source = "explicit_genre_fallback" if genre else "empty_csv_fallback"
+        return {
+            "meta": {
+                "primary_genre": fallback_genre,
+                "route_source": route_source,
+                "genre_filter": fallback_genre,
+                "recommended_base_tables": ["命名规则", "人设与关系"],
+                "recommended_dynamic_tables": ["桥段套路", "爽点与节奏", "场景写法"],
+            },
+            "core_tone": "",
+            "pacing_strategy": "",
+            "route_anti_patterns": [],
+            "recommended_base_tables": ["命名规则", "人设与关系"],
+            "recommended_dynamic_tables": ["桥段套路", "爽点与节奏", "场景写法"],
+            "genre_filter": fallback_genre,
+            "default_query": "",
+            "source_trace": [{"table": "题材与调性推理", "id": "", "reason": f"{route_source}:{query}"}],
+        }

webnovel-writer/scripts/data_modules/summary_projection_writer.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+from pathlib import Path
+
+
+def append_summary_projection(project_root: Path, commit_payload: dict) -> dict:
+    chapter = int(commit_payload.get("meta", {}).get("chapter") or 0)
+    summary_text = str(commit_payload.get("summary_text") or "").strip()
+    if chapter <= 0 or not summary_text:
+        return {"applied": False, "writer": "summary", "reason": "missing_summary"}
+
+    target = Path(project_root) / ".webnovel" / "summaries" / f"ch{chapter:04d}.md"
+    target.parent.mkdir(parents=True, exist_ok=True)
+    if "## 剧情摘要" not in summary_text:
+        summary_text = f"## 剧情摘要\n{summary_text}\n"
+    target.write_text(summary_text, encoding="utf-8")
+    return {"applied": True, "writer": "summary", "path": str(target)}
+
+
+class SummaryProjectionWriter:
+    def __init__(self, project_root: Path):
+        self.project_root = Path(project_root)
+
+    def apply(self, commit_payload: dict) -> dict:
+        if commit_payload["meta"]["status"] != "accepted":
+            return {"applied": False, "writer": "summary", "reason": "commit_rejected"}
+        return append_summary_projection(self.project_root, commit_payload)

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

@@ -0,0 +1,131 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import sys
+from pathlib import Path
+
+from data_modules.chapter_commit_service import ChapterCommitService
+from data_modules.config import DataModulesConfig
+from data_modules.index_manager import IndexManager
+
+
+def test_commit_service_rejects_when_missed_nodes_exist(tmp_path):
+    service = ChapterCommitService(tmp_path)
+    payload = service.build_commit(
+        chapter=3,
+        review_result={"blocking_count": 0},
+        fulfillment_result={"planned_nodes": ["发现陷阱"], "missed_nodes": ["发现陷阱"]},
+        disambiguation_result={"pending": []},
+        extraction_result={"state_deltas": [], "entity_deltas": [], "accepted_events": []},
+    )
+    assert payload["meta"]["status"] == "rejected"
+
+
+def test_commit_service_accepts_when_all_checks_pass(tmp_path):
+    service = ChapterCommitService(tmp_path)
+    payload = service.build_commit(
+        chapter=3,
+        review_result={"blocking_count": 0},
+        fulfillment_result={"planned_nodes": ["发现陷阱"], "covered_nodes": ["发现陷阱"], "missed_nodes": [], "extra_nodes": []},
+        disambiguation_result={"pending": []},
+        extraction_result={"state_deltas": [], "entity_deltas": [], "accepted_events": []},
+    )
+    assert payload["meta"]["status"] == "accepted"
+    assert payload["contract_refs"]["master"] == "MASTER_SETTING.json"
+    assert payload["contract_refs"]["chapter"] == "chapter_003.json"
+    assert payload["outline_snapshot"]["covered_nodes"] == ["发现陷阱"]
+
+
+def test_chapter_commit_cli_builds_and_persists_commit(tmp_path, monkeypatch):
+    review_path = tmp_path / "review.json"
+    fulfillment_path = tmp_path / "fulfillment.json"
+    disambiguation_path = tmp_path / "disambiguation.json"
+    extraction_path = tmp_path / "extraction.json"
+    review_path.write_text('{"blocking_count": 0}', encoding="utf-8")
+    fulfillment_path.write_text(
+        '{"planned_nodes": ["发现陷阱"], "covered_nodes": ["发现陷阱"], "missed_nodes": [], "extra_nodes": []}',
+        encoding="utf-8",
+    )
+    disambiguation_path.write_text('{"pending": []}', encoding="utf-8")
+    extraction_path.write_text('{"state_deltas": [], "entity_deltas": [], "accepted_events": []}', encoding="utf-8")
+
+    scripts_dir = Path(__file__).resolve().parents[2]
+    if str(scripts_dir) not in sys.path:
+        sys.path.insert(0, str(scripts_dir))
+
+    from chapter_commit import main
+
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "chapter_commit",
+            "--project-root",
+            str(tmp_path),
+            "--chapter",
+            "3",
+            "--review-result",
+            str(review_path),
+            "--fulfillment-result",
+            str(fulfillment_path),
+            "--disambiguation-result",
+            str(disambiguation_path),
+            "--extraction-result",
+            str(extraction_path),
+        ],
+    )
+    main()
+
+    assert (tmp_path / ".story-system" / "commits" / "chapter_003.commit.json").is_file()
+
+
+def test_apply_projections_writes_events_and_amend_proposals(tmp_path):
+    service = ChapterCommitService(tmp_path)
+    payload = service.build_commit(
+        chapter=3,
+        review_result={"blocking_count": 0},
+        fulfillment_result={
+            "planned_nodes": ["发现陷阱"],
+            "covered_nodes": ["发现陷阱"],
+            "missed_nodes": [],
+            "extra_nodes": [],
+        },
+        disambiguation_result={"pending": []},
+        extraction_result={
+            "state_deltas": [],
+            "entity_deltas": [],
+            "summary_text": "",
+            "accepted_events": [
+                {
+                    "event_id": "evt-001",
+                    "chapter": 3,
+                    "event_type": "world_rule_broken",
+                    "subject": "金手指",
+                    "payload": {
+                        "field": "world_rule",
+                        "base_value": "每日一次",
+                        "proposed_value": "短时失控突破",
+                    },
+                }
+            ],
+        },
+    )
+
+    service.apply_projections(payload)
+
+    assert (tmp_path / ".story-system" / "events" / "chapter_003.events.json").is_file()
+    manager = IndexManager(DataModulesConfig.from_project_root(tmp_path))
+    with manager._get_conn() as conn:
+        row = conn.execute(
+            """
+            SELECT record_type, field, override_value, status
+            FROM override_contracts
+            ORDER BY id DESC
+            LIMIT 1
+            """
+        ).fetchone()
+
+    assert row["record_type"] == "amend_proposal"
+    assert row["field"] == "world_rule"
+    assert row["override_value"] == "短时失控突破"
+    assert row["status"] == "pending"

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

@@ -188,6 +188,96 @@ def test_context_manager_loads_volume_outline_file(temp_project):
     assert "测试大纲" in outline
 
 
+def test_context_manager_includes_story_contract_and_prewrite_validation(temp_project):
+    state = {
+        "progress": {"volumes_planned": [{"volume": 1, "chapters_range": "1-10"}]},
+        "protagonist_state": {"name": "萧炎"},
+        "chapter_meta": {},
+        "disambiguation_warnings": [{"mention": "宗主"}],
+        "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.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").mkdir(parents=True, exist_ok=True)
+    (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").mkdir(parents=True, exist_ok=True)
+    (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": {"wave": "先压后爆"},
+                "selected_scenes": [],
+                "anti_patterns": [],
+                "system_constraints": [],
+                "overrides": {"locked": {}, "append_only": {}, "override_allowed": {}},
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+    (story_root / "reviews").mkdir(parents=True, exist_ok=True)
+    (story_root / "reviews" / "chapter_003.review.json").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)
+    payload = manager.build_context(3, use_snapshot=False, save_snapshot=False)
+
+    sections = payload["sections"]
+    assert "story_contract" in sections
+    assert "prewrite_validation" in sections
+    assert sections["story_contract"]["content"]["review_contract"]["meta"]["contract_type"] == "REVIEW_CONTRACT"
+    assert sections["prewrite_validation"]["content"]["fulfillment_seed"]["planned_nodes"] == ["发现陷阱"]
+    assert list(sections.keys()).index("story_contract") < list(sections.keys()).index("scene")
+
+
 def test_query_router():
     router = QueryRouter()
     assert router.route("角色是谁") == "entity"

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

@@ -0,0 +1,82 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import sqlite3
+import sys
+from pathlib import Path
+
+from data_modules.event_log_store import EventLogStore
+
+
+def _ensure_scripts_on_path() -> None:
+    scripts_dir = Path(__file__).resolve().parents[2]
+    if str(scripts_dir) not in sys.path:
+        sys.path.insert(0, str(scripts_dir))
+
+
+def test_event_log_store_writes_per_chapter_file_and_sqlite_mirror(tmp_path):
+    store = EventLogStore(tmp_path)
+    store.write_events(
+        3,
+        [
+            {
+                "event_id": "evt-001",
+                "chapter": 3,
+                "event_type": "open_loop_created",
+                "subject": "三年之约",
+                "payload": {},
+            }
+        ],
+    )
+    assert (tmp_path / ".story-system" / "events" / "chapter_003.events.json").is_file()
+
+    conn = sqlite3.connect(tmp_path / ".webnovel" / "index.db")
+    try:
+        row = conn.execute(
+            "SELECT event_id, chapter, event_type FROM story_events"
+        ).fetchone()
+    finally:
+        conn.close()
+    assert row == ("evt-001", 3, "open_loop_created")
+
+
+def test_event_log_store_ignores_duplicate_event_id(tmp_path):
+    store = EventLogStore(tmp_path)
+    event = {
+        "event_id": "evt-001",
+        "chapter": 3,
+        "event_type": "open_loop_created",
+        "subject": "三年之约",
+        "payload": {},
+    }
+    store.write_events(3, [event])
+    store.write_events(3, [event])
+
+    conn = sqlite3.connect(tmp_path / ".webnovel" / "index.db")
+    try:
+        count = conn.execute("SELECT COUNT(*) FROM story_events").fetchone()[0]
+    finally:
+        conn.close()
+    assert count == 1
+
+
+def test_story_events_cli_reads_chapter_file(tmp_path, monkeypatch, capsys):
+    _ensure_scripts_on_path()
+    events_dir = tmp_path / ".story-system" / "events"
+    events_dir.mkdir(parents=True, exist_ok=True)
+    (events_dir / "chapter_003.events.json").write_text(
+        '[{"event_id":"evt-001","chapter":3,"event_type":"open_loop_created","subject":"三年之约","payload":{}}]',
+        encoding="utf-8",
+    )
+
+    from story_events import main
+
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        ["story_events", "--project-root", str(tmp_path), "--chapter", "3"],
+    )
+    main()
+
+    out = capsys.readouterr().out
+    assert "open_loop_created" in out

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

@@ -0,0 +1,42 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+from data_modules.event_projection_router import EventProjectionRouter
+
+
+def test_router_maps_power_breakthrough_to_state_and_memory():
+    router = EventProjectionRouter()
+    targets = router.route(
+        {"event_type": "power_breakthrough", "subject": "xiaoyan", "payload": {}}
+    )
+    assert targets == ["state", "memory"]
+
+
+def test_router_maps_relationship_changed_to_index():
+    router = EventProjectionRouter()
+    targets = router.route(
+        {
+            "event_type": "relationship_changed",
+            "subject": "xiaoyan",
+            "payload": {"to": "yaolao"},
+        }
+    )
+    assert "index" in targets
+
+
+def test_router_collects_required_writers_from_commit_payload():
+    router = EventProjectionRouter()
+    targets = router.required_writers(
+        {
+            "accepted_events": [
+                {"event_type": "power_breakthrough", "subject": "xiaoyan", "payload": {}},
+                {
+                    "event_type": "relationship_changed",
+                    "subject": "xiaoyan",
+                    "payload": {"to": "yaolao"},
+                },
+            ],
+            "summary_text": "本章摘要",
+        }
+    )
+    assert targets == ["index", "memory", "state", "summary"]

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

@@ -349,3 +349,40 @@ def test_render_text_contains_plot_structure_section(tmp_path):
     assert "- CPN1: 发现石碑异常" in text
     assert "- CEN: 决定深入遗迹核心" in text
     assert "- 本章禁区: 不能提前拿到终极传承" in text
+
+
+def test_render_text_contains_contract_first_runtime_section(tmp_path):
+    scripts_dir = Path(__file__).resolve().parents[2]
+    if str(scripts_dir) not in sys.path:
+        sys.path.insert(0, str(scripts_dir))
+
+    from extract_chapter_context import _render_text
+
+    payload = {
+        "chapter": 12,
+        "outline": "测试大纲",
+        "previous_summaries": [],
+        "state_summary": "状态",
+        "context_contract_version": "v2",
+        "context_weight_stage": "mid",
+        "story_contract": {
+            "review_contract": {
+                "blocking_rules": ["不可提前摊牌", "不能让配角代替主角兑现"],
+            }
+        },
+        "prewrite_validation": {
+            "blocking": False,
+            "forbidden_zones": ["不可提前摊牌"],
+            "fulfillment_seed": {"planned_nodes": ["发现陷阱", "决定隐忍"]},
+        },
+        "plot_structure": {},
+        "reader_signal": {},
+        "genre_profile": {},
+        "writing_guidance": {},
+        "rag_assist": {"invoked": False, "hits": []},
+    }
+
+    text = _render_text(payload)
+    assert "## Contract-First Runtime" in text
+    assert "- Review blocking rules: 2" in text
+    assert "- Prewrite blocking: False" in text

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

@@ -0,0 +1,91 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+from data_modules.config import DataModulesConfig
+from data_modules.index_manager import IndexManager
+from data_modules.override_ledger_service import (
+    AmendProposalTrigger,
+    ensure_override_ledger_columns,
+    normalize_override_record,
+    persist_amend_proposals,
+)
+
+
+def test_normalize_override_record_sets_record_type():
+    row = normalize_override_record(
+        record_type="contract_override",
+        field="core_tone",
+        base_value="先压后爆",
+        override_value="当场爆发",
+        source_level="chapter",
+    )
+    assert row["record_type"] == "contract_override"
+    assert row["field"] == "core_tone"
+
+
+def test_normalize_override_record_supports_amend_proposal():
+    row = normalize_override_record(
+        record_type="amend_proposal",
+        field="world_rule",
+        base_value="金手指每日一次",
+        override_value="金手指失控突破",
+        source_level="master",
+    )
+    assert row["record_type"] == "amend_proposal"
+
+
+def test_world_rule_broken_generates_amend_proposal():
+    trigger = AmendProposalTrigger()
+    proposals = trigger.check(
+        chapter=3,
+        events=[
+            {
+                "event_id": "evt-001",
+                "event_type": "world_rule_broken",
+                "subject": "金手指",
+                "payload": {
+                    "field": "world_rule",
+                    "base_value": "每日一次",
+                    "proposed_value": "短时失控突破",
+                },
+            }
+        ],
+    )
+    assert len(proposals) == 1
+    assert proposals[0]["target_level"] == "master"
+    assert proposals[0]["field"] == "world_rule"
+
+
+def test_persist_amend_proposals_writes_pending_rows(tmp_path):
+    manager = IndexManager(DataModulesConfig.from_project_root(tmp_path))
+    proposals = [
+        {
+            "proposal_id": "amend-3-evt-001",
+            "chapter": 3,
+            "target_level": "master",
+            "field": "world_rule",
+            "base_value": "每日一次",
+            "proposed_value": "短时失控突破",
+            "reason_tag": "world_rule_broken",
+        }
+    ]
+
+    with manager._get_conn() as conn:
+        ensure_override_ledger_columns(conn)
+        inserted = persist_amend_proposals(conn, 3, proposals)
+        conn.commit()
+
+    with manager._get_conn() as conn:
+        row = conn.execute(
+            """
+            SELECT record_type, field, override_value, source_level, status
+            FROM override_contracts
+            """
+        ).fetchone()
+
+    assert inserted == 1
+    assert row["record_type"] == "amend_proposal"
+    assert row["field"] == "world_rule"
+    assert row["override_value"] == "短时失控突破"
+    assert row["source_level"] == "master"
+    assert row["status"] == "pending"

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

@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import json
+
+from data_modules.prewrite_validator import PrewriteValidator
+
+
+def test_prewrite_validator_builds_disambiguation_domain_and_fulfillment_seed(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": [{"mention": "宗主"}],
+                "chapter_meta": {},
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+    review_contract = {"must_check": ["发现陷阱"], "blocking_rules": ["不可提前摊牌"]}
+    plot_structure = {"mandatory_nodes": ["发现陷阱"], "prohibitions": ["不可提前摊牌"]}
+
+    payload = PrewriteValidator(project_root).build(
+        chapter=3,
+        review_contract=review_contract,
+        plot_structure=plot_structure,
+    )
+
+    assert payload["blocking"] is False
+    assert payload["fulfillment_seed"]["planned_nodes"] == ["发现陷阱"]
+    assert payload["disambiguation_domain"]["pending_count"] == 0

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

@@ -0,0 +1,202 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import json
+
+from data_modules.chapter_commit_service import ChapterCommitService
+from data_modules.config import DataModulesConfig
+from data_modules.index_manager import IndexManager
+from data_modules.memory.store import ScratchpadManager
+from data_modules.index_projection_writer import IndexProjectionWriter
+from data_modules.memory_projection_writer import MemoryProjectionWriter
+from data_modules.state_projection_writer import StateProjectionWriter
+from data_modules.summary_projection_writer import SummaryProjectionWriter
+
+
+def test_state_projection_writer_skips_rejected_commit(tmp_path):
+    writer = StateProjectionWriter(tmp_path)
+    result = writer.apply({"meta": {"status": "rejected"}, "state_deltas": []})
+    assert result["applied"] is False
+
+
+def test_state_projection_writer_applies_accepted_commit(tmp_path):
+    (tmp_path / ".webnovel").mkdir(parents=True, exist_ok=True)
+    (tmp_path / ".webnovel" / "state.json").write_text("{}", encoding="utf-8")
+    writer = StateProjectionWriter(tmp_path)
+    result = writer.apply(
+        {
+            "meta": {"status": "accepted", "chapter": 3},
+            "state_deltas": [{"entity_id": "x", "field": "realm", "new": "斗者"}],
+        }
+    )
+    assert result["applied"] is True
+    payload = json.loads((tmp_path / ".webnovel" / "state.json").read_text(encoding="utf-8"))
+    assert payload["entity_state"]["x"]["realm"] == "斗者"
+    assert payload["progress"]["chapter_status"]["3"] == "chapter_committed"
+
+
+def test_state_projection_writer_derives_delta_from_power_breakthrough_event(tmp_path):
+    (tmp_path / ".webnovel").mkdir(parents=True, exist_ok=True)
+    (tmp_path / ".webnovel" / "state.json").write_text("{}", encoding="utf-8")
+    writer = StateProjectionWriter(tmp_path)
+    result = writer.apply(
+        {
+            "meta": {"status": "accepted", "chapter": 3},
+            "state_deltas": [],
+            "accepted_events": [
+                {
+                    "event_id": "evt-001",
+                    "chapter": 3,
+                    "event_type": "power_breakthrough",
+                    "subject": "xiaoyan",
+                    "payload": {"from": "斗者", "to": "斗师"},
+                }
+            ],
+        }
+    )
+
+    payload = json.loads((tmp_path / ".webnovel" / "state.json").read_text(encoding="utf-8"))
+    assert result["applied"] is True
+    assert payload["entity_state"]["xiaoyan"]["realm"] == "斗师"
+
+
+def test_accepted_commit_updates_state_json_end_to_end(tmp_path):
+    (tmp_path / ".webnovel").mkdir(parents=True, exist_ok=True)
+    (tmp_path / ".webnovel" / "state.json").write_text("{}", encoding="utf-8")
+
+    service = ChapterCommitService(tmp_path)
+    commit_payload = service.build_commit(
+        chapter=3,
+        review_result={"blocking_count": 0},
+        fulfillment_result={"planned_nodes": ["发现陷阱"], "covered_nodes": ["发现陷阱"], "missed_nodes": [], "extra_nodes": []},
+        disambiguation_result={"pending": []},
+        extraction_result={"state_deltas": [{"entity_id": "x", "field": "realm", "new": "斗者"}], "entity_deltas": [], "accepted_events": []},
+    )
+
+    StateProjectionWriter(tmp_path).apply(commit_payload)
+    payload = json.loads((tmp_path / ".webnovel" / "state.json").read_text(encoding="utf-8"))
+    assert payload["entity_state"]["x"]["realm"] == "斗者"
+
+
+def test_index_projection_writer_applies_entity_delta(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": [
+                {
+                    "entity_id": "xiaoyan",
+                    "canonical_name": "萧炎",
+                    "type": "角色",
+                    "current": {"realm": "斗者"},
+                    "chapter": 3,
+                }
+            ],
+        }
+    )
+
+    entity = IndexManager(cfg).get_entity("xiaoyan")
+    assert result["applied"] is True
+    assert entity["canonical_name"] == "萧炎"
+    assert entity["current_json"]["realm"] == "斗者"
+
+
+def test_index_projection_writer_derives_relationship_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-001",
+                    "chapter": 3,
+                    "event_type": "relationship_changed",
+                    "subject": "xiaoyan",
+                    "payload": {
+                        "to_entity": "yaolao",
+                        "relationship_type": "师徒",
+                        "description": "关系正式确立",
+                    },
+                }
+            ],
+        }
+    )
+
+    rels = IndexManager(cfg).get_relationship_between("xiaoyan", "yaolao")
+    assert result["applied"] is True
+    assert rels[0]["type"] == "师徒"
+
+
+def test_summary_projection_writer_writes_summary_markdown(tmp_path):
+    cfg = DataModulesConfig.from_project_root(tmp_path)
+    cfg.ensure_dirs()
+    writer = SummaryProjectionWriter(tmp_path)
+
+    result = writer.apply(
+        {
+            "meta": {"status": "accepted", "chapter": 3},
+            "summary_text": "本章主角发现陷阱并决定隐忍。",
+        }
+    )
+
+    summary_path = tmp_path / ".webnovel" / "summaries" / "ch0003.md"
+    assert result["applied"] is True
+    assert summary_path.is_file()
+    assert "剧情摘要" in summary_path.read_text(encoding="utf-8")
+
+
+def test_memory_projection_writer_maps_commit_into_scratchpad(tmp_path):
+    cfg = DataModulesConfig.from_project_root(tmp_path)
+    cfg.ensure_dirs()
+    writer = MemoryProjectionWriter(tmp_path)
+
+    result = writer.apply(
+        {
+            "meta": {"status": "accepted", "chapter": 3},
+            "state_deltas": [
+                {"entity_id": "xiaoyan", "field": "realm", "old": "斗者", "new": "斗师"}
+            ],
+            "entity_deltas": [],
+            "accepted_events": [],
+        }
+    )
+
+    store = ScratchpadManager(cfg)
+    chars = store.query(category="character_state", status="active")
+    assert result["applied"] is True
+    assert any(x.subject == "xiaoyan" and x.field == "realm" for x in chars)
+
+
+def test_memory_projection_writer_maps_open_loop_event_into_scratchpad(tmp_path):
+    cfg = DataModulesConfig.from_project_root(tmp_path)
+    cfg.ensure_dirs()
+    writer = MemoryProjectionWriter(tmp_path)
+
+    result = writer.apply(
+        {
+            "meta": {"status": "accepted", "chapter": 3},
+            "state_deltas": [],
+            "entity_deltas": [],
+            "accepted_events": [
+                {
+                    "event_id": "evt-001",
+                    "chapter": 3,
+                    "event_type": "open_loop_created",
+                    "subject": "三年之约",
+                    "payload": {"content": "三年之约"},
+                }
+            ],
+        }
+    )
+
+    store = ScratchpadManager(cfg)
+    loops = store.query(category="open_loop", status="active")
+    assert result["applied"] is True
+    assert any("三年之约" in x.subject for x in loops)

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

@@ -34,6 +34,7 @@ REGISTERED_CLI_SUBCOMMANDS = {
     "index", "state", "rag", "style", "entity", "context", "memory",
     "migrate", "status", "update-state", "backup", "archive",
     "init", "extract-context", "memory-contract", "review-pipeline",
+    "story-system", "chapter-commit", "story-events",
 }
 
 
@@ -254,3 +255,9 @@ def test_webnovel_review_skill_uses_unified_reviewer_pipeline():
         assert legacy_agent not in skill_text
 
     assert " workflow " not in skill_text
+
+
+def test_story_system_runtime_contract_commands_exist():
+    text = (SKILLS_DIR / "webnovel-write" / "SKILL.md").read_text(encoding="utf-8")
+    block = re.search(r"story-system[\s\S]+--emit-runtime-contracts[\s\S]+REVIEW_CONTRACT", text)
+    assert block, "webnovel-write skill 必须包含生成 runtime contracts 的完整步骤块"

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

@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import json
+
+from data_modules.runtime_contract_builder import RuntimeContractBuilder
+
+
+def test_runtime_contract_builder_creates_volume_and_review_contracts(tmp_path):
+    project_root = tmp_path
+    (project_root / ".webnovel").mkdir(parents=True, exist_ok=True)
+    (project_root / ".webnovel" / "state.json").write_text(
+        json.dumps(
+            {
+                "progress": {"volumes_planned": [{"volume": 1, "chapters_range": "1-20"}]},
+                "chapter_meta": {},
+                "disambiguation_pending": [],
+                "disambiguation_warnings": [],
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+    (project_root / ".story-system" / "MASTER_SETTING.json").parent.mkdir(parents=True, exist_ok=True)
+    (project_root / ".story-system" / "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": {"locked": ["route.primary_genre"], "append_only": ["anti_patterns"], "override_allowed": []},
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+    (project_root / ".story-system" / "anti_patterns.json").write_text(
+        json.dumps([{"text": "配角不能抢主角兑现"}], ensure_ascii=False),
+        encoding="utf-8",
+    )
+    (project_root / "大纲").mkdir(parents=True, exist_ok=True)
+    (project_root / "大纲" / "第1卷-详细大纲.md").write_text(
+        "### 第3章：试压\nCBN：继续压迫\n必须覆盖节点：发现陷阱、决定隐忍\n本章禁区：不可提前摊牌",
+        encoding="utf-8",
+    )
+
+    builder = RuntimeContractBuilder(project_root)
+    volume_brief, review_contract = builder.build_for_chapter(3)
+
+    assert volume_brief["meta"]["contract_type"] == "VOLUME_BRIEF"
+    assert review_contract["meta"]["contract_type"] == "REVIEW_CONTRACT"
+    assert "发现陷阱" in review_contract["must_check"]
+    assert "不可提前摊牌" in review_contract["blocking_rules"]

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

@@ -0,0 +1,60 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import pytest
+
+from data_modules.story_contract_schema import ChapterBrief, MasterSetting, ReviewContract, VolumeBrief
+
+
+def test_master_setting_and_chapter_brief_accept_phase1_seed_shape():
+    master = MasterSetting.model_validate(
+        {
+            "meta": {"schema_version": "story-system/v1", "contract_type": "MASTER_SETTING"},
+            "route": {"primary_genre": "玄幻退婚流"},
+            "master_constraints": {"core_tone": "先压后爆", "pacing_strategy": "三章内首个反打"},
+            "base_context": [],
+            "source_trace": [],
+            "override_policy": {"locked": ["route.primary_genre"], "append_only": ["anti_patterns"], "override_allowed": []},
+        }
+    )
+    chapter = ChapterBrief.model_validate(
+        {
+            "meta": {"schema_version": "story-system/v1", "contract_type": "CHAPTER_BRIEF"},
+            "override_allowed": {"chapter_focus": "退婚现场反打"},
+            "dynamic_context": [],
+            "source_trace": [],
+        }
+    )
+    assert master.route["primary_genre"] == "玄幻退婚流"
+    assert chapter.override_allowed["chapter_focus"] == "退婚现场反打"
+
+
+def test_volume_brief_requires_selected_fields():
+    payload = {
+        "meta": {"schema_version": "story-system/v1", "contract_type": "VOLUME_BRIEF"},
+        "volume_goal": {"summary": "卷一站稳脚跟"},
+        "selected_tropes": ["退婚反击"],
+        "selected_pacing": {"wave": "压抑后爆"},
+        "selected_scenes": ["宗门大厅", "资源争夺"],
+        "anti_patterns": ["配角抢主角兑现"],
+        "system_constraints": ["金手指每日限一次"],
+        "overrides": {"locked": {}, "append_only": {}, "override_allowed": {}},
+    }
+    model = VolumeBrief.model_validate(payload)
+    assert model.volume_goal["summary"] == "卷一站稳脚跟"
+
+
+def test_review_contract_requires_blocking_rules_list():
+    with pytest.raises(Exception):
+        ReviewContract.model_validate(
+            {
+                "meta": {"schema_version": "story-system/v1", "contract_type": "REVIEW_CONTRACT"},
+                "must_check": ["mandatory_nodes"],
+                "blocking_rules": "not-a-list",
+                "genre_specific_risks": [],
+                "anti_patterns": [],
+                "system_constraints": [],
+                "review_thresholds": {"blocking_count": 0},
+                "overrides": {"locked": {}, "append_only": {}, "override_allowed": {}},
+            }
+        )

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

@@ -0,0 +1,75 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import json
+
+import pytest
+
+from data_modules.story_contracts import (
+    StoryContractPaths,
+    merge_anti_patterns,
+    merge_contract_layers,
+    read_json_if_exists,
+)
+
+
+def test_story_contract_paths_resolve_expected_locations(tmp_path):
+    paths = StoryContractPaths.from_project_root(tmp_path)
+
+    assert paths.root == tmp_path.resolve() / ".story-system"
+    assert paths.master_json == paths.root / "MASTER_SETTING.json"
+    assert paths.anti_patterns_json == paths.root / "anti_patterns.json"
+    assert paths.chapter_json(7) == paths.root / "chapters" / "chapter_007.json"
+
+
+def test_merge_contract_layers_preserves_locked_and_merges_append_only():
+    merged = merge_contract_layers(
+        {
+            "locked": {"core_tone": "先压后爆"},
+            "append_only": {"anti_patterns": ["配角连续抢戏超过 300 字"]},
+            "override_allowed": {"scene_focus": "退婚当场反杀"},
+        },
+        {
+            "append_only": {"anti_patterns": ["本章禁止解释性旁白"]},
+            "override_allowed": {"chapter_focus": "退婚当场反杀"},
+        },
+    )
+
+    assert merged["locked"]["core_tone"] == "先压后爆"
+    assert merged["append_only"]["anti_patterns"] == [
+        "配角连续抢戏超过 300 字",
+        "本章禁止解释性旁白",
+    ]
+    assert merged["override_allowed"]["scene_focus"] == "退婚当场反杀"
+    assert merged["override_allowed"]["chapter_focus"] == "退婚当场反杀"
+
+
+def test_merge_anti_patterns_deduplicates_by_text():
+    rows = merge_anti_patterns(
+        [{"text": "打脸节奏不能缺补刀", "source_table": "题材与调性推理", "source_id": "GR-001"}],
+        [{"text": "打脸节奏不能缺补刀", "source_table": "爽点与节奏", "source_id": "PA-002"}],
+    )
+
+    assert [item["text"] for item in rows] == ["打脸节奏不能缺补刀"]
+    assert rows[0]["source_table"] == "题材与调性推理"
+
+
+def test_read_json_if_exists_returns_none_for_missing_file(tmp_path):
+    assert read_json_if_exists(tmp_path / "missing.json") is None
+
+
+def test_read_json_if_exists_raises_value_error_with_path(tmp_path):
+    bad_path = tmp_path / "bad.json"
+    bad_path.write_text("{bad json", encoding="utf-8")
+
+    with pytest.raises(ValueError) as exc:
+        read_json_if_exists(bad_path)
+
+    assert str(bad_path) in str(exc.value)
+
+
+def test_read_json_if_exists_loads_valid_json(tmp_path):
+    path = tmp_path / "payload.json"
+    path.write_text(json.dumps({"ok": True}, ensure_ascii=False), encoding="utf-8")
+
+    assert read_json_if_exists(path) == {"ok": True}

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

@@ -0,0 +1,32 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import pytest
+
+from data_modules.story_event_schema import StoryEvent
+
+
+def test_story_event_supports_power_breakthrough():
+    event = StoryEvent.model_validate(
+        {
+            "event_id": "evt-001",
+            "chapter": 3,
+            "event_type": "power_breakthrough",
+            "subject": "xiaoyan",
+            "payload": {"from": "斗之气三段", "to": "斗者"},
+        }
+    )
+    assert event.event_type == "power_breakthrough"
+
+
+def test_story_event_rejects_unknown_event_type():
+    with pytest.raises(ValueError):
+        StoryEvent.model_validate(
+            {
+                "event_id": "evt-002",
+                "chapter": 3,
+                "event_type": "unknown_event",
+                "subject": "xiaoyan",
+                "payload": {},
+            }
+        )

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

@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import csv
+import json
+import sys
+
+
+def _write_csv(path, headers, rows):
+    with open(path, "w", encoding="utf-8-sig", newline="") as f:
+        writer = csv.DictWriter(f, fieldnames=headers)
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+def test_story_system_persist_writes_master_chapter_and_anti_patterns(tmp_path, monkeypatch):
+    project_root = tmp_path / "book"
+    (project_root / ".webnovel").mkdir(parents=True, exist_ok=True)
+    (project_root / ".webnovel" / "state.json").write_text("{}", encoding="utf-8")
+
+    csv_dir = tmp_path / "csv"
+    csv_dir.mkdir()
+    _write_csv(
+        csv_dir / "题材与调性推理.csv",
+        [
+            "编号", "适用技能", "分类", "层级", "关键词", "意图与同义词", "适用题材",
+            "大模型指令", "核心摘要", "详细展开", "题材/流派", "题材别名", "核心调性",
+            "节奏策略", "强制禁忌/毒点", "推荐基础检索表", "推荐动态检索表", "默认查询词",
+        ],
+        [
+            {
+                "编号": "GR-001",
+                "适用技能": "write",
+                "分类": "题材路由",
+                "层级": "知识补充",
+                "关键词": "玄幻退婚流",
+                "意图与同义词": "退婚流",
+                "适用题材": "玄幻",
+                "大模型指令": "先压后爆",
+                "核心摘要": "退婚起手",
+                "详细展开": "",
+                "题材/流派": "玄幻退婚流",
+                "题材别名": "退婚流",
+                "核心调性": "先压后爆",
+                "节奏策略": "三章内反打",
+                "强制禁忌/毒点": "打脸不能软收尾",
+                "推荐基础检索表": "命名规则",
+                "推荐动态检索表": "桥段套路",
+                "默认查询词": "退婚|打脸",
+            }
+        ],
+    )
+    _write_csv(csv_dir / "命名规则.csv", ["编号", "适用技能", "分类", "层级", "关键词", "适用题材", "核心摘要"], [])
+    _write_csv(csv_dir / "桥段套路.csv", ["编号", "适用技能", "分类", "层级", "关键词", "适用题材", "核心摘要", "忌讳写法"], [])
+
+    from story_system import main
+
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "story_system",
+            "玄幻退婚流",
+            "--project-root",
+            str(project_root),
+            "--chapter",
+            "1",
+            "--persist",
+            "--csv-dir",
+            str(csv_dir),
+            "--format",
+            "both",
+        ],
+    )
+    main()
+
+    story_root = project_root / ".story-system"
+    assert (story_root / "MASTER_SETTING.json").is_file()
+    assert (story_root / "MASTER_SETTING.md").is_file()
+    assert (story_root / "anti_patterns.json").is_file()
+    assert (story_root / "chapters" / "chapter_001.json").is_file()
+    assert (story_root / "chapters" / "chapter_001.md").is_file()
+
+    payload = json.loads((story_root / "MASTER_SETTING.json").read_text(encoding="utf-8"))
+    assert payload["route"]["primary_genre"] == "玄幻退婚流"
+
+
+def test_markdown_writer_preserves_manual_notes_outside_markers(tmp_path):
+    from data_modules.story_contracts import write_marked_markdown
+
+    target = tmp_path / "MASTER_SETTING.md"
+    target.write_text(
+        "# 手工说明\n手工备注\n<!-- STORY-SYSTEM:BEGIN -->\n旧内容\n<!-- STORY-SYSTEM:END -->\n",
+        encoding="utf-8",
+    )
+
+    write_marked_markdown(target, "## Auto\n新内容\n")
+
+    text = target.read_text(encoding="utf-8")
+    assert "# 手工说明" in text
+    assert "手工备注" in text
+    assert "## Auto" in text
+    assert "旧内容" not in text

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

@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import csv
+
+from data_modules.story_system_engine import StorySystemEngine
+
+
+def _write_csv(path, headers, rows):
+    with open(path, "w", encoding="utf-8-sig", newline="") as f:
+        writer = csv.DictWriter(f, fieldnames=headers)
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+def test_story_system_routes_explicit_genre_and_collects_anti_patterns(tmp_path):
+    csv_dir = tmp_path / "csv"
+    csv_dir.mkdir()
+
+    _write_csv(
+        csv_dir / "题材与调性推理.csv",
+        [
+            "编号", "适用技能", "分类", "层级", "关键词", "意图与同义词", "适用题材",
+            "大模型指令", "核心摘要", "详细展开", "题材/流派", "题材别名", "核心调性",
+            "节奏策略", "强制禁忌/毒点", "推荐基础检索表", "推荐动态检索表", "默认查询词",
+        ],
+        [
+            {
+                "编号": "GR-001",
+                "适用技能": "write|plan",
+                "分类": "题材路由",
+                "层级": "知识补充",
+                "关键词": "玄幻退婚流|退婚打脸",
+                "意图与同义词": "退婚流|废材逆袭",
+                "适用题材": "玄幻",
+                "大模型指令": "先给压抑，再给爆发兑现。",
+                "核心摘要": "玄幻退婚流需要耻辱起手和强兑现。",
+                "详细展开": "",
+                "题材/流派": "玄幻退婚流",
+                "题材别名": "退婚流|废材逆袭",
+                "核心调性": "压抑蓄势后爆裂反击",
+                "节奏策略": "前压后爆，三章内必须首个反打",
+                "强制禁忌/毒点": "打脸节奏不能缺最后一拍补刀|配角不能压过主角兑现",
+                "推荐基础检索表": "命名规则|人设与关系|金手指与设定",
+                "推荐动态检索表": "桥段套路|爽点与节奏|场景写法",
+                "默认查询词": "退婚|打脸|废材逆袭",
+            }
+        ],
+    )
+
+    _write_csv(
+        csv_dir / "桥段套路.csv",
+        ["编号", "适用技能", "分类", "层级", "关键词", "适用题材", "核心摘要", "桥段名称", "忌讳写法"],
+        [
+            {
+                "编号": "TR-001",
+                "适用技能": "write",
+                "分类": "桥段",
+                "层级": "知识补充",
+                "关键词": "退婚|打脸",
+                "适用题材": "玄幻",
+                "核心摘要": "退婚现场要给足羞辱和反击空间",
+                "桥段名称": "退婚反击",
+                "忌讳写法": "主角还没反打就被配角替他出手",
+            }
+        ],
+    )
+
+    _write_csv(
+        csv_dir / "爽点与节奏.csv",
+        ["编号", "适用技能", "分类", "层级", "关键词", "适用题材", "核心摘要", "常见崩盘误区", "节奏类型"],
+        [
+            {
+                "编号": "PA-001",
+                "适用技能": "write",
+                "分类": "节奏",
+                "层级": "知识补充",
+                "关键词": "打脸|兑现",
+                "适用题材": "玄幻",
+                "核心摘要": "兑现必须补刀",
+                "常见崩盘误区": "打脸收尾太软，没有读者情绪补刀",
+                "节奏类型": "爆发期",
+            }
+        ],
+    )
+
+    engine = StorySystemEngine(csv_dir=csv_dir)
+    contract = engine.build(query="玄幻退婚流", genre=None, chapter=None)
+
+    assert contract["master_setting"]["route"]["primary_genre"] == "玄幻退婚流"
+    assert contract["master_setting"]["master_constraints"]["core_tone"] == "压抑蓄势后爆裂反击"
+    assert "命名规则" in contract["master_setting"]["route"]["recommended_base_tables"]
+    assert {
+        item["text"] for item in contract["anti_patterns"]
+    } >= {
+        "打脸节奏不能缺最后一拍补刀",
+        "主角还没反打就被配角替他出手",
+        "打脸收尾太软，没有读者情绪补刀",
+    }
+
+
+def test_story_system_falls_back_to_explicit_genre(tmp_path):
+    csv_dir = tmp_path / "csv"
+    csv_dir.mkdir()
+
+    _write_csv(
+        csv_dir / "题材与调性推理.csv",
+        [
+            "编号", "适用技能", "分类", "层级", "关键词", "意图与同义词", "适用题材",
+            "大模型指令", "核心摘要", "详细展开", "题材/流派", "题材别名", "核心调性",
+            "节奏策略", "强制禁忌/毒点", "推荐基础检索表", "推荐动态检索表", "默认查询词",
+        ],
+        [],
+    )
+
+    engine = StorySystemEngine(csv_dir=csv_dir)
+    contract = engine.build(query="压抑一点，后面爆", genre="现言", chapter=None)
+
+    assert contract["master_setting"]["route"]["primary_genre"] == "现言"
+    assert contract["master_setting"]["route"]["route_source"] == "explicit_genre_fallback"
+    assert contract["master_setting"]["route"]["recommended_dynamic_tables"] == ["桥段套路", "爽点与节奏", "场景写法"]

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

@@ -93,6 +93,127 @@ def test_extract_context_forwards_with_resolved_project_root(monkeypatch, tmp_pa
     ]
 
 
+def test_webnovel_story_system_forwards_with_resolved_project_root(monkeypatch, tmp_path):
+    module = _load_webnovel_module()
+
+    book_root = (tmp_path / "book").resolve()
+    called = {}
+
+    def _fake_resolve(explicit_project_root=None):
+        return book_root
+
+    def _fake_run_script(script_name, argv):
+        called["script_name"] = script_name
+        called["argv"] = list(argv)
+        return 0
+
+    monkeypatch.setattr(module, "_resolve_root", _fake_resolve)
+    monkeypatch.setattr(module, "_run_script", _fake_run_script)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "webnovel",
+            "--project-root",
+            str(tmp_path),
+            "story-system",
+            "玄幻退婚流",
+        ],
+    )
+
+    with pytest.raises(SystemExit) as exc:
+        module.main()
+
+    assert int(exc.value.code or 0) == 0
+    assert called["script_name"] == "story_system.py"
+    assert called["argv"][:2] == ["--project-root", str(book_root)]
+
+
+def test_webnovel_story_system_runtime_forwards(monkeypatch, tmp_path):
+    module = _load_webnovel_module()
+
+    project_root = (tmp_path / "book").resolve()
+    called = {}
+
+    def _fake_resolve(explicit_project_root=None):
+        return project_root
+
+    def _fake_run_script(script_name, argv):
+        called["script_name"] = script_name
+        called["argv"] = list(argv)
+        return 0
+
+    monkeypatch.setattr(module, "_resolve_root", _fake_resolve)
+    monkeypatch.setattr(module, "_run_script", _fake_run_script)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "webnovel",
+            "--project-root",
+            str(project_root),
+            "story-system",
+            "玄幻退婚流",
+            "--emit-runtime-contracts",
+        ],
+    )
+
+    with pytest.raises(SystemExit) as exc:
+        module.main()
+
+    assert int(exc.value.code or 0) == 0
+    assert called["script_name"] == "story_system.py"
+    assert "--emit-runtime-contracts" in called["argv"]
+
+
+def test_webnovel_commit_forwards(monkeypatch, tmp_path):
+    module = _load_webnovel_module()
+    project_root = tmp_path / "book"
+    (project_root / ".webnovel").mkdir(parents=True, exist_ok=True)
+    (project_root / ".webnovel" / "state.json").write_text("{}", encoding="utf-8")
+    called = {}
+
+    def _fake_run_script(script_name, argv):
+        called["script_name"] = script_name
+        called["argv"] = list(argv)
+        return 0
+
+    monkeypatch.setattr(module, "_run_script", _fake_run_script)
+    monkeypatch.setattr(sys, "argv", ["webnovel", "--project-root", str(project_root), "chapter-commit", "--chapter", "3"])
+
+    with pytest.raises(SystemExit) as exc:
+        module.main()
+
+    assert int(exc.value.code or 0) == 0
+    assert called["script_name"] == "chapter_commit.py"
+
+
+def test_webnovel_story_events_forwards(monkeypatch, tmp_path):
+    module = _load_webnovel_module()
+    project_root = tmp_path / "book"
+    (project_root / ".webnovel").mkdir(parents=True, exist_ok=True)
+    (project_root / ".webnovel" / "state.json").write_text("{}", encoding="utf-8")
+    called = {}
+
+    def _fake_run_script(script_name, argv):
+        called["script_name"] = script_name
+        called["argv"] = list(argv)
+        return 0
+
+    monkeypatch.setattr(module, "_run_script", _fake_run_script)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        ["webnovel", "--project-root", str(project_root), "story-events", "--chapter", "3"],
+    )
+
+    with pytest.raises(SystemExit) as exc:
+        module.main()
+
+    assert int(exc.value.code or 0) == 0
+    assert called["script_name"] == "story_events.py"
+
+
 def test_preflight_succeeds_for_valid_project_root(monkeypatch, tmp_path, capsys):
     module = _load_webnovel_module()
 

webnovel-writer/scripts/data_modules/webnovel.py

@@ -249,6 +249,21 @@ def main() -> None:
     p_extract_context.add_argument("--chapter", type=int, required=True, help="目标章节号")
     p_extract_context.add_argument("--format", choices=["text", "json"], default="text", help="输出格式")
 
+    p_story_system = sub.add_parser("story-system", help="转发到 story_system.py")
+    p_story_system.add_argument("args", nargs=argparse.REMAINDER)
+
+    p_story_events = sub.add_parser("story-events", help="转发到 story_events.py")
+    p_story_events.add_argument("--chapter", type=int, default=0, help="目标章节号")
+    p_story_events.add_argument("--limit", type=int, default=200, help="查询条数")
+    p_story_events.add_argument("--health", action="store_true", help="输出事件链健康信息")
+
+    p_commit = sub.add_parser("chapter-commit", help="转发到 chapter_commit.py")
+    p_commit.add_argument("--chapter", type=int, required=True, help="目标章节号")
+    p_commit.add_argument("--review-result", default="", help="review_result JSON 文件")
+    p_commit.add_argument("--fulfillment-result", default="", help="fulfillment_result JSON 文件")
+    p_commit.add_argument("--disambiguation-result", default="", help="disambiguation_result JSON 文件")
+    p_commit.add_argument("--extraction-result", default="", help="extraction_result JSON 文件")
+
     p_memory_contract = sub.add_parser("memory-contract", help="转发到 memory_cli.py")
     p_memory_contract.add_argument("args", nargs=argparse.REMAINDER)
 
@@ -312,6 +327,26 @@ def main() -> None:
     if tool == "extract-context":
         return_args = [*forward_args, "--chapter", str(args.chapter), "--format", str(args.format)]
         raise SystemExit(_run_script("extract_chapter_context.py", return_args))
+    if tool == "story-system":
+        raise SystemExit(_run_script("story_system.py", [*forward_args, *rest]))
+    if tool == "story-events":
+        return_args = [*forward_args, "--limit", str(args.limit)]
+        if args.chapter:
+            return_args.extend(["--chapter", str(args.chapter)])
+        if args.health:
+            return_args.append("--health")
+        raise SystemExit(_run_script("story_events.py", return_args))
+    if tool == "chapter-commit":
+        return_args = [*forward_args, "--chapter", str(args.chapter)]
+        if args.review_result:
+            return_args.extend(["--review-result", str(args.review_result)])
+        if args.fulfillment_result:
+            return_args.extend(["--fulfillment-result", str(args.fulfillment_result)])
+        if args.disambiguation_result:
+            return_args.extend(["--disambiguation-result", str(args.disambiguation_result)])
+        if args.extraction_result:
+            return_args.extend(["--extraction-result", str(args.extraction_result)])
+        raise SystemExit(_run_script("chapter_commit.py", return_args))
     if tool == "memory-contract":
         raise SystemExit(_run_script("memory_cli.py", [*forward_args, *rest]))
     if tool == "review-pipeline":

webnovel-writer/scripts/extract_chapter_context.py

@@ -311,6 +311,8 @@ def _load_contract_context(project_root: Path, chapter_num: int) -> Dict[str, An
     return {
         "context_contract_version": (payload.get("meta") or {}).get("context_contract_version"),
         "context_weight_stage": (payload.get("meta") or {}).get("context_weight_stage"),
+        "story_contract": (sections.get("story_contract") or {}).get("content", {}),
+        "prewrite_validation": (sections.get("prewrite_validation") or {}).get("content", {}),
         "reader_signal": (sections.get("reader_signal") or {}).get("content", {}),
         "genre_profile": (sections.get("genre_profile") or {}).get("content", {}),
         "writing_guidance": (sections.get("writing_guidance") or {}).get("content", {}),
@@ -340,6 +342,8 @@ def build_chapter_context_payload(project_root: Path, chapter_num: int) -> Dict[
         "state_summary": state_summary,
         "context_contract_version": contract_context.get("context_contract_version"),
         "context_weight_stage": contract_context.get("context_weight_stage"),
+        "story_contract": contract_context.get("story_contract", {}),
+        "prewrite_validation": contract_context.get("prewrite_validation", {}),
         "reader_signal": contract_context.get("reader_signal", {}),
         "genre_profile": contract_context.get("genre_profile", {}),
         "writing_guidance": contract_context.get("writing_guidance", {}),
@@ -385,6 +389,26 @@ def _render_text(payload: Dict[str, Any]) -> str:
             lines.append(f"- 上下文阶段权重: {stage}")
             lines.append("")
 
+    story_contract = payload.get("story_contract") or {}
+    review_contract = story_contract.get("review_contract") or {}
+    prewrite_validation = payload.get("prewrite_validation") or {}
+    if review_contract or prewrite_validation:
+        lines.append("## Contract-First Runtime")
+        lines.append("")
+        lines.append(
+            f"- Review blocking rules: {len(review_contract.get('blocking_rules') or [])}"
+        )
+        lines.append(f"- Prewrite blocking: {prewrite_validation.get('blocking')}")
+        forbidden_zones = prewrite_validation.get("forbidden_zones") or []
+        if forbidden_zones:
+            lines.append(f"- Forbidden zones: {len(forbidden_zones)}")
+        planned_nodes = (
+            (prewrite_validation.get("fulfillment_seed") or {}).get("planned_nodes") or []
+        )
+        if planned_nodes:
+            lines.append(f"- Planned nodes: {len(planned_nodes)}")
+        lines.append("")
+
     plot_structure = payload.get("plot_structure") or {}
     if plot_structure:
         lines.append("## 情节结构")

webnovel-writer/scripts/story_events.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+from runtime_compat import enable_windows_utf8_stdio
+
+from data_modules.event_log_store import EventLogStore
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Story events CLI")
+    parser.add_argument("--project-root", required=True)
+    parser.add_argument("--chapter", type=int, default=0)
+    parser.add_argument("--limit", type=int, default=200)
+    parser.add_argument("--health", action="store_true")
+    args = parser.parse_args()
+
+    store = EventLogStore(Path(args.project_root))
+
+    if args.health:
+        print(json.dumps(store.health(), ensure_ascii=False))
+        return
+
+    if args.chapter:
+        print(
+            json.dumps(
+                {"chapter": args.chapter, "events": store.read_events(args.chapter)},
+                ensure_ascii=False,
+            )
+        )
+        return
+
+    print(json.dumps({"events": store.list_recent(limit=args.limit)}, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    if sys.platform == "win32":
+        enable_windows_utf8_stdio()
+    main()

webnovel-writer/scripts/story_system.py

@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+from runtime_compat import enable_windows_utf8_stdio
+
+from data_modules.runtime_contract_builder import RuntimeContractBuilder
+from data_modules.story_contracts import persist_runtime_contracts, persist_story_seed
+from data_modules.story_system_engine import StorySystemEngine
+
+
+def _default_csv_dir() -> Path:
+    return Path(__file__).resolve().parent.parent / "references" / "csv"
+
+
+def _resolve_project_root(raw: str) -> Path:
+    if raw:
+        return Path(raw).expanduser().resolve()
+
+    from project_locator import resolve_project_root
+
+    return resolve_project_root()
+
+
+def _render_output(format_name: str, contract: dict) -> str:
+    if format_name == "json":
+        return json.dumps(contract, ensure_ascii=False, indent=2)
+    if format_name == "markdown":
+        lines = [
+            "# Story System",
+            f"- 题材：{contract['master_setting']['route'].get('primary_genre', '')}",
+        ]
+        if contract.get("chapter_brief"):
+            lines.append(
+                f"- 章节焦点：{contract['chapter_brief']['override_allowed'].get('chapter_focus', '')}"
+            )
+        return "\n".join(lines)
+    return json.dumps(
+        {
+            "master": contract["master_setting"].get("route", {}),
+            "chapter": (contract.get("chapter_brief") or {}).get("override_allowed", {}),
+            "anti_patterns": [row.get("text", "") for row in contract.get("anti_patterns", [])],
+        },
+        ensure_ascii=False,
+        indent=2,
+    )
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Story system seed generator")
+    parser.add_argument("query", help="题材 / 需求描述")
+    parser.add_argument("--project-root", default="")
+    parser.add_argument("--genre", default="")
+    parser.add_argument("--chapter", type=int, default=0)
+    parser.add_argument("--persist", action="store_true")
+    parser.add_argument("--emit-runtime-contracts", action="store_true")
+    parser.add_argument("--csv-dir", default="")
+    parser.add_argument("--format", choices=["json", "markdown", "both"], default="json")
+
+    args = parser.parse_args()
+    project_root = _resolve_project_root(args.project_root)
+    csv_dir = Path(args.csv_dir).expanduser().resolve() if args.csv_dir else _default_csv_dir()
+    engine = StorySystemEngine(csv_dir=csv_dir)
+    contract = engine.build(
+        query=args.query,
+        genre=args.genre or None,
+        chapter=args.chapter or None,
+    )
+
+    if args.persist:
+        persist_story_seed(
+            project_root=project_root,
+            master_payload=contract["master_setting"],
+            chapter_payload=contract.get("chapter_brief"),
+            anti_patterns=contract["anti_patterns"],
+        )
+    if args.emit_runtime_contracts:
+        if not args.chapter:
+            raise ValueError("--emit-runtime-contracts 需要 --chapter")
+        volume_brief, review_contract = RuntimeContractBuilder(project_root).build_for_chapter(args.chapter)
+        persist_runtime_contracts(project_root, args.chapter, volume_brief, review_contract)
+
+    print(_render_output(args.format, contract))
+
+
+if __name__ == "__main__":
+    if sys.platform == "win32":
+        enable_windows_utf8_stdio()
+    main()

webnovel-writer/skills/webnovel-plan/SKILL.md

@@ -53,6 +53,13 @@ export SCRIPTS_DIR="${CLAUDE_PLUGIN_ROOT}/scripts"
 export PROJECT_ROOT="$(python "${SCRIPTS_DIR}/webnovel.py" --project-root "${WORKSPACE_ROOT}" where)"
 ```
 
+若本次规划会直接落到具体章节，还必须先刷新 Story System runtime 合同：
+
+```bash
+python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "${WORKSPACE_ROOT}" \
+  story-system "{chapter_goal}" --chapter {chapter_num} --persist --emit-runtime-contracts --format both
+```
+
 ## 引用加载策略
 
 ### md 必读
@@ -61,17 +68,17 @@ export PROJECT_ROOT="$(python "${SCRIPTS_DIR}/webnovel.py" --project-root "${WOR
 |------|---------|-----------|
 | Step 4 | always | `templates/output/大纲-卷节拍表.md` |
 | Step 5 | always | `templates/output/大纲-卷时间线.md` |
-| Step 6 | always | `references/genre-profiles.md` |
-| Step 6 | always | `references/shared/strand-weave-pattern.md` |
-| 章纲拆分 | always | `references/outlining/plot-signal-vs-spoiler.md` |
+| Step 6 | always | `../../references/genre-profiles.md` |
+| Step 6 | always | `../../references/shared/strand-weave-pattern.md` |
+| 章纲拆分 | always | `../../references/outlining/plot-signal-vs-spoiler.md` |
 
 ### md 按需
 
 | Step | Trigger | Reference |
 |------|---------|-----------|
-| Step 6 | 需要爽点设计 | `references/shared/cool-points-guide.md` |
+| Step 6 | 需要爽点设计 | `../../references/shared/cool-points-guide.md` |
 | Step 6/7 | 需要冲突设计 | `references/outlining/conflict-design.md` |
-| Step 7 | 需要追读力分析 | `references/reading-power-taxonomy.md` |
+| Step 7 | 需要追读力分析 | `../../references/reading-power-taxonomy.md` |
 | Step 7 | 需要章纲细化 | `references/outlining/chapter-planning.md` |
 | Step 6/7 | 特定题材节奏 | `references/outlining/genre-volume-pacing.md` |
 

webnovel-writer/skills/webnovel-review/SKILL.md

@@ -47,6 +47,13 @@ export SCRIPTS_DIR="${CLAUDE_PLUGIN_ROOT}/scripts"
 export PROJECT_ROOT="$(python "${SCRIPTS_DIR}/webnovel.py" --project-root "${WORKSPACE_ROOT}" where)"
 ```
 
+若目标章缺少 runtime 合同，先补齐：
+
+```bash
+python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "${WORKSPACE_ROOT}" \
+  story-system "{chapter_goal}" --chapter {chapter_num} --persist --emit-runtime-contracts --format both
+```
+
 要求：
 - `PROJECT_ROOT` 必须包含 `.webnovel/state.json`
 - 任一关键目录不存在时立即阻断
@@ -57,17 +64,17 @@ export PROJECT_ROOT="$(python "${SCRIPTS_DIR}/webnovel.py" --project-root "${WOR
 
 | Trigger | Reference |
 |---------|-----------|
-| always | `references/shared/core-constraints.md` |
-| always | `references/review-schema.md` |
+| always | `../../references/shared/core-constraints.md` |
+| always | `../../references/review-schema.md` |
 
 #### md 按需
 
 | Trigger | Reference |
 |---------|-----------|
-| 审查涉及爽点或钩子分析 | `references/shared/cool-points-guide.md` |
-| 审查涉及多线交织 | `references/shared/strand-weave-pattern.md` |
-| ai_flavor issue ≥ 3 | `skills/webnovel-write/references/anti-ai-guide.md` |
-| blocking issue 需用户决策 (Step 6) | `references/review/blocking-override-guidelines.md` |
+| 审查涉及爽点或钩子分析 | `../../references/shared/cool-points-guide.md` |
+| 审查涉及多线交织 | `../../references/shared/strand-weave-pattern.md` |
+| ai_flavor issue ≥ 3 | `../../skills/webnovel-write/references/anti-ai-guide.md` |
+| blocking issue 需用户决策 (Step 6) | `../../references/review/blocking-override-guidelines.md` |
 
 ### Step 3：加载项目状态与待审正文
 

webnovel-writer/skills/webnovel-write/SKILL.md

@@ -135,6 +135,22 @@ export PROJECT_ROOT="$(python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-roo
 - `preflight` 必须成功。
 - 任一核心输入缺失立即阻断。
 
+### 准备阶段：生成 Story System runtime 合同
+
+在进入 Step 0.5 之前，必须先生成并刷新本章的合同树：
+
+```bash
+python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "${WORKSPACE_ROOT}" \
+  story-system "{chapter_goal}" --chapter {chapter_num} --persist --emit-runtime-contracts --format both
+```
+
+至少确认以下文件已存在：
+- `.story-system/MASTER_SETTING.json`
+- `.story-system/volumes/volume_{volume_num}.json`
+- `.story-system/reviews/chapter_{chapter_num}.review.json`（`REVIEW_CONTRACT`）
+
+若合同缺失或生成失败，直接阻断，不进入正文起草。
+
 ### Step 0.5：轻量节点预检
 
 目的：在不阻断流程的前提下，对章纲中的结构化节点做轻量一致性提醒。