refactor: slim context_manager to pure JSON assembler, remove snapshot

- Delete SnapshotManager and snapshot_manager.py entirely
- Replace assemble_context with _assemble_json_payload returning flat dict
- Remove _is_snapshot_compatible, _story_contract_signature, _payload_signature, _compact_json_text
- Remove snapshot args from build_context and CLI main()
- Update _load_contract_context to read flat payload instead of sections nesting
- Replace _render_text with pure JSON serialization
- Update all tests to match new flat payload structure (v3 contract)
- Remove snapshot-specific tests

webnovel-writer/scripts/data_modules/__init__.py

@@ -45,7 +45,6 @@ __all__ = [
     "SearchResult",
     "ContextManager",
     "ContextRanker",
-    "SnapshotManager",
     "QueryRouter",
     # Style Sampler
     "StyleSampler",
@@ -88,7 +87,6 @@ _LAZY_EXPORTS: dict[str, tuple[str, str]] = {
     "SearchResult": (".rag_adapter", "SearchResult"),
     "ContextManager": (".context_manager", "ContextManager"),
     "ContextRanker": (".context_ranker", "ContextRanker"),
-    "SnapshotManager": (".snapshot_manager", "SnapshotManager"),
     "QueryRouter": (".query_router", "QueryRouter"),
     # Style Sampler
     "StyleSampler": (".style_sampler", "StyleSampler"),

webnovel-writer/scripts/data_modules/context_manager.py

@@ -9,7 +9,6 @@ import json
 import re
 import sys
 import logging
-import hashlib
 from pathlib import Path
 
 from runtime_compat import enable_windows_utf8_stdio
@@ -30,7 +29,6 @@ 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 .story_runtime_sources import RuntimeSourceSnapshot, load_runtime_sources
 from .context_weights import (
@@ -96,61 +94,15 @@ class ContextManager:
     ]
     SUMMARY_SECTION_RE = re.compile(r"##\s*剧情摘要\s*\r?\n(.*?)(?=\r?\n##|\Z)", re.DOTALL)
 
-    def __init__(self, config=None, snapshot_manager: Optional[SnapshotManager] = None):
+    def __init__(self, config=None):
         self.config = config or get_config()
-        self.snapshot_manager = snapshot_manager or SnapshotManager(self.config)
         self.index_manager = IndexManager(self.config)
         self.context_ranker = ContextRanker(self.config)
 
-    def _is_snapshot_compatible(self, cached: Dict[str, Any], template: str) -> bool:
-        """判断快照是否可用于当前模板。"""
-        if not isinstance(cached, dict):
-            return False
-
-        meta = cached.get("meta")
-        if not isinstance(meta, dict):
-            # 兼容旧快照：未记录 template 时仅允许默认模板复用
-            return template == self.DEFAULT_TEMPLATE
-
-        cached_template = meta.get("template")
-        if not isinstance(cached_template, str):
-            return template == self.DEFAULT_TEMPLATE
-
-        if cached_template != template:
-            return False
-
-        payload = cached.get("payload", cached)
-        if not isinstance(payload, dict):
-            return False
-        sections = payload.get("sections")
-        if not isinstance(sections, dict):
-            return False
-
-        required_sections = {
-            "plot_structure",
-            "long_term_memory",
-            "story_contract",
-            "runtime_status",
-            "latest_commit",
-            "prewrite_validation",
-        }
-        if not required_sections.issubset(set(sections.keys())):
-            return False
-
-        chapter = int(cached.get("chapter") or (payload.get("meta") or {}).get("chapter") or 0)
-        if chapter <= 0:
-            return False
-
-        snapshot_signature = meta.get("story_contract_signature")
-        current_signature = self._story_contract_signature(chapter)
-        return snapshot_signature == current_signature
-
     def build_context(
         self,
         chapter: int,
         template: str | None = None,
-        use_snapshot: bool = True,
-        save_snapshot: bool = True,
         max_chars: Optional[int] = None,
     ) -> Dict[str, Any]:
         template = template or self.DEFAULT_TEMPLATE
@@ -159,62 +111,34 @@ class ContextManager:
             template = self.DEFAULT_TEMPLATE
             self._active_template = template
 
-        if use_snapshot:
-            try:
-                cached = self.snapshot_manager.load_snapshot(chapter)
-                if cached and self._is_snapshot_compatible(cached, template):
-                    return cached.get("payload", cached)
-            except SnapshotVersionMismatch:
-                # Snapshot incompatible; rebuild below.
-                pass
-
         pack = self._build_pack(chapter)
         if getattr(self.config, "context_ranker_enabled", True):
             pack = self.context_ranker.rank_pack(pack, chapter)
-        assembled = self.assemble_context(pack, template=template, max_chars=max_chars)
 
-        if save_snapshot:
-            meta = {
-                "template": template,
-                "story_contract_signature": self._story_contract_signature(chapter),
-            }
-            self.snapshot_manager.save_snapshot(chapter, assembled, meta=meta)
+        return self._assemble_json_payload(pack, template=template)
 
-        return assembled
-
-    def assemble_context(
-        self,
-        pack: Dict[str, Any],
-        template: str = DEFAULT_TEMPLATE,
-        max_chars: Optional[int] = None,
-    ) -> Dict[str, Any]:
+    def _assemble_json_payload(self, pack: Dict[str, Any], template: str = DEFAULT_TEMPLATE) -> Dict[str, Any]:
         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)
 
-        sections = {}
+        payload: Dict[str, Any] = {
+            "meta": {
+                **(pack.get("meta") or {}),
+                "context_contract_version": "v3",
+            },
+        }
+
         for section_name in self.SECTION_ORDER:
-            if section_name in pack:
-                sections[section_name] = pack[section_name]
-
-        assembled: Dict[str, Any] = {"meta": pack.get("meta", {}), "sections": {}}
-        for name, content in sections.items():
-            weight = weights.get(name, 0.0)
-            if weight > 0:
-                budget = int(max_chars * weight)
-            elif name in self.EXTRA_SECTIONS and extra_budget > 0:
-                budget = extra_budget
-            else:
-                budget = None
-            text = self._compact_json_text(content, budget)
-            assembled["sections"][name] = {"content": content, "text": text, "budget": budget}
+            if section_name in pack and section_name != "global":
+                content = pack[section_name]
+                weight = weights.get(section_name, 0.0)
+                if weight > 0 or section_name in self.EXTRA_SECTIONS:
+                    payload[section_name] = content
 
-        assembled["template"] = template
-        assembled["weights"] = weights
         if chapter > 0:
-            assembled.setdefault("meta", {})["context_weight_stage"] = self._resolve_context_stage(chapter)
-        return assembled
+            payload["meta"]["context_weight_stage"] = self._resolve_context_stage(chapter)
+
+        return payload
 
     def filter_invalid_items(self, items: List[Dict[str, Any]], source_type: str, id_key: str) -> List[Dict[str, Any]]:
         confirmed = self.index_manager.get_invalid_ids(source_type, status="confirmed")
@@ -746,23 +670,6 @@ class ContextManager:
         profile_key = to_profile_key(genre)
         return profile_key in whitelist
 
-    def _compact_json_text(self, content: Any, budget: Optional[int]) -> str:
-        raw = json.dumps(content, ensure_ascii=False)
-        if budget is None or len(raw) <= budget:
-            return raw
-        if not getattr(self.config, "context_compact_text_enabled", True):
-            return raw[:budget]
-
-        min_budget = max(1, int(getattr(self.config, "context_compact_min_budget", 120)))
-        if budget <= min_budget:
-            return raw[:budget]
-
-        head_ratio = float(getattr(self.config, "context_compact_head_ratio", 0.65))
-        head_budget = int(budget * max(0.2, min(0.9, head_ratio)))
-        tail_budget = max(0, budget - head_budget - 10)
-        compact = f"{raw[:head_budget]}…[TRUNCATED]{raw[-tail_budget:] if tail_budget else ''}"
-        return compact[:budget]
-
     def _extract_genre_section(self, text: str, genre: str) -> str:
         return extract_genre_section(text, genre)
 
@@ -791,37 +698,6 @@ class ContextManager:
             "anti_patterns": read_json_if_exists(story_root / "anti_patterns.json") or [],
         }
 
