feat(context): implement Phase F-I scoring, orchestration, dynamic budget, composite genre

.claude/references/context-contract-v2.md

@@ -83,3 +83,27 @@ Phase C:
 - `context_writing_guidance_max_items`
 - `context_writing_guidance_low_score_threshold`
 - `context_writing_guidance_hook_diversify`
+
+Phase E:
+- `context_writing_checklist_enabled`
+- `context_writing_checklist_min_items`
+- `context_writing_checklist_max_items`
+- `context_writing_checklist_default_weight`
+
+Phase F:
+- `context_writing_score_persist_enabled`
+- `context_writing_score_include_reader_trend`
+- `context_writing_score_trend_window`
+- `writing_guidance.checklist_score` 写入 `index.db -> writing_checklist_scores`
+
+Phase H:
+- `context_dynamic_budget_enabled`
+- `context_dynamic_budget_early_chapter`
+- `context_dynamic_budget_late_chapter`
+- 新增 `meta.context_weight_stage`（early/mid/late）
+
+Phase I:
+- `context_genre_profile_support_composite`
+- `context_genre_profile_max_genres`
+- `context_genre_profile_separators`
+- 新增 `genre_profile.genres/composite/composite_hints`

.claude/scripts/data_modules/config.py

@@ -167,6 +167,26 @@ class DataModulesConfig:
     context_writing_checklist_min_items: int = 3
     context_writing_checklist_max_items: int = 6
     context_writing_checklist_default_weight: float = 1.0
+    context_writing_score_persist_enabled: bool = True
+    context_writing_score_include_reader_trend: bool = True
+    context_writing_score_trend_window: int = 10
+    context_dynamic_budget_enabled: bool = True
+    context_dynamic_budget_early_chapter: int = 30
+    context_dynamic_budget_late_chapter: int = 120
+    context_dynamic_budget_early_core_bonus: float = 0.08
+    context_dynamic_budget_early_scene_bonus: float = 0.04
+    context_dynamic_budget_late_global_bonus: float = 0.08
+    context_dynamic_budget_late_scene_penalty: float = 0.06
+    context_genre_profile_support_composite: bool = True
+    context_genre_profile_max_genres: int = 2
+    context_genre_profile_separators: tuple[str, ...] = (
+        "+",
+        "/",
+        "|",
+        ",",
+        "，",
+        "、",
+    )
 
     export_recent_changes_slice: int = 20
     export_disambiguation_slice: int = 20

.claude/scripts/data_modules/context_manager.py

@@ -11,7 +11,7 @@ from pathlib import Path
 from typing import Any, Dict, List, Optional
 
 from .config import get_config
-from .index_manager import IndexManager
+from .index_manager import IndexManager, WritingChecklistScoreMeta
 from .context_ranker import ContextRanker
 from .snapshot_manager import SnapshotManager, SnapshotVersionMismatch
 
@@ -24,6 +24,27 @@ class ContextManager:
         "emotion": {"core": 0.45, "scene": 0.35, "global": 0.20},
         "transition": {"core": 0.50, "scene": 0.25, "global": 0.25},
     }
+
+    TEMPLATE_WEIGHTS_DYNAMIC = {
+        "early": {
+            "plot": {"core": 0.48, "scene": 0.39, "global": 0.13},
+            "battle": {"core": 0.42, "scene": 0.50, "global": 0.08},
+            "emotion": {"core": 0.52, "scene": 0.38, "global": 0.10},
+            "transition": {"core": 0.56, "scene": 0.28, "global": 0.16},
+        },
+        "mid": {
+            "plot": {"core": 0.40, "scene": 0.35, "global": 0.25},
+            "battle": {"core": 0.35, "scene": 0.45, "global": 0.20},
+            "emotion": {"core": 0.45, "scene": 0.35, "global": 0.20},
+            "transition": {"core": 0.50, "scene": 0.25, "global": 0.25},
+        },
+        "late": {
+            "plot": {"core": 0.36, "scene": 0.29, "global": 0.35},
+            "battle": {"core": 0.31, "scene": 0.39, "global": 0.30},
+            "emotion": {"core": 0.41, "scene": 0.29, "global": 0.30},
+            "transition": {"core": 0.46, "scene": 0.21, "global": 0.33},
+        },
+    }
     EXTRA_SECTIONS = {
         "story_skeleton",
         "memory",
@@ -78,8 +99,10 @@ class ContextManager:
         max_chars: Optional[int] = None,
     ) -> Dict[str, Any]:
         template = template or self.DEFAULT_TEMPLATE
+        self._active_template = template
         if template not in self.TEMPLATE_WEIGHTS:
             template = self.DEFAULT_TEMPLATE
+            self._active_template = template
 
         if use_snapshot:
             try:
@@ -107,7 +130,8 @@ class ContextManager:
         template: str = DEFAULT_TEMPLATE,
         max_chars: Optional[int] = None,
     ) -> Dict[str, Any]:
-        weights = self.TEMPLATE_WEIGHTS.get(template, self.TEMPLATE_WEIGHTS[self.DEFAULT_TEMPLATE])
+        chapter = int((pack.get("meta") or {}).get("chapter") or 0)
+        weights = self._resolve_template_weights(template=template, chapter=chapter)
         max_chars = max_chars or 8000
         extra_budget = int(self.config.context_extra_section_budget or 0)
 
@@ -130,6 +154,8 @@ class ContextManager:
 
         assembled["template"] = template
         assembled["weights"] = weights
