feat: wire contract v2 guidance into chapter context extraction

.claude/agents/context-agent.md

@@ -80,6 +80,14 @@ tools: Read, Grep, Bash
 python -m data_modules.context_manager --chapter {NNNN} --project-root "{project_root}"
 ```
 
+### Step 0.5: Contract v2 上下文包
+```bash
+python "${CLAUDE_PLUGIN_ROOT}/scripts/extract_chapter_context.py" --chapter {NNNN} --project-root "{project_root}" --format json
+```
+
+- 必须读取：`writing_guidance.guidance_items`
+- 推荐读取：`reader_signal` 与 `genre_profile.reference_hints`
+
 ### Step 1: 读取大纲与状态
 - 大纲：`大纲/卷N/第XXX章.md` 或 `大纲/第{卷}卷-详细大纲.md`
   - 若大纲含“反派层级”，必须提取并写入任务书

.claude/scripts/data_modules/tests/test_extract_chapter_context.py

@@ -36,3 +36,81 @@ def test_extract_state_summary_accepts_dominant_key(tmp_path):
     assert "Ch10:quest" in text
     assert "Ch11:fire" in text
 
+
+def test_build_chapter_context_payload_includes_contract_sections(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 build_chapter_context_payload
+    from data_modules.config import DataModulesConfig
+    from data_modules.index_manager import IndexManager, ChapterReadingPowerMeta, ReviewMetrics
+
+    cfg = DataModulesConfig.from_project_root(tmp_path)
+    cfg.ensure_dirs()
+
+    state = {
+        "project": {"genre": "xuanhuan"},
+        "progress": {"current_chapter": 3, "total_words": 9000},
+        "protagonist_state": {
+            "power": {"realm": "筑基", "layer": 2},
+            "location": "宗门",
+            "golden_finger": {"name": "系统", "level": 1},
+        },
+        "strand_tracker": {"history": [{"chapter": 2, "dominant": "quest"}]},
+        "chapter_meta": {},
+        "disambiguation_warnings": [],
+        "disambiguation_pending": [],
+    }
+    (cfg.webnovel_dir / "state.json").write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
+
+    summaries_dir = cfg.webnovel_dir / "summaries"
+    summaries_dir.mkdir(parents=True, exist_ok=True)
+    (summaries_dir / "ch0002.md").write_text("## 剧情摘要\n上一章总结", encoding="utf-8")
+
+    outline_dir = tmp_path / "大纲"
+    outline_dir.mkdir(parents=True, exist_ok=True)
+    (outline_dir / "第1卷 详细大纲.md").write_text("### 第3章：测试标题\n测试大纲", encoding="utf-8")
+
+    refs_dir = tmp_path / ".claude" / "references"
+    refs_dir.mkdir(parents=True, exist_ok=True)
+    (refs_dir / "genre-profiles.md").write_text("## xuanhuan\n- 升级线清晰", encoding="utf-8")
+    (refs_dir / "reading-power-taxonomy.md").write_text("## xuanhuan\n- 悬念钩优先", encoding="utf-8")
+
+    idx = IndexManager(cfg)
+    idx.save_chapter_reading_power(
+        ChapterReadingPowerMeta(chapter=2, hook_type="悬念钩", hook_strength="strong", coolpoint_patterns=["身份掉马"])
+    )
+    idx.save_review_metrics(
+        ReviewMetrics(start_chapter=1, end_chapter=2, overall_score=71, dimension_scores={"plot": 71})
+    )
+
+    payload = build_chapter_context_payload(tmp_path, 3)
+    assert payload["context_contract_version"] == "v2"
+    assert "writing_guidance" in payload
+    assert isinstance(payload["writing_guidance"].get("guidance_items"), list)
+    assert payload["genre_profile"].get("genre") == "xuanhuan"
+
+
+def test_render_text_contains_writing_guidance_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": 10,
+        "outline": "测试大纲",
+        "previous_summaries": ["### 第9章摘要\n上一章"],
+        "state_summary": "状态",
+        "context_contract_version": "v2",
+        "reader_signal": {"review_trend": {"overall_avg": 72}, "low_score_ranges": [{"start_chapter": 8, "end_chapter": 9}]},
+        "genre_profile": {"genre": "xuanhuan", "reference_hints": ["升级线清晰"]},
+        "writing_guidance": {"guidance_items": ["先修低分", "钩子差异化"]},
+    }
+
+    text = _render_text(payload)
+    assert "## 写作执行建议" in text
+    assert "先修低分" in text
+    assert "## Contract (v2)" in text

.claude/scripts/extract_chapter_context.py

@@ -1,22 +1,23 @@
 #!/usr/bin/env python3
+# -*- coding: utf-8 -*-
 """
-extract_chapter_context.py - 提取章节创作所需的精简上下文
+extract_chapter_context.py - extract chapter writing context
 
-功能：
-- 提取当前章节的大纲片段（~500字）
-- 提取前2章的摘要（优先 .webnovel/summaries）
-- 提取 state.json 关键字段（~300字）
-
-用法：
-    python extract_chapter_context.py --chapter 7
-    python extract_chapter_context.py --chapter 7 --project-root ./webnovel-project
+Features:
+- chapter outline snippet
+- previous chapter summaries (prefers .webnovel/summaries)
+- compact state summary
+- ContextManager contract sections (reader_signal / genre_profile / writing_guidance)
 """
 
+from __future__ import annotations
+
 import argparse
 import json
 import re
 import sys
 from pathlib import Path
+from typing import Any, Dict, List
 
 try:
     from chapter_paths import find_chapter_file
@@ -24,8 +25,14 @@ except ImportError:  # pragma: no cover
     from scripts.chapter_paths import find_chapter_file
 
 
-def find_project_root(start_path: Path = None) -> Path:
-    """查找包含 .webnovel 目录的项目根目录"""
+def _ensure_scripts_path():
+    scripts_dir = Path(__file__).resolve().parent
+    if str(scripts_dir) not in sys.path:
+        sys.path.insert(0, str(scripts_dir))
+
+
+def find_project_root(start_path: Path | None = None) -> Path:
+    """Find project root containing `.webnovel` directory."""
     if start_path is None:
         start_path = Path.cwd()
 
@@ -43,33 +50,23 @@ def find_project_root(start_path: Path = None) -> Path:
 
 
 def extract_chapter_outline(project_root: Path, chapter_num: int) -> str:
-    """从大纲文件中提取指定章节的大纲片段"""
+    """Extract chapter outline segment from volume outline file."""
     volume_num = (chapter_num - 1) // 50 + 1
-    outline_file = project_root / "大纲" / f"第{volume_num}卷-详细大纲.md"
+    outline_file = project_root / "大纲" / f"第{volume_num}卷 详细大纲.md"
 
     if not outline_file.exists():
         return f"⚠️ 大纲文件不存在: {outline_file}"
 
     content = outline_file.read_text(encoding="utf-8")
 
-    # 匹配章节大纲块
-    # 格式：### 第 N 章：标题 或 ### 第 N 章: 标题
     pattern = rf"###\s*第\s*{chapter_num}\s*章[：:]\s*(.+?)(?=###\s*第\s*\d+\s*章|##\s|$)"
     match = re.search(pattern, content, re.DOTALL)
+    if not match:
+        pattern2 = rf"###\s*第{chapter_num}章[：:]\s*(.+?)(?=###\s*第\d+章|##\s|$)"
+        match = re.search(pattern2, content, re.DOTALL)
 
     if match:
         outline = match.group(0).strip()
-        # 限制长度
-        if len(outline) > 1500:
-            outline = outline[:1500] + "\n...(已截断)"
-        return outline
-
-    # 尝试另一种格式：### 第 1 章：标题（无空格）
-    pattern2 = rf"###\s*第{chapter_num}章[：:]\s*(.+?)(?=###\s*第\d+章|##\s|$)"
-    match2 = re.search(pattern2, content, re.DOTALL)
-
-    if match2:
-        outline = match2.group(0).strip()
         if len(outline) > 1500:
             outline = outline[:1500] + "\n...(已截断)"
         return outline
@@ -78,7 +75,7 @@ def extract_chapter_outline(project_root: Path, chapter_num: int) -> str:
 
 
 def _load_summary_file(project_root: Path, chapter_num: int) -> str:
-    """从 .webnovel/summaries/chNNNN.md 提取剧情摘要"""
+    """Load summary section from `.webnovel/summaries/chNNNN.md`."""
     summary_path = project_root / ".webnovel" / "summaries" / f"ch{chapter_num:04d}.md"
     if not summary_path.exists():
         return ""
@@ -91,89 +88,200 @@ def _load_summary_file(project_root: Path, chapter_num: int) -> str:
 
 
 def extract_chapter_summary(project_root: Path, chapter_num: int) -> str:
-    """提取指定章节摘要（优先 summaries/，再降级正文）"""
+    """Extract chapter summary, fallback to chapter body head."""
     summary = _load_summary_file(project_root, chapter_num)
     if summary:
         return summary
 
     chapter_file = find_chapter_file(project_root, chapter_num)
     if not chapter_file or not chapter_file.exists():
-        return f"⚠️ 第 {chapter_num} 章文件不存在"
+        return f"⚠️ 第{chapter_num}章文件不存在"
 
     content = chapter_file.read_text(encoding="utf-8")
 
-    # 兼容旧格式：尝试提取"本章摘要"部分
     summary_match = re.search(r"##\s*本章摘要\s*\r?\n(.+?)(?=\r?\n##|$)", content, re.DOTALL)
     if summary_match:
         return summary_match.group(1).strip()
 
-    # 如果没有摘要，提取"本章统计"部分
     stats_match = re.search(r"##\s*本章统计\s*\r?\n(.+?)(?=\r?\n##|$)", content, re.DOTALL)
     if stats_match:
         return f"[无摘要，仅统计]\n{stats_match.group(1).strip()}"
 
-    # 最后降级：提取前500字作为摘要
     lines = content.split("\n")
-    text_lines = [l for l in lines if not l.startswith("#") and l.strip()]
+    text_lines = [line for line in lines if not line.startswith("#") and line.strip()]
     text = "\n".join(text_lines)[:500]
     return f"[自动截取前500字]\n{text}..."
 
 
 def extract_state_summary(project_root: Path) -> str:
-    """提取 state.json 的关键字段"""
+    """Extract key fields from `.webnovel/state.json`."""
     state_file = project_root / ".webnovel" / "state.json"
-
     if not state_file.exists():
         return "⚠️ state.json 不存在"
 
     state = json.loads(state_file.read_text(encoding="utf-8"))
+    summary_parts: List[str] = []
 
-    # 提取关键字段
-    summary_parts = []
-
-    # 进度
     if "progress" in state:
-        p = state["progress"]
-        summary_parts.append(f"**进度**: 第 {p.get('current_chapter', '?')} 章 / {p.get('total_words', '?')} 字")
+        progress = state["progress"]
+        summary_parts.append(
+            f"**进度**: 第{progress.get('current_chapter', '?')}章 / {progress.get('total_words', '?')}字"
+        )
 
-    # 主角状态
     if "protagonist_state" in state:
         ps = state["protagonist_state"]
         power = ps.get("power", {})
         summary_parts.append(f"**主角实力**: {power.get('realm', '?')} {power.get('layer', '?')}层")
         summary_parts.append(f"**当前位置**: {ps.get('location', '?')}")
+        golden_finger = ps.get("golden_finger", {})
+        summary_parts.append(
+            f"**金手指**: {golden_finger.get('name', '?')} Lv.{golden_finger.get('level', '?')}"
+        )
 
-        gf = ps.get("golden_finger", {})
-        summary_parts.append(f"**金手指**: {gf.get('name', '?')} Lv.{gf.get('level', '?')}")
-
-    # Strand 追踪
     if "strand_tracker" in state:
-        st = state["strand_tracker"]
-        history = st.get("history", [])[-5:]  # 最近5章
+        tracker = state["strand_tracker"]
+        history = tracker.get("history", [])[-5:]
         if history:
-            items = []
-            for h in history:
-                if not isinstance(h, dict):
+            items: List[str] = []
+            for row in history:
+                if not isinstance(row, dict):
                     continue
-                ch = h.get("chapter", "?")
-                strand = h.get("strand") or h.get("dominant") or "unknown"
-                items.append(f"Ch{ch}:{strand}")
-            strand_str = ", ".join(items)
-            summary_parts.append(f"**近5章Strand**: {strand_str}")
+                chapter = row.get("chapter", "?")
+                strand = row.get("strand") or row.get("dominant") or "unknown"
+                items.append(f"Ch{chapter}:{strand}")
+            if items:
+                summary_parts.append(f"**近5章Strand**: {', '.join(items)}")
 
-    # 活跃伏笔（只显示紧急的）
     plot_threads = state.get("plot_threads", {}) if isinstance(state.get("plot_threads"), dict) else {}
     foreshadowing = plot_threads.get("foreshadowing", [])
     if isinstance(foreshadowing, list) and foreshadowing:
-        active = [f for f in foreshadowing if f.get("status") in {"active", "未回收"}]
-        urgent = [f for f in active if f.get("urgency", 0) > 50]
+        active = [row for row in foreshadowing if row.get("status") in {"active", "未回收"}]
+        urgent = [row for row in active if row.get("urgency", 0) > 50]
         if urgent:
-            urgent_list = [f"{f.get('content', '?')[:30]}... (紧急度:{f.get('urgency')})" for f in urgent[:3]]
+            urgent_list = [
+                f"{row.get('content', '?')[:30]}... (紧急度:{row.get('urgency')})"
+                for row in urgent[:3]
+            ]
             summary_parts.append(f"**紧急伏笔**: {'; '.join(urgent_list)}")
 
     return "\n".join(summary_parts)
 
 
+def _load_contract_context(project_root: Path, chapter_num: int) -> Dict[str, Any]:
+    """Build context via ContextManager and return selected sections."""
+    _ensure_scripts_path()
+    from data_modules.config import DataModulesConfig
+    from data_modules.context_manager import ContextManager
+
+    config = DataModulesConfig.from_project_root(project_root)
+    manager = ContextManager(config)
+    payload = manager.build_context(
+        chapter=chapter_num,
+        template="plot",
+        use_snapshot=True,
+        save_snapshot=True,
+        max_chars=8000,
+    )
+
+    sections = payload.get("sections", {})
+    return {
+        "context_contract_version": (payload.get("meta") or {}).get("context_contract_version"),
+        "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", {}),
+    }
+
+
+def build_chapter_context_payload(project_root: Path, chapter_num: int) -> Dict[str, Any]:
+    """Assemble full chapter context payload for text/json output."""
+    outline = extract_chapter_outline(project_root, chapter_num)
+
+    prev_summaries = []
+    for prev_ch in range(max(1, chapter_num - 2), chapter_num):
+        summary = extract_chapter_summary(project_root, prev_ch)
+        prev_summaries.append(f"### 第{prev_ch}章摘要\n{summary}")
+
+    state_summary = extract_state_summary(project_root)
+    contract_context = _load_contract_context(project_root, chapter_num)
+
+    return {
+        "chapter": chapter_num,
+        "outline": outline,
+        "previous_summaries": prev_summaries,
+        "state_summary": state_summary,
+        "context_contract_version": contract_context.get("context_contract_version"),
+        "reader_signal": contract_context.get("reader_signal", {}),
+        "genre_profile": contract_context.get("genre_profile", {}),
+        "writing_guidance": contract_context.get("writing_guidance", {}),
+    }
+
+
+def _render_text(payload: Dict[str, Any]) -> str:
+    chapter_num = payload.get("chapter")
+    lines: List[str] = []
+
+    lines.append(f"# 第 {chapter_num} 章创作上下文")
+    lines.append("")
+
+    lines.append("## 本章大纲")
+    lines.append("")
+    lines.append(str(payload.get("outline", "")))
+    lines.append("")
+    lines.append("---")
+    lines.append("")
+
+    lines.append("## 前文摘要")
+    lines.append("")
+    for item in payload.get("previous_summaries", []):
+        lines.append(item)
+        lines.append("")
+
+    lines.append("---")
+    lines.append("")
+    lines.append("## 当前状态")
+    lines.append("")
+    lines.append(str(payload.get("state_summary", "")))
+    lines.append("")
+
+    contract_version = payload.get("context_contract_version")
+    if contract_version:
+        lines.append(f"## Contract ({contract_version})")
+        lines.append("")
+
+    writing_guidance = payload.get("writing_guidance") or {}
+    guidance_items = writing_guidance.get("guidance_items") or []
+    if guidance_items:
+        lines.append("## 写作执行建议")
+        lines.append("")
+        for idx, item in enumerate(guidance_items, start=1):
+            lines.append(f"{idx}. {item}")
+        lines.append("")
+
+    reader_signal = payload.get("reader_signal") or {}
+    review_trend = reader_signal.get("review_trend") or {}
+    if review_trend:
+        overall_avg = review_trend.get("overall_avg")
+        lines.append("## 追读信号")
+        lines.append("")
+        lines.append(f"- 最近审查均分: {overall_avg}")
+        low_ranges = reader_signal.get("low_score_ranges") or []
+        if low_ranges:
+            lines.append(f"- 低分区间数: {len(low_ranges)}")
+        lines.append("")
+
+    genre_profile = payload.get("genre_profile") or {}
+    if genre_profile.get("genre"):
+        lines.append("## 题材锚定")
+        lines.append("")
+        lines.append(f"- 题材: {genre_profile.get('genre')}")
+        refs = genre_profile.get("reference_hints") or []
+        for row in refs[:3]:
+            lines.append(f"- {row}")
+        lines.append("")
+
+    return "\n".join(lines).rstrip() + "\n"
+
+
 def main():
     parser = argparse.ArgumentParser(description="提取章节创作所需的精简上下文")
     parser.add_argument("--chapter", type=int, required=True, help="目标章节号")
@@ -183,49 +291,19 @@ def main():
     args = parser.parse_args()
 
     try:
-        if args.project_root:
-            project_root = Path(args.project_root)
-        else:
-            project_root = find_project_root()
-
-        chapter_num = args.chapter
-
-        # 提取各部分
-        outline = extract_chapter_outline(project_root, chapter_num)
-
-        # 提取前2章摘要
-        prev_summaries = []
-        for prev_ch in range(max(1, chapter_num - 2), chapter_num):
-            summary = extract_chapter_summary(project_root, prev_ch)
-            prev_summaries.append(f"### 第 {prev_ch} 章摘要\n{summary}")
-
-        state_summary = extract_state_summary(project_root)
+        project_root = Path(args.project_root) if args.project_root else find_project_root()
+        payload = build_chapter_context_payload(project_root, args.chapter)
 
         if args.format == "json":
-            result = {
-                "chapter": chapter_num,
-                "outline": outline,
-                "previous_summaries": prev_summaries,
-                "state_summary": state_summary,
-            }
-            print(json.dumps(result, ensure_ascii=False, indent=2))
+            print(json.dumps(payload, ensure_ascii=False, indent=2))
         else:
-            print(f"# 第 {chapter_num} 章创作上下文\n")
-            print("## 本章大纲\n")
-            print(outline)
-            print("\n---\n")
-            print("## 前文摘要\n")
-            for s in prev_summaries:
-                print(s)
-                print()
-            print("---\n")
-            print("## 当前状态\n")
-            print(state_summary)
-
-    except Exception as e:
-        print(f"❌ 错误: {e}", file=sys.stderr)
+            print(_render_text(payload), end="")
+
+    except Exception as exc:
+        print(f"❌ 错误: {exc}", file=sys.stderr)
         sys.exit(1)
 
 
 if __name__ == "__main__":
     main()
+

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

@@ -43,6 +43,15 @@ allowed-tools: Read Write Edit Grep Bash Task
 
 **要求**：创作任务书必须包含“反派层级”（从大纲/章纲提取）。
 
+### Step 1.5: Contract v2 Guidance 注入
+
+```bash
+python "${CLAUDE_PLUGIN_ROOT}/scripts/extract_chapter_context.py" --chapter {chapter_num} --project-root "{PROJECT_ROOT}" --format json
+```
+
+- 必须读取：`writing_guidance.guidance_items`
+- 推荐读取：`reader_signal` 与 `genre_profile.reference_hints`
+
 ## Step 2: 写作
 
 - 遵循三大原则：大纲即法律 / 设定即物理 / 新实体需记录。

README.md

@@ -779,6 +779,12 @@ git checkout ch0045
 - 新增 `writing_guidance`：按章生成可执行写作建议（低分修复/钩子差异化/题材锚定）
 - 新增紧凑文本策略：超预算 section 使用 `…[TRUNCATED]` 保留头尾关键信息
 - 目标：在有限上下文预算下提升“可写性”和“网文感”
+
+### Context Contract v2（阶段 D）
+
+- `extract_chapter_context.py` 已接入 Contract v2 输出
+- JSON 输出新增：`context_contract_version` / `reader_signal` / `genre_profile` / `writing_guidance`
+- text 输出新增：`写作执行建议` 板块，供 Context Agent / Writer 直接使用
 - **invalid_facts 表**：追踪无效事实，支持 pending/confirmed 状态
 - **父子向量索引**：parent_chunk_id 支持摘要-场景层级检索
 - **Token 预算管理**：ContextManager 实现 40%/35%/25% 优先级分配