-    def _story_contract_signature(self, chapter: int) -> Dict[str, str]:
-        story_root = self.config.story_system_dir
-        runtime_sources = load_runtime_sources(self.config.project_root, chapter)
-        volume_path = story_root / "volumes"
-        volume_ref = runtime_sources.contracts.get("volume") or {}
-        volume_num = int((volume_ref.get("meta") or {}).get("volume") or 0)
-        volume_path = volume_path / f"volume_{max(volume_num, 1):03d}.json"
-        paths = {
-            "master_setting": story_root / "MASTER_SETTING.json",
-            "chapter_brief": story_root / "chapters" / f"chapter_{chapter:03d}.json",
-            "volume_brief": volume_path,
-            "review_contract": story_root / "reviews" / f"chapter_{chapter:03d}.review.json",
-            "anti_patterns": story_root / "anti_patterns.json",
-        }
-        signature: Dict[str, str] = {}
-        for name, path in paths.items():
-            if not path.is_file():
-                signature[name] = "missing"
-                continue
-            digest = hashlib.sha1(path.read_bytes()).hexdigest()
-            signature[name] = digest
-        signature["latest_commit"] = self._payload_signature(runtime_sources.latest_commit)
-        signature["latest_accepted_commit"] = self._payload_signature(runtime_sources.latest_accepted_commit)
-        return signature
-
-    def _payload_signature(self, payload: Any) -> str:
-        if not payload:
-            return "missing"
-        encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True).encode("utf-8")
-        return hashlib.sha1(encoded).hexdigest()
-
     def _load_recent_summaries(self, chapter: int, window: int = 3) -> List[Dict[str, Any]]:
         summaries = []
         for ch in range(max(1, chapter - window), chapter):
@@ -914,14 +790,12 @@ def main():
     parser.add_argument("--project-root", type=str, help="项目根目录")
     parser.add_argument("--chapter", type=int, required=True)
     parser.add_argument("--template", type=str, default=ContextManager.DEFAULT_TEMPLATE)
-    parser.add_argument("--no-snapshot", action="store_true")
-    parser.add_argument("--max-chars", type=int, default=8000)
 
     args = parser.parse_args()
 
     config = None
     if args.project_root:
-        # 允许传入“工作区根目录”，统一解析到真正的 book project_root（必须包含 .webnovel/state.json）
+        # 允许传入"工作区根目录"，统一解析到真正的 book project_root（必须包含 .webnovel/state.json）
         from project_locator import resolve_project_root
         from .config import DataModulesConfig
 
@@ -933,9 +807,6 @@ def main():
         payload = manager.build_context(
             chapter=args.chapter,
             template=args.template,
-            use_snapshot=not args.no_snapshot,
-            save_snapshot=True,
-            max_chars=args.max_chars,
         )
         print_success(payload, message="context_built")
         try:

webnovel-writer/scripts/data_modules/snapshot_manager.py