+        if chapter > 0:
+            assembled.setdefault("meta", {})["context_weight_stage"] = self._resolve_context_stage(chapter)
         return assembled
 
     def filter_invalid_items(self, items: List[Dict[str, Any]], source_type: str, id_key: str) -> List[Dict[str, Any]]:
@@ -260,25 +286,50 @@ class ContextManager:
             return {}
 
         fallback = str(getattr(self.config, "context_genre_profile_fallback", "shuangwen") or "shuangwen")
-        genre = str((state.get("project") or {}).get("genre") or fallback)
+        genre_raw = str((state.get("project") or {}).get("genre") or fallback)
+        genres = self._parse_genre_tokens(genre_raw)
+        if not genres:
+            genres = [fallback]
+        max_genres = max(1, int(getattr(self.config, "context_genre_profile_max_genres", 2)))
+        genres = genres[:max_genres]
+
+        primary_genre = genres[0]
+        secondary_genres = genres[1:]
+        composite = len(genres) > 1
         profile_path = self.config.project_root / ".claude" / "references" / "genre-profiles.md"
         taxonomy_path = self.config.project_root / ".claude" / "references" / "reading-power-taxonomy.md"
 
         profile_text = profile_path.read_text(encoding="utf-8") if profile_path.exists() else ""
         taxonomy_text = taxonomy_path.read_text(encoding="utf-8") if taxonomy_path.exists() else ""
 
-        profile_excerpt = self._extract_genre_section(profile_text, genre)
-        taxonomy_excerpt = self._extract_genre_section(taxonomy_text, genre)
+        profile_excerpt = self._extract_genre_section(profile_text, primary_genre)
+        taxonomy_excerpt = self._extract_genre_section(taxonomy_text, primary_genre)
+
+        secondary_profiles: List[str] = []
+        secondary_taxonomies: List[str] = []
+        for extra in secondary_genres:
+            secondary_profiles.append(self._extract_genre_section(profile_text, extra))
+            secondary_taxonomies.append(self._extract_genre_section(taxonomy_text, extra))
+
         refs = self._extract_markdown_refs(
-            profile_excerpt,
+            "\n".join([profile_excerpt] + secondary_profiles),
             max_items=int(getattr(self.config, "context_genre_profile_max_refs", 8)),
         )
 
+        composite_hints = self._build_composite_genre_hints(genres, refs)
+
         return {
-            "genre": genre,
+            "genre": primary_genre,
+            "genre_raw": genre_raw,
+            "genres": genres,
+            "composite": composite,
+            "secondary_genres": secondary_genres,
             "profile_excerpt": profile_excerpt,
             "taxonomy_excerpt": taxonomy_excerpt,
+            "secondary_profile_excerpts": secondary_profiles,
+            "secondary_taxonomy_excerpts": secondary_taxonomies,
             "reference_hints": refs,
+            "composite_hints": composite_hints,
         }
 
     def _build_writing_guidance(
@@ -334,6 +385,10 @@ class ContextManager:
         if refs:
             guidance.append(f"题材策略可执行提示：{refs[0]}")
 
+        composite_hints = genre_profile.get("composite_hints") or []
+        if composite_hints:
+            guidance.append(f"复合题材协同：{composite_hints[0]}")
+
         if not guidance:
             guidance.append("本章执行默认高可读策略：冲突前置、信息后置、段末留钩。")
 
@@ -344,10 +399,20 @@ class ContextManager:
             genre_profile=genre_profile,
         )
 
+        checklist_score = self._compute_writing_checklist_score(
+            chapter=chapter,
+            checklist=checklist,
+            reader_signal=reader_signal,
+        )
+
+        if getattr(self.config, "context_writing_score_persist_enabled", True):
+            self._persist_writing_checklist_score(checklist_score)
+
         return {
             "chapter": chapter,
             "guidance_items": guidance[:limit],
             "checklist": checklist,
+            "checklist_score": checklist_score,
             "signals_used": {
                 "has_low_score_ranges": bool(low_ranges),
                 "hook_types": list(hook_usage.keys())[:3],
@@ -360,6 +425,181 @@ class ContextManager:
             },
         }
 
+    def _compute_writing_checklist_score(
+        self,
+        chapter: int,
+        checklist: List[Dict[str, Any]],
+        reader_signal: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        total_items = len(checklist)
+        required_items = 0
+        completed_items = 0
+        completed_required = 0
+        total_weight = 0.0
+        completed_weight = 0.0
+        pending_labels: List[str] = []
+
+        for item in checklist:
+            if not isinstance(item, dict):
+                continue
+            required = bool(item.get("required"))
+            weight = float(item.get("weight") or 1.0)
+            total_weight += weight
+            if required:
+                required_items += 1
+
+            completed = self._is_checklist_item_completed(item, reader_signal)
+            if completed:
+                completed_items += 1
+                completed_weight += weight
+                if required:
+                    completed_required += 1
+            else:
+                pending_labels.append(str(item.get("label") or item.get("id") or "未命名项"))
+
+        completion_rate = (completed_items / total_items) if total_items > 0 else 1.0
+        weighted_rate = (completed_weight / total_weight) if total_weight > 0 else completion_rate
+        required_rate = (completed_required / required_items) if required_items > 0 else 1.0
+
+        score = 100.0 * (0.5 * weighted_rate + 0.3 * required_rate + 0.2 * completion_rate)
+
+        if getattr(self.config, "context_writing_score_include_reader_trend", True):
+            trend_window = max(1, int(getattr(self.config, "context_writing_score_trend_window", 10)))
+            trend = self.index_manager.get_writing_checklist_score_trend(last_n=trend_window)
+            baseline = float(trend.get("score_avg") or 0.0)
+            if baseline > 0:
+                score += max(-10.0, min(10.0, (score - baseline) * 0.1))
+
+        score = round(max(0.0, min(100.0, score)), 2)
+
+        return {
+            "chapter": chapter,
+            "score": score,
+            "completion_rate": round(completion_rate, 4),
+            "weighted_completion_rate": round(weighted_rate, 4),
+            "required_completion_rate": round(required_rate, 4),
+            "total_items": total_items,
+            "required_items": required_items,
+            "completed_items": completed_items,
+            "completed_required": completed_required,
+            "total_weight": round(total_weight, 2),
+            "completed_weight": round(completed_weight, 2),
+            "pending_items": pending_labels,
+            "trend_window": int(getattr(self.config, "context_writing_score_trend_window", 10)),
+        }
+
+    def _is_checklist_item_completed(self, item: Dict[str, Any], reader_signal: Dict[str, Any]) -> bool:
+        item_id = str(item.get("id") or "")
+        if item_id in {"fix_low_score_range", "readability_loop"}:
+            review_trend = reader_signal.get("review_trend") or {}
+            overall = review_trend.get("overall_avg")
+            return isinstance(overall, (int, float)) and float(overall) >= 75.0
+
+        if item_id == "hook_diversification":
+            hook_usage = reader_signal.get("hook_type_usage") or {}
+            return len(hook_usage) >= 2
+
+        if item_id == "coolpoint_combo":
+            pattern_usage = reader_signal.get("pattern_usage") or {}
+            return len(pattern_usage) >= 2
+
+        if item_id == "genre_anchor_consistency":
+            return True
+
+        source = str(item.get("source") or "")
+        if source.startswith("fallback"):
+            return True
+
+        return False
+
+    def _persist_writing_checklist_score(self, checklist_score: Dict[str, Any]) -> None:
+        if not checklist_score:
+            return
+        try:
+            self.index_manager.save_writing_checklist_score(
+                WritingChecklistScoreMeta(
+                    chapter=int(checklist_score.get("chapter") or 0),
+                    template=str(getattr(self, "_active_template", self.DEFAULT_TEMPLATE) or self.DEFAULT_TEMPLATE),
+                    total_items=int(checklist_score.get("total_items") or 0),
+                    required_items=int(checklist_score.get("required_items") or 0),
+                    completed_items=int(checklist_score.get("completed_items") or 0),
+                    completed_required=int(checklist_score.get("completed_required") or 0),
+                    total_weight=float(checklist_score.get("total_weight") or 0.0),
+                    completed_weight=float(checklist_score.get("completed_weight") or 0.0),
+                    completion_rate=float(checklist_score.get("completion_rate") or 0.0),
+                    score=float(checklist_score.get("score") or 0.0),
+                    score_breakdown={
+                        "weighted_completion_rate": checklist_score.get("weighted_completion_rate"),
+                        "required_completion_rate": checklist_score.get("required_completion_rate"),
+                        "trend_window": checklist_score.get("trend_window"),
+                    },
+                    pending_items=list(checklist_score.get("pending_items") or []),
+                    source="context_manager",
+                )
+            )
+        except Exception:
+            pass
+
+    def _resolve_context_stage(self, chapter: int) -> str:
+        early = max(1, int(getattr(self.config, "context_dynamic_budget_early_chapter", 30)))
+        late = max(early + 1, int(getattr(self.config, "context_dynamic_budget_late_chapter", 120)))
+        if chapter <= early:
+            return "early"
+        if chapter >= late:
+            return "late"
+        return "mid"
+
+    def _resolve_template_weights(self, template: str, chapter: int) -> Dict[str, float]:
+        template_key = template if template in self.TEMPLATE_WEIGHTS else self.DEFAULT_TEMPLATE
+        base = dict(self.TEMPLATE_WEIGHTS.get(template_key, self.TEMPLATE_WEIGHTS[self.DEFAULT_TEMPLATE]))
+        if not getattr(self.config, "context_dynamic_budget_enabled", True):
+            return base
+
+        stage = self._resolve_context_stage(chapter)
+        staged = self.TEMPLATE_WEIGHTS_DYNAMIC.get(stage, {}).get(template_key)
+        if staged:
+            return dict(staged)
+
+        return base
+
+    def _parse_genre_tokens(self, genre_raw: str) -> List[str]:
+        text = str(genre_raw or "").strip()
+        if not text:
+            return []
+        if not getattr(self.config, "context_genre_profile_support_composite", True):
+            return [text]
+
+        separators = getattr(self.config, "context_genre_profile_separators", ("+", "/", "|", ",", "，", "、"))
+        pattern = "|".join(re.escape(str(token)) for token in separators if str(token))
+        if not pattern:
+            return [text]
+
+        tokens = [chunk.strip() for chunk in re.split(pattern, text) if chunk and chunk.strip()]
+        deduped: List[str] = []
+        seen = set()
+        for token in tokens:
+            lower = token.lower()
+            if lower in seen:
+                continue
+            seen.add(lower)
+            deduped.append(token)
+        return deduped or [text]
+
+    def _build_composite_genre_hints(self, genres: List[str], refs: List[str]) -> List[str]:
+        if len(genres) <= 1:
+            return []
+
+        primary = genres[0]
+        secondaries = genres[1:]
+        hints: List[str] = []
+        hints.append(
+            f"以“{primary}”作为主引擎推进主线，每章至少保留1处“{'/'.join(secondaries)}”特征表达。"
+        )
+        if refs:
+            hints.append(f"复合题材执行参考：{refs[0]}")
+        hints.append("主辅题材冲突时，优先保证主题材读者承诺，辅题材用于制造新鲜感。")
+        return hints
+
     def _build_writing_checklist(
         self,
         chapter: int,

.claude/scripts/data_modules/index_manager.py

@@ -179,6 +179,26 @@ class ReviewMetrics:
     notes: str = ""
 
 
+@dataclass
+class WritingChecklistScoreMeta:
+    """写作清单评分记录（Context Contract v2 Phase F）"""
+
+    chapter: int
+    template: str = "plot"
+    total_items: int = 0
+    required_items: int = 0
+    completed_items: int = 0
+    completed_required: int = 0
+    total_weight: float = 0.0
+    completed_weight: float = 0.0
+    completion_rate: float = 0.0
+    score: float = 0.0
+    score_breakdown: Dict[str, Any] = field(default_factory=dict)
+    pending_items: List[str] = field(default_factory=list)
+    source: str = "context_manager"
+    notes: str = ""
+
+
 class IndexManager:
     """索引管理器"""
 
@@ -515,6 +535,31 @@ class IndexManager:
                 "CREATE INDEX IF NOT EXISTS idx_tool_stats_chapter ON tool_call_stats(chapter)"
             )
 
+            # 写作清单评分记录（Phase F）
+            cursor.execute("""
+                CREATE TABLE IF NOT EXISTS writing_checklist_scores (
+                    chapter INTEGER PRIMARY KEY,
+                    template TEXT DEFAULT 'plot',
+                    total_items INTEGER DEFAULT 0,
+                    required_items INTEGER DEFAULT 0,
+                    completed_items INTEGER DEFAULT 0,
+                    completed_required INTEGER DEFAULT 0,
+                    total_weight REAL DEFAULT 0,
+                    completed_weight REAL DEFAULT 0,
+                    completion_rate REAL DEFAULT 0,
+                    score REAL DEFAULT 0,
+                    score_breakdown TEXT,
+                    pending_items TEXT,
+                    source TEXT,
+                    notes TEXT,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                )
+            """)
+            cursor.execute(
+                "CREATE INDEX IF NOT EXISTS idx_checklist_score_value ON writing_checklist_scores(score)"
+            )
+
             conn.commit()
 
     @contextmanager
@@ -1928,6 +1973,132 @@ class IndexManager:
             "recent_ranges": recent_ranges,
         }
 