@@ -1,95 +0,0 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-"""
-Context snapshot manager.
-"""
-from __future__ import annotations
-
-import json
-from dataclasses import dataclass
-from datetime import datetime, timezone
-from filelock import FileLock
-from pathlib import Path
-from typing import Any, Dict, Optional
-
-from .config import get_config
-
-try:
-    # 当 scripts 目录在 sys.path 中
-    from security_utils import atomic_write_json
-except ImportError:  # pragma: no cover
-    # 当以 python -m scripts.data_modules... 形式运行
-    from scripts.security_utils import atomic_write_json
-
-SNAPSHOT_VERSION = "1.3"
-
-
-class SnapshotVersionMismatch(RuntimeError):
-    def __init__(self, expected: str, actual: str) -> None:
-        super().__init__(f"snapshot version mismatch: expected {expected}, got {actual}")
-        self.expected = expected
-        self.actual = actual
-
-
-@dataclass
-class SnapshotMeta:
-    chapter: int
-    version: str
-    saved_at: str
-
-
-class SnapshotManager:
-    def __init__(self, config=None, version: str = SNAPSHOT_VERSION):
-        self.config = config or get_config()
-        self.version = version
-        self.snapshot_dir = self.config.webnovel_dir / "context_snapshots"
-        self.snapshot_dir.mkdir(parents=True, exist_ok=True)
-
-    def _snapshot_path(self, chapter: int) -> Path:
-        return self.snapshot_dir / f"ch{chapter:04d}.json"
-
-    def _snapshot_lock_path(self, chapter: int) -> Path:
-        return self._snapshot_path(chapter).with_suffix(".json.lock")
-
-    def save_snapshot(self, chapter: int, payload: Dict[str, Any], meta: Optional[Dict[str, Any]] = None) -> Path:
-        data: Dict[str, Any] = {
-            "version": self.version,
-            "chapter": chapter,
-            "saved_at": datetime.now(timezone.utc).isoformat(),
-            "payload": payload,
-        }
-        if meta:
-            data["meta"] = meta
-
-        path = self._snapshot_path(chapter)
-        lock = FileLock(str(self._snapshot_lock_path(chapter)), timeout=10)
-        with lock:
-            atomic_write_json(path, data, use_lock=False, backup=False)
-        return path
-
-    def load_snapshot(self, chapter: int) -> Optional[Dict[str, Any]]:
-        path = self._snapshot_path(chapter)
-        lock = FileLock(str(self._snapshot_lock_path(chapter)), timeout=10)
-        with lock:
-            if not path.exists():
-                return None
-            try:
-                data = json.loads(path.read_text(encoding="utf-8"))
-            except (json.JSONDecodeError, OSError):
-                return None
-            version = str(data.get("version", ""))
-            if version != self.version:
-                raise SnapshotVersionMismatch(self.version, version)
-            return data
-
-    def delete_snapshot(self, chapter: int) -> bool:
-        path = self._snapshot_path(chapter)
-        lock = FileLock(str(self._snapshot_lock_path(chapter)), timeout=10)
-        with lock:
-            if path.exists():
-                path.unlink()
-                return True
-        return False
-
-    def list_snapshots(self) -> list[str]:
-        return sorted(p.name for p in self.snapshot_dir.glob("ch*.json"))

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

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 """
-ContextManager and SnapshotManager tests
+ContextManager tests
 """
 
 import json
@@ -17,7 +17,6 @@ from data_modules.index_manager import (
     ReviewMetrics,
 )
 from data_modules.context_manager import ContextManager
-from data_modules.snapshot_manager import SnapshotManager, SnapshotVersionMismatch
 from data_modules.query_router import QueryRouter
 
 
@@ -28,30 +27,6 @@ def temp_project(tmp_path):
     return cfg
 
 