+    # ==================== 写作清单评分（Phase F） ====================
+
+    def save_writing_checklist_score(self, meta: WritingChecklistScoreMeta) -> None:
+        """保存章节写作清单评分。"""
+        with self._get_conn() as conn:
+            cursor = conn.cursor()
+            cursor.execute(
+                """
+                INSERT INTO writing_checklist_scores (
+                    chapter, template, total_items, required_items,
+                    completed_items, completed_required,
+                    total_weight, completed_weight, completion_rate, score,
+                    score_breakdown, pending_items, source, notes
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                ON CONFLICT(chapter) DO UPDATE SET
+                    template=excluded.template,
+                    total_items=excluded.total_items,
+                    required_items=excluded.required_items,
+                    completed_items=excluded.completed_items,
+                    completed_required=excluded.completed_required,
+                    total_weight=excluded.total_weight,
+                    completed_weight=excluded.completed_weight,
+                    completion_rate=excluded.completion_rate,
+                    score=excluded.score,
+                    score_breakdown=excluded.score_breakdown,
+                    pending_items=excluded.pending_items,
+                    source=excluded.source,
+                    notes=excluded.notes,
+                    updated_at=CURRENT_TIMESTAMP
+            """,
+                (
+                    meta.chapter,
+                    meta.template,
+                    meta.total_items,
+                    meta.required_items,
+                    meta.completed_items,
+                    meta.completed_required,
+                    meta.total_weight,
+                    meta.completed_weight,
+                    meta.completion_rate,
+                    meta.score,
+                    json.dumps(meta.score_breakdown, ensure_ascii=False),
+                    json.dumps(meta.pending_items, ensure_ascii=False),
+                    meta.source,
+                    meta.notes,
+                ),
+            )
+            conn.commit()
+
+    def get_writing_checklist_score(self, chapter: int) -> Optional[Dict[str, Any]]:
+        """获取指定章节的写作清单评分。"""
+        with self._get_conn() as conn:
+            cursor = conn.cursor()
+            cursor.execute(
+                "SELECT * FROM writing_checklist_scores WHERE chapter = ?",
+                (chapter,),
+            )
+            row = cursor.fetchone()
+            if not row:
+                return None
+            return self._row_to_dict(row, parse_json=["score_breakdown", "pending_items"])
+
+    def get_recent_writing_checklist_scores(self, limit: int = 10) -> List[Dict[str, Any]]:
+        """获取最近章节写作清单评分。"""
+        with self._get_conn() as conn:
+            cursor = conn.cursor()
+            cursor.execute(
+                """
+                SELECT * FROM writing_checklist_scores
+                ORDER BY chapter DESC
+                LIMIT ?
+            """,
+                (limit,),
+            )
+            return [
+                self._row_to_dict(row, parse_json=["score_breakdown", "pending_items"])
+                for row in cursor.fetchall()
+            ]
+
+    def get_writing_checklist_score_trend(self, last_n: int = 10) -> Dict[str, Any]:
+        """获取写作清单评分趋势统计。"""
+        records = self.get_recent_writing_checklist_scores(limit=max(1, int(last_n)))
+        if not records:
+            return {
+                "count": 0,
+                "score_avg": 0.0,
+                "completion_avg": 0.0,
+                "required_completion_avg": 0.0,
+                "recent": [],
+            }
+
+        scores: List[float] = []
+        completion_rates: List[float] = []
+        required_rates: List[float] = []
+        for row in records:
+            try:
+                scores.append(float(row.get("score", 0.0)))
+            except (TypeError, ValueError):
+                pass
+            try:
+                completion_rates.append(float(row.get("completion_rate", 0.0)))
+            except (TypeError, ValueError):
+                pass
+
+            required_items = int(row.get("required_items") or 0)
+            completed_required = int(row.get("completed_required") or 0)
+            if required_items > 0:
+                required_rates.append(completed_required / required_items)
+            else:
+                required_rates.append(1.0)
+
+        return {
+            "count": len(records),
+            "score_avg": round(sum(scores) / len(scores), 2) if scores else 0.0,
+            "completion_avg": round(sum(completion_rates) / len(completion_rates), 4) if completion_rates else 0.0,
+            "required_completion_avg": round(sum(required_rates) / len(required_rates), 4) if required_rates else 0.0,
+            "recent": [
+                {
+                    "chapter": row.get("chapter"),
+                    "score": row.get("score"),
+                    "completion_rate": row.get("completion_rate"),
+                }
+                for row in records
+            ],
+        }
+
     def get_debt_summary(self) -> Dict[str, Any]:
         """获取债务汇总信息"""
         with self._get_conn() as conn:
@@ -2357,6 +2528,18 @@ def main():
     review_trend_parser = subparsers.add_parser("get-review-trend-stats")
     review_trend_parser.add_argument("--last-n", type=int, default=5)
 
+    checklist_score_save_parser = subparsers.add_parser("save-writing-checklist-score")
+    checklist_score_save_parser.add_argument("--data", required=True, help="JSON 格式的写作清单评分数据")
+
+    checklist_score_get_parser = subparsers.add_parser("get-writing-checklist-score")
+    checklist_score_get_parser.add_argument("--chapter", type=int, required=True)
+
+    checklist_score_recent_parser = subparsers.add_parser("get-recent-writing-checklist-scores")
+    checklist_score_recent_parser.add_argument("--limit", type=int, default=10)
+
+    checklist_score_trend_parser = subparsers.add_parser("get-writing-checklist-score-trend")
+    checklist_score_trend_parser.add_argument("--last-n", type=int, default=10)
+
     # ==================== v5.3 引入命令 ====================
 
     # 获取债务汇总
@@ -2634,6 +2817,42 @@ def main():
         stats = manager.get_review_trend_stats(args.last_n)
         emit_success(stats, message="review_trend_stats")
 
+    elif args.command == "save-writing-checklist-score":
+        data = json.loads(args.data)
+        metrics = WritingChecklistScoreMeta(
+            chapter=data["chapter"],
+            template=data.get("template", "plot"),
+            total_items=data.get("total_items", 0),
+            required_items=data.get("required_items", 0),
+            completed_items=data.get("completed_items", 0),
+            completed_required=data.get("completed_required", 0),
+            total_weight=data.get("total_weight", 0.0),
+            completed_weight=data.get("completed_weight", 0.0),
+            completion_rate=data.get("completion_rate", 0.0),
+            score=data.get("score", 0.0),
+            score_breakdown=data.get("score_breakdown", {}),
+            pending_items=data.get("pending_items", []),
+            source=data.get("source", "context_manager"),
+            notes=data.get("notes", ""),
+        )
+        manager.save_writing_checklist_score(metrics)
+        emit_success({"chapter": metrics.chapter, "score": metrics.score}, message="writing_checklist_score_saved")
+
+    elif args.command == "get-writing-checklist-score":
+        score = manager.get_writing_checklist_score(args.chapter)
+        if score:
+            emit_success(score, message="writing_checklist_score")
+        else:
+            emit_error("NOT_FOUND", f"未找到第 {args.chapter} 章的写作清单评分")
+
+    elif args.command == "get-recent-writing-checklist-scores":
+        scores = manager.get_recent_writing_checklist_scores(args.limit)
+        emit_success(scores, message="recent_writing_checklist_scores")
+
+    elif args.command == "get-writing-checklist-score-trend":
+        trend = manager.get_writing_checklist_score_trend(args.last_n)
+        emit_success(trend, message="writing_checklist_score_trend")
+
     # ==================== v5.3 引入命令处理 ====================
 
     elif args.command == "get-debt-summary":

.claude/scripts/data_modules/tests/test_context_manager.py

@@ -281,10 +281,70 @@ def test_context_manager_includes_writing_guidance(temp_project):
     checklist = guidance.get("checklist") or []
     assert isinstance(checklist, list)
     assert checklist
+    checklist_score = guidance.get("checklist_score") or {}
+    assert isinstance(checklist_score, dict)
+    assert "score" in checklist_score
+    assert "completion_rate" in checklist_score
     first_item = checklist[0]
     assert isinstance(first_item, dict)
     assert {"id", "label", "weight", "required", "source", "verify_hint"}.issubset(first_item.keys())
 