-def test_snapshot_manager_roundtrip(temp_project):
-    manager = SnapshotManager(temp_project)
-    payload = {"hello": "world"}
-    manager.save_snapshot(1, payload)
-    loaded = manager.load_snapshot(1)
-    assert loaded["payload"] == payload
-
-
-def test_snapshot_version_mismatch(temp_project):
-    manager = SnapshotManager(temp_project, version="1.0")
-    manager.save_snapshot(1, {"a": 1})
-    other = SnapshotManager(temp_project, version="2.0")
-    with pytest.raises(SnapshotVersionMismatch):
-        other.load_snapshot(1)
-
-
-def test_snapshot_delete_roundtrip(temp_project):
-    manager = SnapshotManager(temp_project)
-    manager.save_snapshot(2, {"x": 1})
-
-    assert manager.delete_snapshot(2) is True
-    assert manager.load_snapshot(2) is None
-
-
 def test_context_manager_build_and_filter(temp_project):
     state = {
         "protagonist_state": {"name": "萧炎", "location": {"current": "天云宗"}},
@@ -90,12 +65,12 @@ def test_context_manager_build_and_filter(temp_project):
     idx.resolve_invalid_fact(invalid_id, "confirm")
 
     manager = ContextManager(temp_project)
-    payload = manager.build_context(1, use_snapshot=False, save_snapshot=False)
-    characters = payload["sections"]["scene"]["content"]["appearing_characters"]
+    payload = manager.build_context(1)
+    characters = payload["scene"]["appearing_characters"]
     assert any(c.get("entity_id") == "xiaoyan" for c in characters)
     assert not any(c.get("entity_id") == "bad" for c in characters)
-    assert payload["sections"]["preferences"]["content"].get("tone") == "热血"
-    assert "long_term_memory" in payload["sections"]
+    assert payload["preferences"].get("tone") == "热血"
+    assert "long_term_memory" in payload
 
 
 def test_context_manager_uses_memory_orchestrator_for_working_when_enabled(temp_project, monkeypatch):
@@ -131,9 +106,9 @@ def test_context_manager_uses_memory_orchestrator_for_working_when_enabled(temp_
 
     monkeypatch.setattr("data_modules.memory.orchestrator.MemoryOrchestrator.build_memory_pack", _fake_pack)
     manager = ContextManager(temp_project)
-    payload = manager.build_context(1, use_snapshot=False, save_snapshot=False)
-    core = payload["sections"]["core"]["content"]
-    long_term_memory = payload["sections"]["long_term_memory"]["content"]
+    payload = manager.build_context(1)
+    core = payload["core"]
+    long_term_memory = payload["long_term_memory"]
 
     assert "working_memory" in long_term_memory
     assert core["chapter_outline"] == "FAKE_OUTLINE"
@@ -156,9 +131,9 @@ def test_context_manager_skips_memory_orchestrator_when_disabled(temp_project, m
 
     monkeypatch.setattr("data_modules.memory.orchestrator.MemoryOrchestrator.build_memory_pack", _boom)
     manager = ContextManager(temp_project)
-    payload = manager.build_context(1, use_snapshot=False, save_snapshot=False)
+    payload = manager.build_context(1)
 
-    assert payload["sections"]["long_term_memory"]["content"] == {}
+    assert payload["long_term_memory"] == {}
 
 
 def test_context_manager_loads_volume_outline_file(temp_project):
@@ -181,9 +156,9 @@ def test_context_manager_loads_volume_outline_file(temp_project):
     )
 
     manager = ContextManager(temp_project)
-    payload = manager.build_context(2, use_snapshot=False, save_snapshot=False)
+    payload = manager.build_context(2)
 
-    outline = payload["sections"]["core"]["content"]["chapter_outline"]
+    outline = payload["core"]["chapter_outline"]
     assert "### 第2章：测试标题" in outline
     assert "测试大纲" in outline
 
@@ -268,14 +243,14 @@ def test_context_manager_includes_story_contract_and_prewrite_validation(temp_pr
     )
 
     manager = ContextManager(temp_project)
-    payload = manager.build_context(3, use_snapshot=False, save_snapshot=False)
+    payload = manager.build_context(3)
 
-    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")
+    assert "story_contract" in payload
+    assert "prewrite_validation" in payload
+    assert payload["story_contract"]["review_contract"]["meta"]["contract_type"] == "REVIEW_CONTRACT"
+    assert payload["prewrite_validation"]["fulfillment_seed"]["planned_nodes"] == ["发现陷阱"]
+    payload_keys = list(payload.keys())
+    assert payload_keys.index("story_contract") < payload_keys.index("scene")
 
 
 def test_context_manager_prefers_contract_route_over_legacy_genre_profile(temp_project):
@@ -308,12 +283,12 @@ def test_context_manager_prefers_contract_route_over_legacy_genre_profile(temp_p
     )
 
     manager = ContextManager(temp_project)
-    payload = manager.build_context(3, use_snapshot=False, save_snapshot=False)
+    payload = manager.build_context(3)
 
-    assert payload["sections"]["story_contract"]["content"]["master_setting"]["route"]["primary_genre"] == "都市异能"
-    assert payload["sections"]["genre_profile"]["content"]["genre"] == "都市异能"
-    assert payload["sections"]["writing_guidance"]["content"]["signals_used"]["genre"] == "都市异能"
-    assert payload["sections"]["runtime_status"]["content"]["fallback_sources"] == [
+    assert payload["story_contract"]["master_setting"]["route"]["primary_genre"] == "都市异能"
+    assert payload["genre_profile"]["genre"] == "都市异能"
+    assert payload["writing_guidance"]["signals_used"]["genre"] == "都市异能"
+    assert payload["runtime_status"]["fallback_sources"] == [
         "missing_volume_contract",
         "missing_chapter_contract",
         "missing_review_contract",
@@ -380,115 +355,10 @@ def test_context_manager_exposes_latest_rejected_commit_not_last_accepted(temp_p
     )
 
     manager = ContextManager(temp_project)
-    payload = manager.build_context(3, use_snapshot=False, save_snapshot=False)
-
-    assert payload["sections"]["latest_commit"]["content"]["meta"]["status"] == "rejected"
-    assert payload["sections"]["runtime_status"]["content"]["latest_accepted_commit"]["meta"]["status"] == "accepted"
-
-
-def test_context_manager_invalidates_snapshot_when_story_contract_changes(temp_project):
-    state = {
-        "progress": {"volumes_planned": [{"volume": 1, "chapters_range": "1-10"}]},
-        "protagonist_state": {"name": "萧炎"},
-        "chapter_meta": {},
-        "disambiguation_warnings": [],
-        "disambiguation_pending": [],
-    }
-    temp_project.state_file.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
-
-    story_root = temp_project.story_system_dir
-    (story_root / "chapters").mkdir(parents=True, exist_ok=True)
-    (story_root / "volumes").mkdir(parents=True, exist_ok=True)
-    (story_root / "reviews").mkdir(parents=True, exist_ok=True)
-    (story_root / "MASTER_SETTING.json").write_text(
-        json.dumps(
-            {
-                "meta": {"schema_version": "story-system/v1", "contract_type": "MASTER_SETTING"},
-                "route": {"primary_genre": "玄幻退婚流"},
-                "master_constraints": {"core_tone": "先压后爆"},
-                "base_context": [],
-                "source_trace": [],
-                "override_policy": {},
-            },
-            ensure_ascii=False,
-        ),
-        encoding="utf-8",
-    )
-    (story_root / "chapters" / "chapter_003.json").write_text(
-        json.dumps(
-            {
-                "meta": {"schema_version": "story-system/v1", "contract_type": "CHAPTER_BRIEF", "chapter": 3},
-                "override_allowed": {"chapter_focus": "旧焦点"},
-                "dynamic_context": [],
-                "source_trace": [],
-            },
-            ensure_ascii=False,
-        ),
-        encoding="utf-8",
-    )
-    (story_root / "volumes" / "volume_001.json").write_text(
-        json.dumps(
-            {
-                "meta": {"schema_version": "story-system/v1", "contract_type": "VOLUME_BRIEF"},
-                "volume_goal": {"summary": "旧卷目标"},
-                "selected_tropes": [],
-                "selected_pacing": {},
-                "selected_scenes": [],
-                "anti_patterns": [],
-                "system_constraints": [],
-                "overrides": {"locked": {}, "append_only": {}, "override_allowed": {}},
-            },
-            ensure_ascii=False,
-        ),
-        encoding="utf-8",
-    )
-    review_path = story_root / "reviews" / "chapter_003.review.json"
-    review_path.write_text(
-        json.dumps(
-            {
-                "meta": {"schema_version": "story-system/v1", "contract_type": "REVIEW_CONTRACT"},
-                "must_check": ["旧节点"],
-                "blocking_rules": ["旧禁区"],
-                "genre_specific_risks": [],
-                "anti_patterns": [],
-                "system_constraints": [],
-                "review_thresholds": {},
-                "overrides": {"locked": {}, "append_only": {}, "override_allowed": {}},
-            },
-            ensure_ascii=False,
-        ),
-        encoding="utf-8",
-    )
-    temp_project.outline_dir.mkdir(parents=True, exist_ok=True)
-    (temp_project.outline_dir / "第1卷-详细大纲.md").write_text(
-        "### 第3章：试炼\n必须覆盖节点：旧节点\n本章禁区：旧禁区",
-        encoding="utf-8",
-    )
-
-    manager = ContextManager(temp_project)
-    first = manager.build_context(3, template="plot", use_snapshot=True, save_snapshot=True)
-    assert first["sections"]["prewrite_validation"]["content"]["forbidden_zones"] == ["旧禁区"]
-
-    review_path.write_text(
-        json.dumps(
-            {
-                "meta": {"schema_version": "story-system/v1", "contract_type": "REVIEW_CONTRACT"},
-                "must_check": ["新节点"],
-                "blocking_rules": ["新禁区"],
-                "genre_specific_risks": [],
-                "anti_patterns": [],
-                "system_constraints": [],
-                "review_thresholds": {},
-                "overrides": {"locked": {}, "append_only": {}, "override_allowed": {}},
-            },
-            ensure_ascii=False,
-        ),
-        encoding="utf-8",
-    )
+    payload = manager.build_context(3)
 
-    second = manager.build_context(3, template="plot", use_snapshot=True, save_snapshot=False)
-    assert second["sections"]["story_contract"]["content"]["review_contract"]["blocking_rules"] == ["新禁区"]
-    assert second["sections"]["prewrite_validation"]["content"]["forbidden_zones"] == ["新禁区"]
+    assert payload["latest_commit"]["meta"]["status"] == "rejected"
+    assert payload["runtime_status"]["latest_accepted_commit"]["meta"]["status"] == "accepted"
 
 
 def test_context_manager_blocks_when_story_contract_missing(temp_project):
@@ -501,9 +371,9 @@ def test_context_manager_blocks_when_story_contract_missing(temp_project):
     temp_project.state_file.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
 
     manager = ContextManager(temp_project)
-    payload = manager.build_context(3, use_snapshot=False, save_snapshot=False)
+    payload = manager.build_context(3)
 
-    prewrite = payload["sections"]["prewrite_validation"]["content"]
+    prewrite = payload["prewrite_validation"]
     assert prewrite["blocking"] is True
     assert any("合同" in reason for reason in prewrite["blocking_reasons"])
 
@@ -523,78 +393,6 @@ def test_query_router():
     assert "A" in router.split("A, B；C")
 
 
-def test_context_snapshot_respects_template(temp_project):
-    state = {
-        "protagonist_state": {"name": "萧炎"},
-        "chapter_meta": {},
-        "disambiguation_warnings": [],
-        "disambiguation_pending": [],
-    }
-    temp_project.state_file.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
-
-    manager = ContextManager(temp_project)
-
-    plot_payload = manager.build_context(1, template="plot", use_snapshot=True, save_snapshot=True)
-    battle_payload = manager.build_context(1, template="battle", use_snapshot=True, save_snapshot=True)
-
-    assert plot_payload.get("template") == "plot"
-    assert battle_payload.get("template") == "battle"
-
-
-def test_context_snapshot_invalidates_legacy_version(temp_project):
-    state = {
-        "project": {"genre": "xuanhuan"},
-        "protagonist_state": {"name": "萧炎"},
-        "chapter_meta": {},
-        "disambiguation_warnings": [],
-        "disambiguation_pending": [],
-    }
-    temp_project.state_file.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
-    temp_project.outline_dir.mkdir(parents=True, exist_ok=True)
-    (temp_project.outline_dir / "第1卷-详细大纲.md").write_text(
-        """### 第4章：试炼
-CBN：进入试炼场
-CPNs：
-- 观察规则
-CEN：决定将计就计
-""",
-        encoding="utf-8",
-    )
-
-    snapshot_path = temp_project.webnovel_dir / "context_snapshots" / "ch0004.json"
-    snapshot_path.parent.mkdir(parents=True, exist_ok=True)
-    snapshot_path.write_text(
-        json.dumps(
-            {
-                "version": "1.2",
-                "chapter": 4,
-                "saved_at": "2026-03-01T00:00:00+00:00",
-                "meta": {"template": "plot"},
-                "payload": {
-                    "meta": {"chapter": 4},
-                    "sections": {
-                        "core": {
-                            "content": {"chapter_outline": "旧快照"},
-                            "text": "{}",
-                            "budget": 1000,
-                        }
-                    },
-                    "template": "plot",
-                    "weights": {},
-                },
-            },
-            ensure_ascii=False,
-        ),
-        encoding="utf-8",
-    )
-
-    manager = ContextManager(temp_project)
-    payload = manager.build_context(4, template="plot", use_snapshot=True, save_snapshot=False)
-
-    assert payload["sections"]["core"]["content"]["chapter_outline"] != "旧快照"
-    assert payload["sections"]["plot_structure"]["content"]["cbn"] == "进入试炼场"
-
-
 def test_context_manager_applies_ranker_and_contract_meta(temp_project):
     state = {
         "protagonist_state": {"name": "萧炎"},
@@ -611,14 +409,14 @@ def test_context_manager_applies_ranker_and_contract_meta(temp_project):
     temp_project.state_file.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
 
     manager = ContextManager(temp_project)
-    payload = manager.build_context(4, use_snapshot=False, save_snapshot=False)
+    payload = manager.build_context(4)
 
-    assert payload["meta"].get("context_contract_version") == "v2"
-    recent_meta = payload["sections"]["core"]["content"]["recent_meta"]
+    assert payload["meta"].get("context_contract_version") == "v3"
+    recent_meta = payload["core"]["recent_meta"]
     if recent_meta:
         assert recent_meta[0]["chapter"] == 3
 
-    warnings = payload["sections"]["alerts"]["content"]["disambiguation_warnings"]
+    warnings = payload["alerts"]["disambiguation_warnings"]
     if warnings and isinstance(warnings[0], dict):
         assert "critical" in str(warnings[0].get("message", "")) or warnings[0].get("severity") == "high"
 
@@ -654,16 +452,16 @@ def test_context_manager_includes_reader_signal_and_genre_profile(temp_project):
     )
 
     manager = ContextManager(temp_project)
-    payload = manager.build_context(4, use_snapshot=False, save_snapshot=False)
+    payload = manager.build_context(4)
 
-    reader_signal = payload["sections"]["reader_signal"]["content"]
+    reader_signal = payload["reader_signal"]
     assert "recent_reading_power" in reader_signal
     assert "pattern_usage" in reader_signal
     assert "hook_type_usage" in reader_signal
     assert "review_trend" in reader_signal
     assert isinstance(reader_signal.get("low_score_ranges"), list)
 
-    genre_profile = payload["sections"]["genre_profile"]["content"]
+    genre_profile = payload["genre_profile"]
     assert genre_profile.get("genre") == "xuanhuan"
     assert "profile_excerpt" in genre_profile
     assert "taxonomy_excerpt" in genre_profile
@@ -752,9 +550,9 @@ def test_context_manager_includes_writing_guidance(temp_project):
     )
 
     manager = ContextManager(temp_project)
-    payload = manager.build_context(4, use_snapshot=False, save_snapshot=False)
+    payload = manager.build_context(4)
 
-    guidance = payload["sections"]["writing_guidance"]["content"]
+    guidance = payload["writing_guidance"]
     assert guidance.get("chapter") == 4
     items = guidance.get("guidance_items") or []
     assert isinstance(items, list)
@@ -811,15 +609,13 @@ def test_context_manager_dynamic_weights_and_composite_genre(temp_project):
     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)
+    payload_early = manager.build_context(10, template="plot")
+    payload_late = manager.build_context(150, template="plot")
 
-    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"]
+    profile = payload_early["genre_profile"]
     assert profile.get("composite") is True
     assert profile.get("genre") == "xuanhuan"
     assert isinstance(profile.get("genres"), list)
@@ -862,8 +658,8 @@ def test_context_manager_genre_alias_guidance_and_heading_extraction(temp_projec
     temp_project.state_file.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
 
     manager = ContextManager(temp_project)
-    payload = manager.build_context(12, template="plot", use_snapshot=False, save_snapshot=False)
-    guidance = payload["sections"]["writing_guidance"]["content"]
+    payload = manager.build_context(12, template="plot")
+    guidance = payload["writing_guidance"]
     items = guidance.get("guidance_items") or []
 
     assert any("战术决策点" in str(text) for text in items)
@@ -919,8 +715,8 @@ def test_context_manager_genre_aliases_normalized_for_profile_lookup(temp_projec
     }
     temp_project.state_file.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
 
-    payload = manager.build_context(20, template="plot", use_snapshot=False, save_snapshot=False)
-    profile = payload["sections"]["genre_profile"]["content"]
+    payload = manager.build_context(20, template="plot")
+    profile = payload["genre_profile"]
 
     assert profile.get("genre") == "电竞"
     assert "直播文" in (profile.get("genres") or [])
@@ -938,9 +734,9 @@ def test_context_manager_enables_methodology_for_xianxia(temp_project):
 
     manager = ContextManager(temp_project)
     manager.config.context_writing_checklist_max_items = 8
-    payload = manager.build_context(21, template="plot", use_snapshot=False, save_snapshot=False)
+    payload = manager.build_context(21, template="plot")
 
-    guidance = payload["sections"]["writing_guidance"]["content"]
+    guidance = payload["writing_guidance"]
     strategy = guidance.get("methodology") or {}
     assert strategy.get("enabled") is True
     assert strategy.get("pilot") == "xianxia"
@@ -960,9 +756,9 @@ def test_context_manager_enables_methodology_for_non_xianxia_by_default(temp_pro
     temp_project.state_file.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
 
     manager = ContextManager(temp_project)
-    payload = manager.build_context(21, template="plot", use_snapshot=False, save_snapshot=False)
+    payload = manager.build_context(21, template="plot")
 
-    guidance = payload["sections"]["writing_guidance"]["content"]
+    guidance = payload["writing_guidance"]
     strategy = guidance.get("methodology") or {}
     assert strategy.get("enabled") is True
     assert strategy.get("genre_profile_key") == "xuanhuan"
@@ -981,30 +777,14 @@ def test_context_manager_allows_methodology_whitelist_restriction(temp_project):
 
     manager = ContextManager(temp_project)
     manager.config.context_methodology_genre_whitelist = ("xianxia",)
-    payload = manager.build_context(21, template="plot", use_snapshot=False, save_snapshot=False)
+    payload = manager.build_context(21, template="plot")
 
-    guidance = payload["sections"]["writing_guidance"]["content"]
+    guidance = payload["writing_guidance"]
     strategy = guidance.get("methodology") or {}
     assert strategy == {}
     assert guidance.get("signals_used", {}).get("methodology_enabled") is False
 
 
-def test_context_manager_compact_text_truncation(temp_project):
-    manager = ContextManager(temp_project)
-    manager.config.context_compact_text_enabled = True
-    manager.config.context_compact_min_budget = 80
-    manager.config.context_compact_head_ratio = 0.6
-
-    content = {"a": "x" * 200, "b": "y" * 200}
-    compact = manager._compact_json_text(content, budget=120)
-    assert len(compact) <= 120
-    assert "[TRUNCATED]" in compact
-
-    manager.config.context_compact_text_enabled = False
-    raw_cut = manager._compact_json_text(content, budget=100)
-    assert len(raw_cut) <= 100
-
-
 def test_context_manager_persist_writing_checklist_score_logs_failure(temp_project, monkeypatch, caplog):
     manager = ContextManager(temp_project)
 
@@ -1119,9 +899,9 @@ CEN：决定将计就计
     )
 
     manager = ContextManager(temp_project)
-    payload = manager.build_context(4, use_snapshot=False, save_snapshot=False)
+    payload = manager.build_context(4)
 
-    plot_structure = payload["sections"]["plot_structure"]["content"]
+    plot_structure = payload["plot_structure"]
     assert plot_structure.get("cbn") == "进入试炼场"
     assert plot_structure.get("cpns") == ["观察规则", "发现陷阱"]
     assert plot_structure.get("cen") == "决定将计就计"

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

@@ -207,7 +207,7 @@ 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["context_contract_version"] == "v3"
     assert payload.get("context_weight_stage") in {"early", "mid", "late"}
     assert "writing_guidance" in payload
     assert isinstance(payload["writing_guidance"].get("guidance_items"), list)
@@ -358,19 +358,16 @@ def test_render_text_contains_writing_guidance_section(tmp_path):
     }
 
     text = _render_text(payload)
-    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
-    assert "## 长篇方法论策略" in text
-    assert "- 适用题材: xianxia" in text
-    assert "next_reason=78.0" in text
+    parsed = json.loads(text)
+    assert "writing_guidance" in parsed
+    assert parsed["writing_guidance"]["guidance_items"][0] == "先修低分"
+    assert parsed["context_contract_version"] == "v2"
+    assert parsed["context_weight_stage"] == "early"
+    assert parsed["writing_guidance"]["checklist"][0]["label"] == "修复低分区间问题"
+    assert parsed["writing_guidance"]["checklist_score"]["score"] == 81.5
+    assert parsed["genre_profile"]["genres"] == ["xuanhuan", "realistic"]
+    assert parsed["writing_guidance"]["methodology"]["enabled"] is True
+    assert parsed["writing_guidance"]["methodology"]["genre_profile_key"] == "xianxia"
 
 
 def test_render_text_contains_rag_assist_section_when_hits_exist(tmp_path):
@@ -407,10 +404,11 @@ def test_render_text_contains_rag_assist_section_when_hits_exist(tmp_path):
     }
 
     text = _render_text(payload)
-    assert "## RAG 检索线索" in text
-    assert "- 模式: auto" in text
-    assert "[graph_hybrid]" in text
-    assert "萧炎与药老" in text
+    parsed = json.loads(text)
+    assert parsed["rag_assist"]["invoked"] is True
+    assert parsed["rag_assist"]["mode"] == "auto"
+    assert parsed["rag_assist"]["hits"][0]["source"] == "graph_hybrid"
+    assert "萧炎与药老" in parsed["rag_assist"]["hits"][0]["content"]
 
 
 def test_build_chapter_context_payload_includes_plot_structure(tmp_path):
@@ -483,11 +481,11 @@ def test_render_text_contains_plot_structure_section(tmp_path):
     }
 
     text = _render_text(payload)