+    persisted = idx.get_writing_checklist_score(4)
+    assert isinstance(persisted, dict)
+    assert persisted.get("chapter") == 4
+    assert persisted.get("score") is not None
+
+
+def test_context_manager_dynamic_weights_and_composite_genre(temp_project):
+    refs_dir = temp_project.project_root / ".claude" / "references"
+    refs_dir.mkdir(parents=True, exist_ok=True)
+    (refs_dir / "genre-profiles.md").write_text(
+        """
+## xuanhuan
+- 升级线清晰
+
+## realistic
+- 社会议题映射
+""".strip(),
+        encoding="utf-8",
+    )
+    (refs_dir / "reading-power-taxonomy.md").write_text(
+        """
+## xuanhuan
+- 钩子强度优先
+
+## realistic
+- 人物动机一致
+""".strip(),
+        encoding="utf-8",
+    )
+
+    state = {
+        "project": {"genre": "xuanhuan+realistic"},
+        "protagonist_state": {"name": "萧炎"},
+        "chapter_meta": {},
+        "disambiguation_warnings": [],
+        "disambiguation_pending": [],
+    }
+    temp_project.state_file.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
+
+    manager = ContextManager(temp_project)
+    payload_early = manager.build_context(10, template="plot", use_snapshot=False, save_snapshot=False)
+    payload_late = manager.build_context(150, template="plot", use_snapshot=False, save_snapshot=False)
+
+    assert payload_early.get("weights", {}).get("core") >= payload_late.get("weights", {}).get("core")
+    assert payload_late.get("weights", {}).get("global") >= payload_early.get("weights", {}).get("global")
+    assert payload_early.get("meta", {}).get("context_weight_stage") == "early"
+    assert payload_late.get("meta", {}).get("context_weight_stage") == "late"
+
+    profile = payload_early["sections"]["genre_profile"]["content"]
+    assert profile.get("composite") is True
+    assert profile.get("genre") == "xuanhuan"
+    assert isinstance(profile.get("genres"), list)
+    assert "realistic" in (profile.get("genres") or [])
+    assert isinstance(profile.get("composite_hints"), list)
+    assert profile.get("composite_hints")
+
 
 def test_context_manager_compact_text_truncation(temp_project):
     manager = ContextManager(temp_project)

.claude/scripts/data_modules/tests/test_data_modules.py

@@ -32,6 +32,7 @@ from data_modules.index_manager import (
     ChaseDebtMeta,
     ChapterReadingPowerMeta,
     ReviewMetrics,
+    WritingChecklistScoreMeta,
 )
 
 
@@ -896,6 +897,56 @@ class TestIndexManager:
         assert trends["overall_avg"] > 0
         assert "爽点密度" in trends["dimension_avg"]
 
+    def test_writing_checklist_score_persistence_and_trend(self, temp_project):
+        manager = IndexManager(temp_project)
+
+        manager.save_writing_checklist_score(
+            WritingChecklistScoreMeta(
+                chapter=10,
+                template="plot",
+                total_items=6,
+                required_items=4,
+                completed_items=4,
+                completed_required=3,
+                total_weight=6.2,
+                completed_weight=4.1,
+                completion_rate=0.6667,
+                score=78.5,
+                score_breakdown={"weighted_completion_rate": 0.66},
+                pending_items=["段末留钩"],
+            )
+        )
+        manager.save_writing_checklist_score(
+            WritingChecklistScoreMeta(
+                chapter=11,
+                template="plot",
+                total_items=6,
+                required_items=4,
+                completed_items=5,
+                completed_required=4,
+                total_weight=6.2,
+                completed_weight=5.4,
+                completion_rate=0.8333,
+                score=86.0,
+                score_breakdown={"weighted_completion_rate": 0.87},
+                pending_items=[],
+            )
+        )
+
+        one = manager.get_writing_checklist_score(10)
+        assert one is not None
+        assert one["chapter"] == 10
+        assert one["score"] == 78.5
+
+        recent = manager.get_recent_writing_checklist_scores(limit=2)
+        assert len(recent) == 2
+        assert recent[0]["chapter"] == 11
+
+        trend = manager.get_writing_checklist_score_trend(last_n=5)
+        assert trend["count"] == 2
+        assert trend["score_avg"] > 0
+        assert trend["completion_avg"] > 0
+
     def test_index_manager_cli(self, temp_project, monkeypatch, capsys):
         root = str(temp_project.project_root)
         manager = IndexManager(temp_project)
@@ -1229,6 +1280,35 @@ class TestIndexManager:
         run_cli(["--project-root", root, "get-recent-review-metrics", "--limit", "5"])
         run_cli(["--project-root", root, "get-review-trend-stats", "--last-n", "5"])
 
+        checklist_payload = {
+            "chapter": 5,
+            "template": "plot",
+            "total_items": 6,
+            "required_items": 4,
+            "completed_items": 4,
+            "completed_required": 3,
+            "total_weight": 6.5,
+            "completed_weight": 4.8,
+            "completion_rate": 0.6667,
+            "score": 79.2,
+            "score_breakdown": {"weighted_completion_rate": 0.73},
+            "pending_items": ["钩子差异化"],
+            "source": "context_manager",
+        }
+        run_cli(
+            [
+                "--project-root",
+                root,
+                "save-writing-checklist-score",
+                "--data",
+                json.dumps(checklist_payload, ensure_ascii=False),
+            ]
+        )
+        run_cli(["--project-root", root, "get-writing-checklist-score", "--chapter", "5"])
+        run_cli(["--project-root", root, "get-writing-checklist-score", "--chapter", "99"])
+        run_cli(["--project-root", root, "get-recent-writing-checklist-scores", "--limit", "5"])
+        run_cli(["--project-root", root, "get-writing-checklist-score-trend", "--last-n", "5"])
+
         capsys.readouterr()
 
 

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

@@ -87,9 +87,11 @@ def test_build_chapter_context_payload_includes_contract_sections(tmp_path):
 
     payload = build_chapter_context_payload(tmp_path, 3)
     assert payload["context_contract_version"] == "v2"
+    assert payload.get("context_weight_stage") in {"early", "mid", "late"}
     assert "writing_guidance" in payload
     assert isinstance(payload["writing_guidance"].get("guidance_items"), list)
     assert isinstance(payload["writing_guidance"].get("checklist"), list)
+    assert isinstance(payload["writing_guidance"].get("checklist_score"), dict)
     assert payload["genre_profile"].get("genre") == "xuanhuan"
 
 
@@ -106,8 +108,14 @@ def test_render_text_contains_writing_guidance_section(tmp_path):
         "previous_summaries": ["### 第9章摘要\n上一章"],
         "state_summary": "状态",
         "context_contract_version": "v2",
+        "context_weight_stage": "early",
         "reader_signal": {"review_trend": {"overall_avg": 72}, "low_score_ranges": [{"start_chapter": 8, "end_chapter": 9}]},
-        "genre_profile": {"genre": "xuanhuan", "reference_hints": ["升级线清晰"]},
+        "genre_profile": {
+            "genre": "xuanhuan",
+            "genres": ["xuanhuan", "realistic"],
+            "composite_hints": ["以玄幻主线推进，同时保留现实议题表达"],
+            "reference_hints": ["升级线清晰"],
+        },
         "writing_guidance": {
             "guidance_items": ["先修低分", "钩子差异化"],
             "checklist": [
@@ -120,6 +128,11 @@ def test_render_text_contains_writing_guidance_section(tmp_path):
                     "verify_hint": "至少完成1处冲突升级",
                 }
             ],
+            "checklist_score": {
+                "score": 81.5,
+                "completion_rate": 0.66,
+                "required_completion_rate": 0.75,
+            },
         },
     }
 
@@ -127,6 +140,10 @@ def test_render_text_contains_writing_guidance_section(tmp_path):
     assert "## 写作执行建议" in text
     assert "先修低分" in text
     assert "## Contract (v2)" in text
+    assert "- 上下文阶段权重: early" in text
     assert "### 执行检查清单（可评分）" in text
     assert "- 总权重: 1.40" in text
     assert "[必做][w=1.4] 修复低分区间问题" in text
+    assert "### 执行评分" in text
+    assert "- 评分: 81.5" in text
+    assert "- 复合题材: xuanhuan + realistic" in text

.claude/scripts/data_modules/tests/test_workflow_manager.py

@@ -76,3 +76,25 @@ def test_complete_step_rejects_mismatch_step_id(tmp_path, monkeypatch):
     assert current_step["id"] == "Step 2A"
     assert current_step["status"] == module.STEP_STATUS_RUNNING
 
+
+def test_workflow_step_owner_and_order_violation_trace(tmp_path, monkeypatch):
+    module = _load_module()
+    monkeypatch.setattr(module, "find_project_root", lambda: tmp_path)
+
+    webnovel_dir = tmp_path / ".webnovel"
+    webnovel_dir.mkdir(parents=True, exist_ok=True)
+
+    assert module.expected_step_owner("webnovel-write", "Step 1") == "context-agent"
+    assert module.expected_step_owner("webnovel-write", "Step 5") == "data-agent"
+
+    module.start_task("webnovel-write", {"chapter_num": 12})
+    module.start_step("Step 3", "Review")
+
+    trace_path = module.get_call_trace_path()
+    lines = [json.loads(line) for line in trace_path.read_text(encoding="utf-8").splitlines() if line.strip()]
+    events = [row.get("event") for row in lines]
+    assert "step_order_violation" in events
+
+    step_started = [row for row in lines if row.get("event") == "step_started"]
+    assert step_started
+    assert step_started[-1].get("payload", {}).get("expected_owner") == "review-agents"

.claude/scripts/extract_chapter_context.py