-    assert "## 情节结构" in text
-    assert "- CBN: 主角进入遗迹" in text
-    assert "- CPN1: 发现石碑异常" in text
-    assert "- CEN: 决定深入遗迹核心" in text
-    assert "- 本章禁区: 不能提前拿到终极传承" in text
+    parsed = json.loads(text)
+    assert parsed["plot_structure"]["cbn"] == "主角进入遗迹"
+    assert parsed["plot_structure"]["cpns"] == ["发现石碑异常", "与守卫短暂交锋"]
+    assert parsed["plot_structure"]["cen"] == "决定深入遗迹核心"
+    assert "不能提前拿到终极传承" in parsed["plot_structure"]["prohibitions"]
 
 
 def test_render_text_contains_contract_first_runtime_section(tmp_path):
@@ -522,9 +520,9 @@ def test_render_text_contains_contract_first_runtime_section(tmp_path):
     }
 
     text = _render_text(payload)
-    assert "## Contract-First Runtime" in text
-    assert "- Review blocking rules: 2" in text
-    assert "- Prewrite blocking: False" in text
+    parsed = json.loads(text)
+    assert len(parsed["story_contract"]["review_contract"]["blocking_rules"]) == 2
+    assert parsed["prewrite_validation"]["blocking"] is False
 
 
 def test_render_text_contains_runtime_status_section(tmp_path):