@@ -186,6 +186,7 @@ def _load_contract_context(project_root: Path, chapter_num: int) -> Dict[str, An
     sections = payload.get("sections", {})
     return {
         "context_contract_version": (payload.get("meta") or {}).get("context_contract_version"),
+        "context_weight_stage": (payload.get("meta") or {}).get("context_weight_stage"),
         "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", {}),
@@ -210,6 +211,7 @@ def build_chapter_context_payload(project_root: Path, chapter_num: int) -> Dict[
         "previous_summaries": prev_summaries,
         "state_summary": state_summary,
         "context_contract_version": contract_context.get("context_contract_version"),
+        "context_weight_stage": contract_context.get("context_weight_stage"),
         "reader_signal": contract_context.get("reader_signal", {}),
         "genre_profile": contract_context.get("genre_profile", {}),
         "writing_guidance": contract_context.get("writing_guidance", {}),
@@ -247,10 +249,15 @@ def _render_text(payload: Dict[str, Any]) -> str:
     if contract_version:
         lines.append(f"## Contract ({contract_version})")
         lines.append("")
+        stage = payload.get("context_weight_stage")
+        if stage:
+            lines.append(f"- 上下文阶段权重: {stage}")
+            lines.append("")
 
     writing_guidance = payload.get("writing_guidance") or {}
     guidance_items = writing_guidance.get("guidance_items") or []
     checklist = writing_guidance.get("checklist") or []
+    checklist_score = writing_guidance.get("checklist_score") or {}
     if guidance_items or checklist:
         lines.append("## 写作执行建议")
         lines.append("")
@@ -289,6 +296,14 @@ def _render_text(payload: Dict[str, Any]) -> str:
                 if verify_hint:
                     lines.append(f"   - 验收: {verify_hint}")
 
+        if checklist_score:
+            lines.append("")
+            lines.append("### 执行评分")
+            lines.append("")
+            lines.append(f"- 评分: {checklist_score.get('score')}")
+            lines.append(f"- 完成率: {checklist_score.get('completion_rate')}")
+            lines.append(f"- 必做完成率: {checklist_score.get('required_completion_rate')}")
+
         lines.append("")
 
     reader_signal = payload.get("reader_signal") or {}
@@ -308,6 +323,12 @@ def _render_text(payload: Dict[str, Any]) -> str:
         lines.append("## 题材锚定")
         lines.append("")
         lines.append(f"- 题材: {genre_profile.get('genre')}")
+        genres = genre_profile.get("genres") or []
+        if len(genres) > 1:
+            lines.append(f"- 复合题材: {' + '.join(str(token) for token in genres)}")
+            composite_hints = genre_profile.get("composite_hints") or []
+            for row in composite_hints[:2]:
+                lines.append(f"- {row}")
         refs = genre_profile.get("reference_hints") or []
         for row in refs[:3]:
             lines.append(f"- {row}")

.claude/scripts/workflow_manager.py

@@ -81,6 +81,43 @@ def safe_append_call_trace(event: str, payload: Optional[Dict[str, Any]] = None)
         pass
 
 
+def expected_step_owner(command: str, step_id: str) -> str:
+    """Resolve expected caller owner by command + step id.
+
+    Returns concise owner tags to align with
+    `.claude/references/claude-code-call-matrix.md`.
+    """
+    if command == "webnovel-write":
+        mapping = {
+            "Step 1": "context-agent",
+            "Step 1.5": "webnovel-write-skill",
+            "Step 2A": "writer-draft",
+            "Step 2B": "style-adapter",
+            "Step 3": "review-agents",
+            "Step 4": "polish-agent",
+            "Step 5": "data-agent",
+            "Step 6": "backup-agent",
+        }
+        return mapping.get(step_id, "webnovel-write-skill")
+
+    if command == "webnovel-review":
+        return "webnovel-review-skill"
+
+    return "unknown"
+
+
+def step_allowed_before(command: str, step_id: str, completed_steps: list[Dict[str, Any]]) -> bool:
+    """Check simple ordering constraints by pending sequence."""
+    sequence = get_pending_steps(command)
+    if step_id not in sequence:
+        return True
+
+    expected_index = sequence.index(step_id)
+    completed_ids = [str(item.get("id")) for item in completed_steps]
+    required_before = sequence[:expected_index]
+    return all(prev in completed_ids for prev in required_before)
+
+
 def _new_task(command: str, args: Dict[str, Any]) -> Dict[str, Any]:
     started_at = now_iso()
     return {
@@ -165,6 +202,19 @@ def start_step(step_id, step_name, progress_note=None):
         print("⚠️ 无活动任务，请先使用 start-task")
         return
 
+    command = str(task.get("command") or "")
+    if not step_allowed_before(command, step_id, task.get("completed_steps", [])):
+        safe_append_call_trace(
+            "step_order_violation",
+            {
+                "step_id": step_id,
+                "command": command,
+                "completed_steps": [row.get("id") for row in task.get("completed_steps", [])],
+            },
+        )
+
+    owner = expected_step_owner(command, step_id)
+
     _finalize_current_step_as_failed(task, reason="step_replaced_before_completion")
 
     started_at = now_iso()
@@ -190,6 +240,7 @@ def start_step(step_id, step_name, progress_note=None):
             "command": task.get("command"),
             "chapter": task.get("args", {}).get("chapter_num"),
             "progress_note": progress_note,
+            "expected_owner": owner,
         },
     )
     print(f"▶️ {step_id} 开始: {step_name}")

README.md

@@ -793,6 +793,36 @@ git checkout ch0045
 - **Pydantic Schema**：DataAgentOutput 等结构化验证
 - **不向前兼容**：vectors.db 表结构变更时自动 DROP+CREATE
 
+### Context Contract v2（阶段 E）
+
+- `writing_guidance` 新增 `checklist`：可执行、可验收、可加权
+- checklist 项包含：`id/label/weight/required/source/verify_hint`
+- `extract_chapter_context.py` 文本输出新增“执行检查清单（可评分）”
+
+### Context Contract v2（阶段 F）
+
+- 新增 `writing_guidance.checklist_score`：章节执行评分与完成率
+- `index.db` 新增 `writing_checklist_scores` 持久化评分记录
+- 支持趋势查询：最近评分、评分均值、完成率均值
+
+### Context Contract v2（阶段 G）
+
+- `workflow_manager.py` 新增 Step 调用方标注（expected owner）
+- 新增步骤顺序违规追踪事件 `step_order_violation`
+- `call_trace.jsonl` 可用于定位“谁在何时调用了哪一步”
+
+### Context Contract v2（阶段 H）
+
+- 新增动态上下文预算：按章节阶段 early/mid/late 调整权重
+- `meta.context_weight_stage` 公开当前权重阶段
+- 目标：开篇重冲突与角色、中后期重世界与线索收束
+
+### Context Contract v2（阶段 I）
+
+- `genre_profile` 支持复合题材（如 `xuanhuan+realistic`）
+- 输出新增：`genres/composite/secondary_genres/composite_hints`
+- 写作建议自动加入“复合题材协同”提示
+
 ### v5.3
 - **追读力分类标准**：钩子5类型、爽点8模式、微兑现7类型
 - **约束分层机制**：Hard Invariants (4条) + Soft Guidance (可Override)