@@ -552,6 +550,6 @@ def test_render_text_contains_runtime_status_section(tmp_path):
         }
     )
 
-    assert "## Runtime Status" in text
-    assert "- 写后事实入口: chapter_commit" in text
-    assert "- Legacy Fallback: missing_accepted_commit" in text
+    parsed = json.loads(text)
+    assert parsed["runtime_status"]["primary_write_source"] == "chapter_commit"
+    assert parsed["runtime_status"]["fallback_sources"] == ["missing_accepted_commit"]

webnovel-writer/scripts/extract_chapter_context.py

@@ -299,29 +299,22 @@ def _load_contract_context(project_root: Path, chapter_num: int) -> Dict[str, An
 
     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", {})
+    payload = manager.build_context(chapter=chapter_num, template="plot")
+
     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", {}),
-        "runtime_status": (sections.get("runtime_status") or {}).get("content", {}),
-        "latest_commit": (sections.get("latest_commit") 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", {}),
-        "plot_structure": (sections.get("plot_structure") or {}).get("content", {}),
-        "long_term_memory": (sections.get("long_term_memory") or {}).get("content", {}),
-        "scene": (sections.get("scene") or {}).get("content", {}),
-        "core": (sections.get("core") or {}).get("content", {}),
+        "story_contract": payload.get("story_contract", {}),
+        "runtime_status": payload.get("runtime_status", {}),
+        "latest_commit": payload.get("latest_commit", {}),
+        "prewrite_validation": payload.get("prewrite_validation", {}),
+        "reader_signal": payload.get("reader_signal", {}),
+        "genre_profile": payload.get("genre_profile", {}),
+        "writing_guidance": payload.get("writing_guidance", {}),
+        "plot_structure": payload.get("plot_structure", {}),
+        "long_term_memory": payload.get("long_term_memory", {}),
+        "scene": payload.get("scene", {}),
+        "core": payload.get("core", {}),
     }
 
 
@@ -362,243 +355,8 @@ def build_chapter_context_payload(project_root: Path, chapter_num: int) -> Dict[
 
 
 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("")
-        stage = payload.get("context_weight_stage")
-        if stage:
-            lines.append(f"- 上下文阶段权重: {stage}")
-            lines.append("")
-
-    runtime_status = payload.get("runtime_status") or {}
-    latest_commit = payload.get("latest_commit") or {}
-    if runtime_status or latest_commit:
-        fallback_sources = runtime_status.get("fallback_sources") or ["none"]
-        lines.append("## Runtime Status")
-        lines.append("")
-        lines.append(f"- 写后事实入口: {runtime_status.get('primary_write_source', 'unknown')}")
-        lines.append(f"- Legacy Fallback: {', '.join(str(item) for item in fallback_sources)}")
-        lines.append(f"- Latest Commit: {(latest_commit.get('meta') or {}).get('status', 'missing')}")
-        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("## 情节结构")
-        lines.append("")
-        cbn = str(plot_structure.get("cbn") or "").strip()
-        if cbn:
-            lines.append(f"- CBN: {cbn}")
-        for idx, item in enumerate(plot_structure.get("cpns") or [], start=1):
-            lines.append(f"- CPN{idx}: {item}")
-        cen = str(plot_structure.get("cen") or "").strip()
-        if cen:
-            lines.append(f"- CEN: {cen}")
-        mandatory_nodes = plot_structure.get("mandatory_nodes") or []
-        if mandatory_nodes:
-            lines.append("- 必须覆盖节点: " + " | ".join(str(x) for x in mandatory_nodes))
-        prohibitions = plot_structure.get("prohibitions") or []
-        if prohibitions:
-            lines.append("- 本章禁区: " + " | ".join(str(x) for x in prohibitions))
-        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 {}
-    methodology = writing_guidance.get("methodology") or {}
-    if guidance_items or checklist:
-        lines.append("## 写作执行建议")
-        lines.append("")
-        for idx, item in enumerate(guidance_items, start=1):
-            lines.append(f"{idx}. {item}")
-
-        if checklist:
-            total_weight = 0.0
-            required_count = 0
-            for row in checklist:
-                if isinstance(row, dict):
-                    try:
-                        total_weight += float(row.get("weight") or 0)
-                    except (TypeError, ValueError):
-                        pass
-                    if row.get("required"):
-                        required_count += 1
-
-            lines.append("")
-            lines.append("### 执行检查清单（可评分）")
-            lines.append("")
-            lines.append(f"- 项目数: {len(checklist)}")
-            lines.append(f"- 总权重: {total_weight:.2f}")
-            lines.append(f"- 必做项: {required_count}")
-            lines.append("")
-
-            for idx, row in enumerate(checklist, start=1):
-                if not isinstance(row, dict):
-                    lines.append(f"{idx}. {row}")
-                    continue
-                label = str(row.get("label") or "").strip() or "未命名项"
-                weight = row.get("weight")
-                required_tag = "必做" if row.get("required") else "可选"
-                verify_hint = str(row.get("verify_hint") or "").strip()
-                lines.append(f"{idx}. [{required_tag}][w={weight}] {label}")
-                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("")
-
-    if isinstance(methodology, dict) and methodology.get("enabled"):
-        lines.append("## 长篇方法论策略")
-        lines.append("")
-        lines.append(f"- 框架: {methodology.get('framework')}")
-        methodology_scope = methodology.get("genre_profile_key") or methodology.get("pilot") or "general"
-        lines.append(f"- 适用题材: {methodology_scope}")
-        lines.append(f"- 章节阶段: {methodology.get('chapter_stage')}")
-        observability = methodology.get("observability") or {}
-        if observability:
-            lines.append(
-                "- 指标: "
-                f"next_reason={observability.get('next_reason_clarity')}, "
-                f"anchor={observability.get('anchor_effectiveness')}, "
-                f"rhythm={observability.get('rhythm_naturalness')}"
-            )
-        signals = methodology.get("signals") or {}
-        risk_flags = list(signals.get("risk_flags") or [])
-        if risk_flags:
-            lines.append(f"- 风险标记: {', '.join(str(flag) for flag in risk_flags)}")
-        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')}")
-        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}")
-        lines.append("")
-
-    long_term_memory = payload.get("long_term_memory") or {}
-    if long_term_memory:
-        lines.append("## 长期记忆")
-        lines.append("")
-        stats = long_term_memory.get("stats") or {}
-        if stats:
-            lines.append(
-                f"- 注入条目: {stats.get('injected', 0)} / 总条目: {stats.get('total', 0)}"
-            )
-        active_constraints = long_term_memory.get("active_constraints") or []
-        if active_constraints:
-            lines.append("- 活跃约束:")
-            for row in active_constraints[:5]:
-                value = str(row.get("value", "") or "").strip()
-                subject = str(row.get("subject", "") or "").strip()
-                if subject:
-                    lines.append(f"  - [{subject}] {value}")
-                else:
-                    lines.append(f"  - {value}")
-        facts = long_term_memory.get("long_term_facts") or []
-        if facts:
-            lines.append("- 关键长期事实:")
-            for row in facts[:5]:
-                category = str(row.get("category", "") or "").strip()
-                subject = str(row.get("subject", "") or "").strip()
-                field = str(row.get("field", "") or "").strip()
-                value = str(row.get("value", "") or "").strip()
-                lines.append(f"  - ({category}) {subject}.{field} = {value}")
-        lines.append("")
-
-    rag_assist = payload.get("rag_assist") or {}
-    hits = rag_assist.get("hits") or []
-    if rag_assist.get("invoked") and hits:
-        lines.append("## RAG 检索线索")
-        lines.append("")
-        lines.append(f"- 模式: {rag_assist.get('mode')}")
-        lines.append(f"- 意图: {rag_assist.get('intent')}")
-        lines.append(f"- 查询: {rag_assist.get('query')}")
-        lines.append("")
-        for idx, row in enumerate(hits[:5], start=1):
-            chapter = row.get("chapter", "?")
-            scene_index = row.get("scene_index", "?")
-            score = row.get("score", 0)
-            source = row.get("source", "unknown")
-            content = row.get("content", "")
-            lines.append(f"{idx}. [Ch{chapter}-S{scene_index}][{source}][score={score}] {content}")
-        lines.append("")
-
-    return "\n".join(lines).rstrip() + "\n"
+    """JSON 序列化输出，text 渲染由 context-agent 负责。"""
+    return json.dumps(payload, ensure_ascii=False, indent=2)
 
 
 def main():