Преглед изворни кода

feat: add structured chapter context and review pipeline

lingfengQAQ пре 3 месеци
родитељ
комит
67db1c7d99
25 измењених фајлова са 2095 додато и 32 уклоњено
  1. 10 0
      docs/README.md
  2. 406 0
      docs/storyteller-implementation-plan.md
  3. 438 0
      docs/storyteller-paper-summary.md
  4. 43 1
      webnovel-writer/agents/consistency-checker.md
  5. 23 6
      webnovel-writer/agents/context-agent.md
  6. 45 0
      webnovel-writer/agents/continuity-checker.md
  7. 1 0
      webnovel-writer/agents/data-agent.md
  8. 93 0
      webnovel-writer/scripts/chapter_outline_loader.py
  9. 9 2
      webnovel-writer/scripts/data_modules/context_manager.py
  10. 193 1
      webnovel-writer/scripts/data_modules/index_manager.py
  11. 4 0
      webnovel-writer/scripts/data_modules/schemas.py
  12. 2 1
      webnovel-writer/scripts/data_modules/state_manager.py
  13. 39 0
      webnovel-writer/scripts/data_modules/state_validator.py
  14. 34 0
      webnovel-writer/scripts/data_modules/tests/test_context_manager.py
  15. 182 0
      webnovel-writer/scripts/data_modules/tests/test_data_modules.py
  16. 77 0
      webnovel-writer/scripts/data_modules/tests/test_extract_chapter_context.py
  17. 0 1
      webnovel-writer/scripts/data_modules/tests/test_state_validator.py
  18. 223 0
      webnovel-writer/scripts/data_modules/tests/test_webnovel_unified_cli.py
  19. 21 0
      webnovel-writer/scripts/data_modules/tests/test_workflow_manager.py
  20. 20 0
      webnovel-writer/scripts/data_modules/webnovel.py
  21. 24 1
      webnovel-writer/scripts/extract_chapter_context.py
  22. 86 0
      webnovel-writer/scripts/review_pipeline.py
  23. 13 2
      webnovel-writer/scripts/workflow_manager.py
  24. 50 4
      webnovel-writer/skills/webnovel-plan/SKILL.md
  25. 59 13
      webnovel-writer/skills/webnovel-write/SKILL.md

+ 10 - 0
docs/README.md

@@ -7,6 +7,8 @@
 - [LIGHT 改造计划](#light-改造计划)
 - [长期记忆研究报告](#长期记忆研究报告)
 - [长期记忆新架构](#长期记忆新架构)
+- [STORYTELLER 论文总结](#storyteller-论文总结)
+- [STORYTELLER 落地方案](#storyteller-落地方案)
 - [命令详解](#命令详解)
 - [RAG 与配置](#rag-与配置)
 - [题材模板](#题材模板)
@@ -32,6 +34,14 @@
 
 - `long-term-memory-architecture-v2.md`:基于调研结论重规划的长期记忆架构
 
+## STORYTELLER 论文总结
+
+- `storyteller-paper-summary.md`:STORYTELLER 论文方法、模块、流程与项目启发总结
+
+## STORYTELLER 落地方案
+
+- `storyteller-implementation-plan.md`:基于当前项目的 STORYTELLER 结构化章纲落地实施方案
+
 ## 命令详解
 
 - `commands.md`:`/webnovel-*` 命令详细说明

+ 406 - 0
docs/storyteller-implementation-plan.md

@@ -0,0 +1,406 @@
+# STORYTELLER 落地实施方案
+
+## 文档目标
+
+本文档基于前面对 `STORYTELLER` 论文、当前 `webnovel-writer` 架构和落地边界的讨论,给出一份从 `skills` 与 `agents` 视角出发的修正版实施方案。
+
+目标不是复制论文原系统,而是以最小破坏、最高复用的方式,将“中层情节结构”前移到规划阶段,并让写作、审查、数据回写围绕结构化章纲形成闭环。
+
+## 一句话结论
+
+最优方案不是新增独立的 `webnovel-structure` 主流程,而是:
+
+- 升级 `webnovel-plan`,输出结构化详细大纲
+- 增强 `context-agent`,把结构化章纲组装进写作包
+- 增强 `continuity-checker` 与 `consistency-checker`,分别负责节点覆盖与结构冲突
+- 增强 `data-agent`,负责落库覆盖结果与偏差说明
+
+这样可以在不打碎现有系统的前提下,为写作链补上 `STORYTELLER` 最核心的中层结构层。
+
+## 核心原则
+
+1. 不新增独立主流程 `webnovel-structure`
+2. 结构前移到 `webnovel-plan`
+3. `context-agent` 只聚合,不生成节点
+4. `continuity-checker` 负责结构覆盖
+5. `consistency-checker` 负责结构冲突
+6. `data-agent` 负责落库,不做主判断
+7. 全程向后兼容,无节点章纲照常运行
+
+## 总体架构
+
+```text
+/webnovel-init
+    -> 初始化项目骨架与状态
+
+/webnovel-plan
+    -> 节拍表
+    -> 时间线
+    -> 结构化详细大纲(新增节点字段)
+
+/webnovel-write
+    -> Step 0.5 轻量节点预检
+    -> Step 1 context-agent 组装写作包
+    -> Step 2A writer 按节点扩写
+    -> Step 3 review agents
+    -> Step 5 data-agent 持久化覆盖与偏差
+```
+
+## Skills 视角实施方案
+
+### 1. `webnovel-init`
+
+定位:保留,不做大改。
+
+职责:
+
+- 初始化 `.webnovel/state.json`
+- 初始化 `index.db / vectors.db`
+- 初始化 `大纲/`、`设定集/`、摘要目录等骨架
+
+建议新增但不强制:
+
+- 为后续结构化章纲预留约定说明
+- 为 `index.db` 预留结构化章纲缓存或索引的升级入口
+
+### 2. `webnovel-plan`
+
+定位:本次改造的主战场。
+
+目标:把现有“详细大纲”升级成“结构化详细大纲”。
+
+现有输出保留:
+
+- `大纲/第{volume_id}卷-节拍表.md`
+- `大纲/第{volume_id}卷-时间线.md`
+- `大纲/第{volume_id}卷-详细大纲.md`
+
+每章新增字段:
+
+- `章节起点(CBN)`
+- `推进节点(CPNs)`
+- `章节终点(CEN)`
+- `必须覆盖节点`
+- `本章禁区`
+
+节点格式建议:
+
+`主体 | 动作/变化 | 对象/结果`
+
+示例:
+
+- `萧炎 | 抵达 | 迦南学院入口`
+- `萧炎 | 展示 | 异火控制力`
+- `药老 | 对萧炎产生 | 明确兴趣`
+
+节点数量约束:
+
+- `1 个 CBN`
+- `2-4 个 CPN`
+- `1 个 CEN`
+
+必须覆盖规则:
+
+- 每章必须覆盖节点最多 `4` 个
+- 建议为:`CBN + CEN + 1~2 个核心 CPN`
+
+章间衔接规则:
+
+- 相邻章节 `CEN -> 下一章 CBN` 必须逻辑承接
+- 若为时间跳转章,需在时间字段中明确说明
+
+本章禁区规则:
+
+- 不超过 `5` 条
+- 只写硬禁区,不写风格建议
+
+批次建议:
+
+- 默认 `10章/批`
+- 复杂题材 `8章/批`
+- 简单升级流上限 `12章/批`
+
+不建议:
+
+- 再单独新增常规 `/webnovel-structure`
+- 在 `plan` 之外再维护一份平行节点文件作为主数据源
+
+### 3. `webnovel-write`
+
+定位:保留为唯一主写作入口。
+
+建议流程:
+
+1. `Step 0.5` 轻量节点预检
+2. `Step 1` `context-agent` 读取结构化章纲、状态和记忆,组装写作包
+3. `Step 2A` writer 按 `CBN -> CPNs -> CEN` 扩写正文
+4. `Step 3` review agents 执行结构覆盖与设定一致性检查
+5. `Step 5` `data-agent` 持久化覆盖结果与偏差说明
+
+新增 Step 0.5 的边界:
+
+- 只检查主角或 POV 角色相关节点
+- 第一版仅检查地点、境界/实力层级
+- 仅附加警告,不阻断流程
+- 不做复杂关系推理或自动重规划
+
+### 4. `webnovel-review`
+
+定位:保留,主要用于区间复盘和独立质量检查。
+
+建议增强:
+
+- 支持读取结构化章纲中的节点字段
+- 支持基于章节范围回看 `plan vs actual` 偏移趋势
+
+## Agents 视角实施方案
+
+### 1. `context-agent`
+
+定位:保留,增强为“结构化章纲聚合器”。
+
+新增读取:
+
+- `CBN`
+- `CPNs`
+- `CEN`
+- `必须覆盖节点`
+- `本章禁区`
+
+新增输出板块:
+
+- `情节结构`
+  - `章节起点`
+  - `推进节点`
+  - `章节终点`
+  - `必须覆盖节点`
+  - `本章禁区`
+
+Context Contract 新增字段:
+
+- `plot_structure`
+  - `cbn`
+  - `cpns`
+  - `cen`
+  - `mandatory_nodes`
+  - `prohibitions`
+
+Step 2A 节拍映射:
+
+- 有节点时:`CBN触发 -> CPN推进 -> CPN受阻/变化 -> CEN收束 -> 章末钩子`
+- 无节点时:保持旧节拍
+
+红线新增:
+
+- 情节结构与任务书方向冲突
+
+边界:
+
+- 不生成节点
+- 不修改节点
+- 只负责读取、聚合、压缩成写作包
+
+### 2. `plot-node-agent`
+
+当前判断:第一阶段不作为必须组件。
+
+原因:
+
+- 既然决定把结构前移到 `plan`
+- 那 `CBN/CPN/CEN` 的主产物应由 `plan` 直接生成
+- 不需要在 `write` 阶段再新增重量级 agent 重复规划
+
+保留可能性:
+
+- 若后续发现章纲与正文脱节严重
+- 再考虑引入轻量 `plot-node-agent`
+- 只做写前局部修正,不做整章重规划
+
+### 3. `continuity-checker`
+
+定位:负责“结构覆盖”和“事件承接”。
+
+新增检查:
+
+1. `CBN 承接`
+2. `CEN 落地`
+3. `必须节点覆盖`
+4. `可选节点覆盖(仅统计)`
+
+输出:
+
+- 节点覆盖表
+- 覆盖评级:`A / B / C / F`
+
+成功标准:
+
+- 有节点时,覆盖评级至少 `B`
+
+边界:
+
+- 不判断设定是否越界
+- 不判断禁区是否违背世界规则
+- 只判断“写到了没有、接顺了没有、收住了没有”
+
+### 4. `consistency-checker`
+
+定位:负责“结构与设定冲突”。
+
+新增检查:
+
+1. 是否违反 `本章禁区`
+2. 正文开头方向是否与 `CBN` 冲突
+3. 正文结尾方向是否与 `CEN` 冲突
+4. 关键节点对应的能力、身份、地点是否违背设定
+
+严重度建议:
+
+- 禁区违反:`high`
+- `CEN` 方向冲突:`medium`
+- `CBN` 方向冲突:`medium`
+
+边界:
+
+- 不负责覆盖率评级
+- 不与 `continuity-checker` 重叠
+
+### 5. `ooc-checker`
+
+定位:保留。
+
+增强方式:
+
+- 读取更结构化的角色状态上下文
+- 结合当前章的节点、角色行为红线和本章变化判断 OOC 风险
+
+### 6. `reader-pull-checker / high-point-checker / pacing-checker`
+
+定位:保留,但执行层收口。
+
+建议:
+
+- 默认写作流程强制启用:
+  - `consistency-checker`
+  - `continuity-checker`
+  - `ooc-checker`
+- 其余按题材、章节类型或区间复盘按需启用
+
+原因:
+
+- 新架构已经把部分“晚审”前移到 `plan`
+- 再全量六并发常驻会过重
+
+### 7. `data-agent`
+
+定位:从“主判断者”改成“持久化与偏差记录器”。
+
+继续保留:
+
+- 实体抽取
+- 状态变更
+- 关系更新
+- 摘要生成
+- `memory facts`
+- 向量切片
+
+新增 Step 5B:
+
+- 写入 `plot_coverage`
+- 写入 `plan_deviation_note`
+
+关键规则:
+
+- `data-agent` 不自己重复做完整覆盖判断
+- 优先消费 `continuity-checker` 与 `consistency-checker` 的结果,再写入 `chapter_meta`
+
+推荐写入结构:
+
+```json
+{
+  "plot_coverage": {
+    "cbn_covered": true,
+    "cen_covered": true,
+    "mandatory_hit_rate": 1.0,
+    "coverage_grade": "A",
+    "prohibitions_violated": []
+  },
+  "plan_deviation_note": ""
+}
+```
+
+填写规则:
+
+- `mandatory_hit_rate < 0.8` 时必须填写 `plan_deviation_note`
+- 存在禁区违反时必须填写 `plan_deviation_note`
+
+边界:
+
+- 不重规划后续章纲
+- 不做复杂 SVO 主链重建
+- 不代替 review
+
+## 实施顺序
+
+### Phase 1:最小闭环
+
+1. `skills/webnovel-plan/SKILL.md`
+2. `agents/context-agent.md`
+3. `skills/webnovel-write/SKILL.md`
+4. `agents/continuity-checker.md`
+5. `agents/consistency-checker.md`
+6. `agents/data-agent.md`
+
+原因:
+
+- 所有下游都依赖 `plan` 的新字段
+- 其余模块都只是消费这些字段
+
+### Phase 2:轻量状态增强
+
+1. 在 `index.db` 中补充结构化章纲缓存或索引
+2. 为 `context-agent` 增加“最近实际承接摘要”读取
+3. 为 `review` 增加区间级偏移复盘
+
+### Phase 3:按需局部修正
+
+1. 若出现大量“章纲和正文脱节”
+2. 再评估是否引入轻量 `plot-node-agent`
+3. 仅做写前局部修正,不做完整重规划
+
+## 验证方案
+
+1. **结构化章纲验证**
+   - 执行 `/webnovel-plan`
+   - 确认详细大纲新增:
+     - `CBN`
+     - `CPNs`
+     - `CEN`
+     - `必须覆盖节点`
+     - `本章禁区`
+
+2. **写作包验证**
+   - 执行 `/webnovel-write`
+   - 确认 `context-agent` 输出包含“情节结构”板块
+
+3. **审查验证**
+   - 确认 `continuity-checker` 输出节点覆盖评级
+   - 确认 `consistency-checker` 输出禁区/方向冲突检查
+
+4. **数据回写验证**
+   - 确认 `chapter_meta` 写入 `plot_coverage`
+   - 确认偏差时有 `plan_deviation_note`
+
+5. **向后兼容验证**
+   - 用旧章纲执行 `/webnovel-write`
+   - 全流程不阻断、不报错
+
+## 最终判断
+
+这版方案的重点不是“多加一个系统层”,而是:
+
+- 让 `plan` 产出更可执行的章纲
+- 让 `context-agent` 成为结构化章纲聚合器
+- 让 `review` 围绕节点覆盖与结构冲突形成清晰分工
+- 让 `data-agent` 成为持久化和偏差记录器
+
+这条路线既吸收了 `STORYTELLER` 的核心思想,又避免了额外主流程和重复组件的膨胀。

+ 438 - 0
docs/storyteller-paper-summary.md

@@ -0,0 +1,438 @@
+# STORYTELLER 论文总结
+
+## 文档目标
+
+本文档基于原论文 `STORYTELLER: An Enhanced Plot-Planning Framework for Coherent and Cohesive Story Generation` 进行整理,重点回答以下问题:
+
+- 这篇工作到底解决了什么问题
+- 它的核心方法由哪些模块组成
+- 三阶段生成流程是如何运作的
+- 论文实验结果说明了什么
+- 对当前 `webnovel-writer` 有哪些直接可借鉴之处
+
+说明:
+
+- 本文优先依据 ACL Anthology 上的论文原文整理
+- 内容偏工程视角,不做逐节逐句翻译
+- 截止本次整理时间,未检索到论文官方公开代码仓库
+
+## 一句话结论
+
+`STORYTELLER` 的核心价值,不是换了一个更大的模型,而是把长篇故事生成改造成了一个 `结构化剧情规划 + 实体关系图维护 + 逐章节审查修正` 的闭环系统。
+
+它通过 `SVO` 事件节点、`STORYLINE` 时间线和 `NEKG` 叙事实体知识图谱,把“故事状态”从隐式上下文记忆转成显式可维护结构,因此显著提升了长篇叙事的一致性、连贯性与可控性。
+
+## 论文信息
+
+- 标题:`STORYTELLER: An Enhanced Plot-Planning Framework for Coherent and Cohesive Story Generation`
+- 会议:`Findings of the Association for Computational Linguistics: ACL 2025`
+- 作者:Jiaming Li, Yukun Chen, Ziqiang Liu 等
+- 论文链接:<https://aclanthology.org/2025.findings-acl.1071/>
+- PDF:<https://aclanthology.org/2025.findings-acl.1071.pdf>
+- arXiv:<https://arxiv.org/abs/2506.02347>
+
+## 论文要解决的问题
+
+论文开篇指出,现有自动故事生成方法虽然能写出流畅文本,但在长篇叙事里经常出现以下问题:
+
+- 风格前后不一致
+- 情节逻辑断裂
+- 角色动机突然变化
+- 章节之间衔接松散
+- 重复、冗长、创造性不足
+
+论文认为,问题的根源在于:
+
+- 既有方法往往只提供高层 outline
+- 后续章节或事件生成彼此相对独立
+- 模型对先前剧情、角色关系和事件因果缺乏持续追踪
+
+因此,单纯“先列大纲再扩写”还不够,还需要一个在生成过程中持续参与的动态结构层。
+
+## 核心思想
+
+作者借鉴人类写作者的认知循环,将故事写作抽象为三类持续交互的行为:
+
+- Retrieval:回看已有剧情与实体关系
+- Evaluation:判断新情节是否合理
+- Generation:在当前状态约束下生成新内容
+
+围绕这个思路,`STORYTELLER` 做了三件关键事:
+
+1. 用 `SVO` 三元组把剧情抽象成情节节点
+2. 用 `STORYLINE` 维护按时间推进的事件链
+3. 用 `NEKG` 维护人物、地点、物品与事件关系图
+
+这意味着它不是让模型直接从 prompt 一路写到底,而是先维护一个不断更新的“剧情状态空间”,再在该状态空间内生成文本。
+
+## 核心模块
+
+### 1. NODES:基于 SVO 的情节节点
+
+论文把故事中的关键事件表示为 `Subject-Verb-Object` 三元组,也就是:
+
+- 谁
+- 做了什么
+- 作用到谁 / 什么
+
+这个设计的意义是:
+
+- 把复杂叙事压缩成可比较、可检索、可审查的事件单元
+- 让剧情推进具有明确的结构锚点
+- 降低长文本生成时的漂移风险
+
+论文进一步把章节内节点分成三类:
+
+- `CBN`:Chapter Begin Node,章节起始节点
+- `CPN`:Chapter Plot Node,章节推进节点
+- `CEN`:Chapter End Node,章节结束节点
+
+可以把它理解成:每一章都被拆成“进入状态、推进事件、收束结果”三个层次。
+
+### 2. STORYLINE:时间序列化剧情主线
+
+`STORYLINE` 用来保存生成过程中出现的所有 `NODE`。
+
+它的关键特征是:
+
+- 每个节点都带有 `time_stamp`
+- 所有节点按时间顺序组织
+- 能形成故事全局事件时间线
+
+论文强调,`STORYLINE` 的作用是:
+
+- 保证事件顺序正确
+- 追踪剧情如何逐步演化
+- 为后续节点生成提供最近事件上下文
+
+从工程角度看,`STORYLINE` 相当于一个面向剧情推进的时序状态表。
+
+### 3. NEKG:叙事实体知识图谱
+
+`NEKG` 全称是 `Narrative Entity Knowledge Graph`。
+
+论文中,它负责用图结构维护叙事中的实体与关系:
+
+- 人物
+- 地点
+- 物品
+- 概念
+- 它们之间的互动与关联
+
+论文明确提到,`NEKG` 使用 `Neo4j` 实现,并且每个生成故事都有自己的图实例。
+
+`NEKG` 的主要作用有三点:
+
+- 记录角色互动和对象变化
+- 提供与当前候选剧情有关的关联节点
+- 支持后续剧情扩展与基于关系的推理
+
+如果说 `STORYLINE` 解决的是“发生了什么,先后顺序如何”,那么 `NEKG` 解决的是“谁和谁有关、关系怎么影响后续剧情”。
+
+## 三阶段生成流程
+
+论文把整个系统分成三个阶段。
+
+## Stage 1:高层故事生成
+
+这一阶段先把用户输入压缩成高层叙事框架。
+
+具体包含两步:
+
+### 1. Premise 和 Synopsis 生成
+
+系统先从用户 prompt 中抽取核心信息,生成:
+
+- `Premise`:故事设定、时代背景、环境与社会语境
+- `Synopsis`:主线剧情、人物关系、主要转折
+
+这一步的作用是先把“灵感提示词”变成“具备世界观和主线的故事蓝图”。
+
+### 2. Chapter Titles 和 Abstracts 生成
+
+系统进一步把长篇故事拆成章节,并为每章生成:
+
+- 章节标题
+- 章节摘要
+
+这些摘要不仅用于组织结构,也作为后续中层情节节点生成的输入约束。
+
+简化理解:
+
+- Stage 1 决定整本故事大致写什么
+- 让后续生成不至于一开始就陷入全文自由扩写
+
+## Stage 2:中层情节结构生成
+
+这是整篇论文最关键的部分。
+
+在这一阶段,系统并不直接写正文,而是先为每一章生成节点结构。
+
+### 1. 先生成 CBN 和 CEN
+
+对于每一章,系统会先生成:
+
+- 本章从什么状态开始
+- 本章在什么状态收束
+
+论文指出,这一步不仅参考当前章节,还会考虑前后相邻章节,从而保证章节切分和过渡更自然。
+
+### 2. 再逐个生成 CPN
+
+有了章节起点和终点之后,系统开始填充中间的推进节点 `CPN`。
+
+这里不是直接一次性生成全部节点,而是循环执行:
+
+1. 先根据已有 `CBN / CEN / 已生成 CPN` 生成一个 `Pseudo CPN`
+2. 以该候选节点中的 `S'` 和 `O'` 为线索,从 `NEKG` 检索相关节点
+3. 同时从 `STORYLINE` 提取最近事件
+4. 让 LLM 审查这个 `Pseudo CPN` 是否合理
+5. 合理则接受,不合理则改写成新的 `CPN`
+6. 重复直到当前章节节点链能够自然走向 `CEN`
+
+这个设计很重要,因为它体现了论文真正的创新点:
+
+- 不是先规划完再写,而是边规划边审查
+- 不是只看局部章节摘要,而是持续引用已有剧情状态
+- 不是把图谱当静态附属物,而是把它作为节点审查依据
+
+### 为什么要先生成 Pseudo CPN 再审查
+
+论文给出的原因很现实:
+
+- `STORYLINE` 和 `NEKG` 中数据会不断增长
+- 长上下文和成本限制不允许每次把全部历史都喂给 LLM
+
+所以它先生成候选节点,再围绕该节点做有目标的检索与校验,从而把上下文使用控制在一个较低成本范围内。
+
+从工程角度看,这一步本质上是:
+
+`候选事件生成 -> 相关证据召回 -> 一致性评审 -> 事件确认`
+
+## Stage 3:细粒度正文写作
+
+在完成章节节点结构之后,系统才开始生成自然语言文本块。
+
+正文生成会综合使用:
+
+- 当前章节标题与摘要
+- 当前章节的 `CBN / CPNs / CEN`
+- 上一个文本块
+
+各个文本块生成后被加入列表,最后拼接成完整故事。
+
+这说明 `STORYTELLER` 的正文生成并不是自由创作模式,而是:
+
+- 被章节结构约束
+- 被节点链约束
+- 被前文文本块衔接约束
+
+因此它更像一个“有状态的扩写器”。
+
+## 方法优势
+
+结合论文设计,我认为它的主要优势有以下几点:
+
+### 1. 把长期一致性显式化
+
+传统方法把一致性寄托在模型上下文记忆里。`STORYTELLER` 把一致性拆成两个显式结构:
+
+- 时间上的一致性:交给 `STORYLINE`
+- 实体关系上的一致性:交给 `NEKG`
+
+这样就把原本隐式、不可控的叙事状态转成了可检查、可更新、可检索的结构。
+
+### 2. 把章节写作变成状态转移
+
+每章不再只是一个文本块,而是:
+
+- 从 `CBN` 进入
+- 经由若干 `CPN` 推进
+- 到 `CEN` 收束
+
+这让长篇故事天然更适合做章节级控制。
+
+### 3. 把“规划”变成动态过程
+
+不少方案只有静态 outline。`STORYTELLER` 则在每次生成中层节点时都重新参考已有状态,因此规划不是一次性产物,而是生成流程中的动态参与者。
+
+### 4. 有利于后续扩展推理
+
+由于 `NEKG` 是图结构,它理论上可以支持:
+
+- 人物关系演化追踪
+- 关键物品因果链分析
+- 支线剧情生长
+- 多角色视角整合
+
+这比纯向量检索更适合小说系统。
+
+## 局限与风险
+
+论文结果不错,但从工程落地看,这套方案也有明显成本。
+
+### 1. 流程较重
+
+它不是一次生成,而是多阶段流水线:
+
+- 高层规划
+- 中层节点规划
+- 候选节点审查
+- 正文块生成
+
+这会带来更高的:
+
+- token 成本
+- 延迟
+- 失败重试复杂度
+
+### 2. 对结构化抽象质量依赖高
+
+如果节点设计不稳定,就会出现:
+
+- 情节抽象过粗,约束力不够
+- 情节抽象过细,生成链路过长
+- 节点语义不准确,导致后续检索噪声
+
+### 3. 实体归一化难度高
+
+小说里同一个角色可能有:
+
+- 姓名
+- 称谓
+- 代词
+- 外号
+
+如果 `NEKG` 没有稳定的实体归一化与别名机制,图谱会迅速碎裂。
+
+### 4. 图谱检索可能引入噪声
+
+当实体越来越多、关系越来越密时,如何只召回“对当前候选节点真正相关”的信息,会成为系统效果上限的关键因素。
+
+## 实验结果
+
+论文报告显示,`STORYTELLER` 在人工偏好评测中平均胜率达到 `84.33%`。
+
+对比对象包括:
+
+- `GPT-4o`
+- `Qwen2-72B-Instruct`
+- `Meta-Llama-3.1-70B-Instruct`
+- `LongWriter-glm4-9b`
+- `LongWriter-llama3.1-8b`
+- `DOC v2`
+
+论文还给出人工偏好评测的代表结果:
+
+- 对 `GPT-4o` 胜率 `91%`
+- 对 `Qwen2-72B-Instruct` 胜率 `86%`
+- 对 `Meta-Llama-3.1-70B-Instruct` 胜率 `85%`
+- 对 `LongWriter-glm4-9b` 胜率 `83%`
+- 对 `LongWriter-llama3.1-8b` 胜率 `79%`
+- 对 `DOC v2` 胜率 `82%`
+
+在细分指标上,论文也报告其在以下维度领先:
+
+- Creativity
+- Coherence
+- Engagement
+- Relevance
+- Overall Performance
+
+其中 Overall 分数为 `89.4`,明显高于论文中的其他对比方法。
+
+## 对 `webnovel-writer` 的启发
+
+这篇论文对当前项目最有价值的,并不是“必须上 Neo4j”,而是它的系统拆法。
+
+### 最值得吸收的三个能力
+
+#### 1. 章节级剧情状态机
+
+当前项目如果想增强长篇稳定性,可以考虑引入类似:
+
+- 章节起点
+- 若干关键推进点
+- 章节终点
+
+不用一开始就完全复制论文里的 `CBN / CPN / CEN` 细节,但这个抽象非常适合作为章节规划层。
+
+#### 2. 角色与事件的显式状态维护
+
+论文说明,仅靠 prompt 历史无法稳定维持长篇人物一致性。
+
+对我们项目而言,更现实的做法是先实现一个轻量版 `NEKG`:
+
+- 人物卡
+- 关系边
+- 关键道具状态
+- 势力关系
+- 最近事件链
+
+第一阶段未必需要图数据库,`JSON + 索引 + 检索` 就能先跑起来。
+
+#### 3. 候选剧情审查器
+
+`Pseudo CPN Review` 的思路非常值得借鉴。
+
+可落地成一个“候选剧情检查器”,专门检查:
+
+- 是否和角色设定冲突
+- 是否和上一章结尾冲突
+- 是否违背世界规则
+- 是否重复已有桥段
+- 是否能顺利过渡到本章目标
+
+这比单纯让模型“继续写下一章”更稳定。
+
+## 建议的落地顺序
+
+如果将论文思想引入当前项目,我建议按以下顺序做简化实现:
+
+1. 先加章节级结构化节点
+2. 再加角色与关系状态存储
+3. 再加候选事件审查与修正
+4. 最后再考虑是否需要真正的图数据库
+
+原因很简单:
+
+- 论文的思想比它的具体技术栈更重要
+- 先把“结构化剧情状态”做起来,收益最大
+- 过早引入 Neo4j 可能增加工程负担
+
+## 我的判断
+
+从研究价值看,`STORYTELLER` 很像是“长篇故事生成的状态化框架”。
+
+它真正证明的是:
+
+- 大纲不够,需要中层情节节点
+- 检索不够,需要时序剧情状态和实体关系状态
+- 生成不够,需要生成中的审查修正闭环
+
+对小说系统来说,这个方向是对的,而且比纯提示词优化更有长期价值。
+
+## 结论
+
+`STORYTELLER` 不是一个靠单次 prompt 取胜的技巧,而是一套完整的叙事编排机制。
+
+它把长篇故事生成重构为:
+
+- 高层故事规划
+- 中层情节节点生成
+- 基于时间线与实体图谱的动态校验
+- 受结构约束的正文扩写
+
+对于 `webnovel-writer`,这篇论文最值得借鉴的不是照搬论文系统,而是吸收它背后的三个原则:
+
+- 故事状态要显式维护
+- 章节推进要结构化建模
+- 正文生成前要先过一致性审查
+
+如果后续要继续落地,我建议下一步直接把这篇论文转成一份“项目改造设计文档”,将 `STORYLINE / NEKG / CPN 审查器` 映射到当前仓库的模块边界。
+
+## 参考链接
+
+- ACL Anthology:<https://aclanthology.org/2025.findings-acl.1071/>
+- PDF:<https://aclanthology.org/2025.findings-acl.1071.pdf>
+- arXiv:<https://arxiv.org/abs/2506.02347>

+ 43 - 1
webnovel-writer/agents/consistency-checker.md

@@ -39,7 +39,7 @@ model: inherit
 3. `设定集/`(世界观圣经)
 4. `大纲/`(对照上下文)
 
-### 第二步: 层一致性检查
+### 第二步: 层一致性检查
 
 #### 第一层: 战力一致性(战力检查)
 
@@ -129,6 +129,38 @@ model: inherit
    → VIOLATION: Time regression without flashback marker
 ```
 
+#### 第四层: 情节结构一致性(有节点时)
+
+仅当章纲包含 `CBN/CEN/本章禁区` 字段时执行,否则跳过。
+
+**校验项**:
+- 正文是否违反本章禁区中的任何条目
+- 正文开头方向是否与 `CBN` 目标冲突
+- 正文结尾状态是否与 `CEN` 目标冲突
+- 关键节点所依赖的能力、身份、地点是否违背现有设定
+
+**危险信号** (PLOT_STRUCTURE_VIOLATION):
+```
+❌ 本章禁区违反:"萧炎不得突破到斗师",但正文描写突破场景
+   → Prohibition: 不得突破 | Actual: 正文第N段描写突破
+   → severity: high
+
+❌ CEN 方向冲突:CEN 为"药老|决定|进一步接触萧炎",但正文结尾药老直接拒绝
+   → CEN: 决定接触 | Actual ending: 拒绝并离开
+   → severity: medium
+
+❌ CBN 方向冲突:CBN 为"萧炎|抵达|迦南学院入口",正文开头却在酒楼闲聊
+   → CBN: 抵达学院 | Actual opening: 酒楼场景
+   → severity: medium
+```
+
+**严重度建议**:
+| 问题类型 | Severity |
+|---------|----------|
+| 禁区违反 | **high** |
+| CEN 方向冲突 | **medium** |
+| CBN 方向冲突 | **medium** |
+
 ### 第三步: 实体一致性检查
 
 **对所有章节中检测到的新实体**:
@@ -175,6 +207,13 @@ model: inherit
 **结论**: 发现 {Z} 处违规
 **严重时间线问题**: {count} 个(必须修复后才能继续)
 
+## 情节结构一致性(有节点时)
+| 检查项 | 结果 | 严重度 | 详情 |
+|--------|------|--------|------|
+| 本章禁区 | ✓/✗ | high | {...} |
+| CBN 方向 | ✓/✗ | medium | {...} |
+| CEN 方向 | ✓/✗ | medium | {...} |
+
 ## 新实体一致性检查
 - ✓ 与世界观一致的新实体: {count}
 - ⚠️ 不一致的实体: {count}(详见下方列表)
@@ -188,6 +227,7 @@ model: inherit
 - [战力冲突] 润色时修改第{M}章,将"破空斩"替换为筑基期可用技能
 - [地点错误] 润色时补充移动过程描述或调整地点设定
 - [时间线问题] 润色时统一时间线推算,修正矛盾
+- [结构冲突] 润色时修正禁区违反或调整章节开头/结尾方向
 - [实体冲突] 润色时确认是否为新设定或需要调整
 
 ## 综合评分
@@ -218,6 +258,7 @@ python -X utf8 "${CLAUDE_PLUGIN_ROOT:?CLAUDE_PLUGIN_ROOT is required}/scripts/we
 ❌ 接受无世界观解释的瞬移
 ❌ **降低 TIMELINE_ISSUE 严重度**(时间问题不得降级)
 ❌ **通过存在严重/高优先级时间线问题的章节**(必须修复)
+❌ **通过存在本章禁区违反的章节**
 
 ## 成功标准
 
@@ -225,4 +266,5 @@ python -X utf8 "${CLAUDE_PLUGIN_ROOT:?CLAUDE_PLUGIN_ROOT is required}/scripts/we
 - 0 个高优先级时间线问题(**倒计时错误、时间回跳、重大事件无时间推进**)
 - 所有新实体与现有世界观一致
 - 地点和时间线过渡合乎逻辑
+- 有节点时,`CBN/CEN/本章禁区` 一致性检查已完成
 - 报告为润色步骤提供具体修复建议

+ 23 - 6
webnovel-writer/agents/context-agent.md

@@ -8,7 +8,7 @@ model: inherit
 # context-agent(上下文搜集 Agent)
 
 > **职责**:生成可直接开写的创作执行包,目标是“信息够用、约束清楚、无需补问”。
-> **原则**:按需召回、推断补全、先接住上章、再锁定本章任务与章末钩子。
+> **原则**:按需召回、推断补全、先接住上章、再锁定本章任务与章末钩子;若章纲提供结构化节点,则将其转化为本章写作节拍
 
 ## 核心参考
 
@@ -33,7 +33,7 @@ model: inherit
 
 输出必须是一个单一执行包,包含以下 3 层内容,且三层信息必须一致。
 
-### 1. 任务书(8 个板块)
+### 1. 任务书(8+1 个板块)
 
 - 本章核心任务:目标、阻力、代价、核心冲突一句话、必须完成、绝对不能、反派层级
 - 接住上章:上章钩子、读者期待、开头建议
@@ -43,6 +43,7 @@ model: inherit
 - 风格指导:本章类型、参考样本、最近模式、本章建议
 - 连续性与伏笔:时间/位置/情绪连贯;必须处理与可选伏笔
 - 追读力策略:未闭合问题、钩子类型/强度、微兑现建议、差异化提示
+- 情节结构(有节点时):CBN、CPNs 序列、CEN、必须覆盖节点、本章禁区
 
 ### 2. Context Contract(内置合同)
 
@@ -50,17 +51,21 @@ model: inherit
 - 开头类型、情绪节奏、信息密度
 - 是否过渡章
 - 追读力设计:钩子类型/强度、微兑现清单、爽点模式
+- `plot_structure`(有节点时):`{cbn, cpns[], cen, mandatory_nodes[], prohibitions[]}`
 
 ### 3. Step 2A 直写提示词
 
-- 章节节拍:开场触发 → 推进/受阻 → 反转/兑现 → 章末钩子
+- 章节节拍:
+  - 有节点时:`CBN触发 -> CPN推进 -> CPN受阻/变化 -> ... -> CEN收束 -> 章末钩子`
+  - 无节点时:`开场触发 -> 推进/受阻 -> 反转/兑现 -> 章末钩子`
 - 不可变事实清单:大纲事实、设定事实、承接事实、长期记忆事实
-- 禁止事项:越级能力、无因果跳转、设定冲突、剧情硬拐
+- 禁止事项:越级能力、无因果跳转、设定冲突、剧情硬拐、违反本章禁区中的任何条目(有节点时)
 - 终检清单:本章必须满足项与 fail 条件
 
 硬规则:
 - 若信息冲突,优先级为 `设定 > 大纲 > 长期记忆 > 风格偏好`。
 - 输出内容必须能直接交给 Step 2A 开写,不得依赖额外补问。
+- 节点结构来自章纲,`context-agent` 不得擅自生成或改写节点。
 
 ## 读取优先级与默认值
 
@@ -80,6 +85,10 @@ model: inherit
   - 第 7 板块仍必须输出;
   - 显式标注“结构化伏笔数据缺失,需人工补录”;
   - 禁止静默跳过伏笔板块。
+- 若章纲无结构化节点字段:
+  - 跳过“情节结构”板块;
+  - 使用旧版节拍生成逻辑;
+  - 不阻断流程。
 
 章节编号统一使用 4 位数,如 `0001`、`0099`、`0100`。
 
@@ -184,6 +193,11 @@ cat "{project_root}/.webnovel/memory_scratchpad.json"
 必须提取并写入任务书:
 - 目标、阻力、代价、反派层级、本章变化、章末未闭合问题、钩子
 
+若章纲包含 `CBN/CPNs/CEN/必须覆盖节点/本章禁区` 字段:
+- 额外提取并组装为“情节结构”板块
+- 将节点映射为 `plot_structure`
+- 缺失时跳过,不阻断
+
 ### Step 6:读取追读力、债务与模式数据
 
 ```bash
@@ -241,10 +255,11 @@ python "${SCRIPTS_DIR}/webnovel.py" --project-root "{project_root}" index recent
 - Step 2A 直写提示词
 
 硬要求:
-- 任务书必须包含 8 个板块,且含“时间约束”。
+- 任务书必须包含 8+1 个板块,且含“时间约束”。
 - “不可变事实清单”必须纳入长期记忆事实。
 - `open_loops` 和 `reader_promises` 若与本章有关,必须进入“连续性与伏笔”或“终检清单”。
 - 若 `必须处理` 伏笔超过 3 条:前 3 条标记“最高优先”,其余标记“本章仍需处理”。
+- 有节点时,必须把 `plot_structure` 纳入合同与节拍映射。
 
 ### Step 10:执行红线校验并输出
 
@@ -257,6 +272,7 @@ python "${SCRIPTS_DIR}/webnovel.py" --project-root "{project_root}" index recent
 - 红线 5:合同与任务书冲突
 - 红线 6:时间逻辑错误
 - 红线 7:长期记忆事实被遗漏或写反
+- 红线 8:情节结构节点与任务书/合同存在方向冲突(有节点时)
 
 通过标准:
 - fail 数 = 0
@@ -266,7 +282,7 @@ python "${SCRIPTS_DIR}/webnovel.py" --project-root "{project_root}" index recent
 ## 成功标准
 
 1. 创作执行包可直接驱动 Step 2A。
-2. 任务书包含 8 个板块,且时间约束完整。
+2. 任务书包含 8+1 个板块,且时间约束完整。
 3. 上章钩子与读者期待明确。
 4. 角色动机与情绪为非空推断结果。
 5. 最近模式已对比,并给出差异化建议。
@@ -275,3 +291,4 @@ python "${SCRIPTS_DIR}/webnovel.py" --project-root "{project_root}" index recent
 8. 长期记忆事实已被读取,并进入不可变事实或终检清单。
 9. 时间逻辑红线通过。
 10. Step 2A 可在不补问的情况下直接开写。
+11. 有节点时,情节结构板块已完整纳入执行包,节拍映射清晰。

+ 45 - 0
webnovel-writer/agents/continuity-checker.md

@@ -153,6 +153,39 @@ vs.
 - **中等**(情节调整): 需标记并确认
 - **重大**(核心冲突变化): 必须标记 `<deviation reason="..."/>` 并说明
 
+### 第三步-B: 情节节点覆盖检查(有节点时)
+
+仅当章纲包含 `CBN/CPNs/CEN/必须覆盖节点` 字段时执行,否则跳过。
+
+**检查项**:
+1. **CBN 承接**:正文开头 1/3 是否体现了 CBN 描述的起始状态或事件方向
+2. **CEN 落地**:正文结尾 1/3 是否体现了 CEN 描述的结束状态或事件方向
+3. **必须节点覆盖**:`必须覆盖节点` 中列出的 CPN 在正文中是否有对应情节
+4. **可选节点覆盖**:非必须 CPN 的覆盖情况(仅统计,不作为 fail 主依据)
+
+判定方式:
+- 语义方向匹配,不要求逐字对应
+- 允许叙述扩展、合并表达和场景重排
+- 不允许跳过 CBN/CEN 或完全缺失必须节点
+
+**覆盖评级**:
+- **A**: CBN+CEN 方向吻合,所有必须节点已覆盖
+- **B**: CBN+CEN 基本吻合,1 个必须节点仅弱覆盖
+- **C**: CBN 或 CEN 偏离,或 2 个以上必须节点未覆盖
+- **F**: CBN+CEN 均偏离,或核心必须节点完全缺失
+
+**报告格式**:
+```markdown
+## 情节节点覆盖
+| 节点 | 类型 | 要求 | 正文对应 | 状态 |
+|------|------|------|---------|------|
+| 萧炎 \| 抵达 \| 迦南学院入口 | CBN | 必须 | 第1段:抵达描写 | ✓ 覆盖 |
+| 萧炎 \| 展示 \| 异火控制力 | CPN | 必须 | 第4-5段:考核表现 | ✓ 覆盖 |
+| 药老 \| 对萧炎产生 \| 明确兴趣 | CEN | 必须 | 末段:态度转变 | ✓ 覆盖 |
+
+**节点覆盖评级**: A
+```
+
 ### 第四步: 拖沓检查(拖沓检查)
 
 **识别拖沓段落**:
@@ -202,6 +235,15 @@ vs.
 
 **伏笔健康度**: {X} 已回收, {Y} 待处理, {Z} 有风险
 
+## 情节节点覆盖(有节点时)
+| 节点 | 类型 | 要求 | 正文对应 | 状态 |
+|------|------|------|---------|------|
+| {CBN} | CBN | 必须 | {...} | ✓/✗ |
+| {CPN1} | CPN | 必须 | {...} | ✓/△/✗ |
+| {CEN} | CEN | 必须 | {...} | ✓/✗ |
+
+**节点覆盖评级**: {A/B/C/F}
+
 ## 逻辑一致性
 | 章节 | 问题 | 类型 | 严重度 |
 |------|------|------|--------|
@@ -226,6 +268,7 @@ vs.
 3. **解决逻辑矛盾**: 第{M}章修改"从未见过"为"很少见到"
 4. **提及休眠线索**: 血煞门入侵线索建议第47章再次提及
 5. **压缩拖沓段落**: 第{N}-{M}章赶路场景合并为1章
+6. **修复节点覆盖问题**: 补写缺失的必须节点或调整结尾落点
 
 ## 综合评分
 **连贯性总评**: {流畅/可接受/生硬/断裂}
@@ -239,6 +282,7 @@ vs.
 ❌ 忽略遗忘伏笔(10+ 章休眠)
 ❌ 接受突兀的场景转换(F 级)
 ❌ 忽视情节漏洞和前后矛盾
+❌ 有节点覆盖 `F` 级评定却未在修复建议中提及
 
 ## 成功标准
 
@@ -248,3 +292,4 @@ vs.
 - 0 个重大逻辑漏洞
 - 大纲偏差已正确标记
 - 报告指出需修复的具体章节
+- 有节点时,节点覆盖评级 ≥ B

+ 1 - 0
webnovel-writer/agents/data-agent.md

@@ -105,6 +105,7 @@ python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "{project_root}" inde
 python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "{project_root}" index get-by-alias --alias "萧炎"
 ```
 
+
 ### Step 3:执行实体提取与语义消歧
 
 由 Data Agent 在同一轮上下文内直接完成,不额外调用独立 LLM Agent。

+ 93 - 0
webnovel-writer/scripts/chapter_outline_loader.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 import json
 import re
 from pathlib import Path
+from typing import Any, Dict
 
 try:
     from chapter_paths import volume_num_for_chapter
@@ -127,3 +128,95 @@ def load_chapter_outline(project_root: Path, chapter_num: int, max_chars: int |
     if max_chars and len(outline) > max_chars:
         return outline[:max_chars] + "\n...(已截断)"
     return outline
+
+_PLOT_SECTION_FIELD_MAP = {
+    "cbn": "cbn",
+    "cpns": "cpns",
+    "cen": "cen",
+    "必须覆盖节点": "mandatory_nodes",
+    "本章禁区": "prohibitions",
+}
+
+
+def _clean_plot_line(line: str) -> str:
+    text = str(line or "").strip()
+    text = re.sub(r"^[\-\*•]+\s*", "", text)
+    text = re.sub(r"^\d+[\.、]\s*", "", text)
+    return text.strip()
+
+
+def _append_plot_value(target: Dict[str, Any], field: str, value: str) -> None:
+    value = _clean_plot_line(value)
+    if not value:
+        return
+
+    if field in {"cpns", "mandatory_nodes", "prohibitions"}:
+        target.setdefault(field, [])
+        candidates = [value]
+        if field in {"mandatory_nodes", "prohibitions"}:
+            split_values = [part.strip() for part in re.split(r"[、,,;;|]+", value) if part.strip()]
+            if split_values:
+                candidates = split_values
+        for item in candidates:
+            if item not in target[field]:
+                target[field].append(item)
+        return
+
+    if field not in target:
+        target[field] = value
+
+
+def parse_chapter_plot_structure(outline_text: str) -> Dict[str, Any]:
+    text = str(outline_text or "")
+    if not text or text.startswith("⚠️"):
+        return {}
+
+    structure: Dict[str, Any] = {}
+    current_field = ""
+
+    for raw_line in text.splitlines():
+        stripped = raw_line.strip()
+        if not stripped:
+            current_field = ""
+            continue
+        if re.match(r"^#{1,6}\s*第\s*\d+\s*章", stripped):
+            current_field = ""
+            continue
+
+        cleaned = _clean_plot_line(stripped)
+        matched_field = ""
+        matched_value = ""
+        for label, field in _PLOT_SECTION_FIELD_MAP.items():
+            match = re.match(rf"^{re.escape(label)}\s*[::]\s*(.*)$", cleaned, re.IGNORECASE)
+            if match:
+                matched_field = field
+                matched_value = match.group(1).strip()
+                break
+
+        if matched_field:
+            current_field = matched_field
+            _append_plot_value(structure, matched_field, matched_value)
+            continue
+
+        if current_field:
+            _append_plot_value(structure, current_field, cleaned)
+
+    cpns = structure.get("cpns") or []
+    mandatory_nodes = structure.get("mandatory_nodes") or []
+    prohibitions = structure.get("prohibitions") or []
+    if not any([structure.get("cbn"), cpns, structure.get("cen"), mandatory_nodes, prohibitions]):
+        return {}
+
+    return {
+        "cbn": str(structure.get("cbn") or "").strip(),
+        "cpns": cpns,
+        "cen": str(structure.get("cen") or "").strip(),
+        "mandatory_nodes": mandatory_nodes,
+        "prohibitions": prohibitions,
+        "source": "chapter_outline",
+    }
+
+
+def load_chapter_plot_structure(project_root: Path, chapter_num: int) -> Dict[str, Any]:
+    outline = load_chapter_outline(project_root, chapter_num, max_chars=None)
+    return parse_chapter_plot_structure(outline)

+ 9 - 2
webnovel-writer/scripts/data_modules/context_manager.py

@@ -15,9 +15,9 @@ from runtime_compat import enable_windows_utf8_stdio
 from typing import Any, Dict, List, Optional
 
 try:
-    from chapter_outline_loader import load_chapter_outline
+    from chapter_outline_loader import load_chapter_outline, load_chapter_plot_structure
 except ImportError:  # pragma: no cover
-    from scripts.chapter_outline_loader import load_chapter_outline
+    from scripts.chapter_outline_loader import load_chapter_outline, load_chapter_plot_structure
 
 from .config import get_config
 from .index_manager import IndexManager, WritingChecklistScoreMeta
@@ -60,6 +60,7 @@ class ContextManager:
         "reader_signal",
         "genre_profile",
         "writing_guidance",
+        "plot_structure",
     }
     SECTION_ORDER = [
         "core",
@@ -68,6 +69,7 @@ class ContextManager:
         "reader_signal",
         "genre_profile",
         "writing_guidance",
+        "plot_structure",
         "story_skeleton",
         "memory",
         "long_term_memory",
@@ -254,6 +256,7 @@ class ContextManager:
         reader_signal = self._load_reader_signal(chapter)
         genre_profile = self._load_genre_profile(state)
         writing_guidance = self._build_writing_guidance(chapter, reader_signal, genre_profile)
+        plot_structure = self._load_plot_structure(chapter)
 
         return {
             "meta": {"chapter": chapter},
@@ -263,6 +266,7 @@ class ContextManager:
             "reader_signal": reader_signal,
             "genre_profile": genre_profile,
             "writing_guidance": writing_guidance,
+            "plot_structure": plot_structure,
             "story_skeleton": story_skeleton,
             "preferences": preferences,
             "memory": memory,
@@ -670,6 +674,9 @@ class ContextManager:
     def _load_outline(self, chapter: int) -> str:
         return load_chapter_outline(self.config.project_root, chapter, max_chars=1500)
 
+    def _load_plot_structure(self, chapter: int) -> Dict[str, Any]:
+        return load_chapter_plot_structure(self.config.project_root, chapter)
+
     def _load_recent_summaries(self, chapter: int, window: int = 3) -> List[Dict[str, Any]]:
         summaries = []
         for ch in range(max(1, chapter - window), chapter):

+ 193 - 1
webnovel-writer/scripts/data_modules/index_manager.py

@@ -39,7 +39,7 @@ from pathlib import Path
 
 from runtime_compat import enable_windows_utf8_stdio
 from typing import Dict, List, Optional, Any, Tuple
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, asdict
 from contextlib import contextmanager
 from datetime import datetime
 
@@ -205,6 +205,45 @@ class ReviewMetrics:
     notes: str = ""
 
 
+@dataclass
+class ReviewAggregateResult:
+    """Step 3 审查聚合结果"""
+
+    chapter: int
+    start_chapter: int
+    end_chapter: int
+    selected_checkers: List[str] = field(default_factory=list)
+    checkers: Dict[str, Dict[str, Any]] = field(default_factory=dict)
+    issues: List[Dict[str, Any]] = field(default_factory=list)
+    overall_score: float = 0.0
+    severity_counts: Dict[str, int] = field(default_factory=dict)
+    critical_issues: List[str] = field(default_factory=list)
+    dimension_scores: Dict[str, float] = field(default_factory=dict)
+    overall: Dict[str, Any] = field(default_factory=dict)
+    notes: str = ""
+    timeline_gate: Dict[str, Any] = field(default_factory=dict)
+
+    def to_review_metrics(self, report_file: str = "") -> ReviewMetrics:
+        notes = str(self.notes or "")
+        timeline_gate = self.timeline_gate or {}
+        if timeline_gate:
+            gate_note = (
+                f"timeline_gate:block={bool(timeline_gate.get('blocked', False))};"
+                f"count={int(timeline_gate.get('blocking_issue_count', 0) or 0)}"
+            )
+            notes = f"{notes} | {gate_note}" if notes else gate_note
+        return ReviewMetrics(
+            start_chapter=self.start_chapter,
+            end_chapter=self.end_chapter,
+            overall_score=self.overall_score,
+            dimension_scores=self.dimension_scores,
+            severity_counts=self.severity_counts,
+            critical_issues=self.critical_issues,
+            report_file=report_file,
+            notes=notes,
+        )
+
+
 @dataclass
 class WritingChecklistScoreMeta:
     """写作清单评分记录(Context Contract v2 Phase F)"""
@@ -634,6 +673,139 @@ class IndexManager(IndexChapterMixin, IndexEntityMixin, IndexDebtMixin, IndexRea
 # ==================== CLI 接口 ====================
 
 
+
+
+def _normalize_checker_issue(issue: object) -> dict:
+    if not isinstance(issue, dict):
+        return {}
+    return {
+        "id": str(issue.get("id") or ""),
+        "type": str(issue.get("type") or ""),
+        "severity": str(issue.get("severity") or "medium"),
+        "location": str(issue.get("location") or ""),
+        "description": str(issue.get("description") or ""),
+        "suggestion": str(issue.get("suggestion") or ""),
+        "can_override": bool(issue.get("can_override", False)),
+    }
+
+
+def _build_timeline_gate(issues: List[Dict[str, Any]]) -> Dict[str, Any]:
+    blocking = []
+    for issue in issues:
+        if not isinstance(issue, dict):
+            continue
+        issue_type = str(issue.get("type") or "").strip()
+        severity = str(issue.get("severity") or "").strip().lower()
+        if issue_type == "TIMELINE_ISSUE" and severity in {"critical", "high"}:
+            blocking.append(issue)
+    return {
+        "blocked": bool(blocking),
+        "blocking_issue_count": len(blocking),
+        "blocking_issues": blocking,
+    }
+
+
+def _aggregate_checker_results(chapter: int, raw_data: object) -> dict:
+    if isinstance(raw_data, dict) and isinstance(raw_data.get("checkers"), dict):
+        checker_map = dict(raw_data.get("checkers") or {})
+    elif isinstance(raw_data, dict):
+        checker_map = dict(raw_data)
+    elif isinstance(raw_data, list):
+        checker_map = {}
+        for item in raw_data:
+            if isinstance(item, dict):
+                agent = str(item.get("agent") or "").strip()
+                if agent:
+                    checker_map[agent] = item
+    else:
+        checker_map = {}
+
+    selected_checkers: list[str] = []
+    issues: list[dict] = []
+    severity_counts = {"critical": 0, "high": 0, "medium": 0, "low": 0}
+    critical_issues: list[str] = []
+    dimension_scores: dict[str, float] = {}
+    score_values: list[float] = []
+    aggregated_checkers: dict[str, dict] = {}
+
+    dimension_alias = {
+        "consistency-checker": "consistency",
+        "continuity-checker": "continuity",
+        "ooc-checker": "ooc",
+        "reader-pull-checker": "reader_pull",
+        "high-point-checker": "high_point",
+        "pacing-checker": "pacing",
+    }
+
+    for agent, payload in checker_map.items():
+        if not isinstance(payload, dict):
+            continue
+        selected_checkers.append(agent)
+        score = payload.get("overall_score")
+        try:
+            numeric_score = float(score)
+            score_values.append(numeric_score)
+            dimension_scores[dimension_alias.get(agent, agent)] = round(numeric_score, 2)
+        except (TypeError, ValueError):
+            numeric_score = None
+
+        raw_issues = payload.get("issues") or []
+        normalized_issues = []
+        critical_count = 0
+        high_count = 0
+        for raw_issue in raw_issues:
+            issue = _normalize_checker_issue(raw_issue)
+            if not issue:
+                continue
+            normalized_issues.append(issue)
+            issues.append({**issue, "agent": agent, "chapter": chapter})
+            severity = issue.get("severity") or "medium"
+            if severity not in severity_counts:
+                severity = "medium"
+            severity_counts[severity] += 1
+            if severity == "critical":
+                critical_count += 1
+                if issue.get("description"):
+                    critical_issues.append(issue["description"])
+            elif severity == "high":
+                high_count += 1
+
+        aggregated_checkers[agent] = {
+            "score": numeric_score,
+            "pass": bool(payload.get("pass", False)),
+            "critical": critical_count,
+            "high": high_count,
+        }
+
+    overall_score = round(sum(score_values) / len(score_values), 2) if score_values else 0.0
+    timeline_gate = _build_timeline_gate(issues)
+    overall = {
+        "score": overall_score,
+        "pass": severity_counts["critical"] == 0 and not timeline_gate.get("blocked", False),
+        "critical_total": severity_counts["critical"],
+        "high_total": severity_counts["high"],
+        "can_proceed": severity_counts["critical"] == 0 and not timeline_gate.get("blocked", False),
+    }
+
+    notes = "selected_checkers=" + ",".join(selected_checkers)
+    result = ReviewAggregateResult(
+        chapter=chapter,
+        start_chapter=chapter,
+        end_chapter=chapter,
+        selected_checkers=selected_checkers,
+        checkers=aggregated_checkers,
+        issues=issues,
+        overall_score=overall_score,
+        severity_counts=severity_counts,
+        critical_issues=critical_issues,
+        dimension_scores=dimension_scores,
+        overall=overall,
+        notes=notes,
+        timeline_gate=timeline_gate,
+    )
+    return asdict(result)
+
+
 def main():
     import argparse
     import sys
@@ -788,6 +960,15 @@ def main():
     review_trend_parser = subparsers.add_parser("get-review-trend-stats")
     review_trend_parser.add_argument("--last-n", type=int, default=5)
 
+    review_aggregate_parser = subparsers.add_parser("aggregate-review-results")
+    review_aggregate_parser.add_argument("--chapter", type=int, required=True)
+    review_aggregate_parser.add_argument("--data", required=True, help="JSON 格式的 checker 原始结果列表或映射")
+
+    review_materialize_parser = subparsers.add_parser("materialize-review-metrics")
+    review_materialize_parser.add_argument("--chapter", type=int, required=True)
+    review_materialize_parser.add_argument("--data", required=True, help="JSON 格式的 checker 原始结果列表或映射")
+    review_materialize_parser.add_argument("--report-file", default="", help="审查报告路径")
+
     checklist_score_save_parser = subparsers.add_parser("save-writing-checklist-score")
     checklist_score_save_parser.add_argument("--data", required=True, help="JSON 格式的写作清单评分数据")
 
@@ -1134,6 +1315,17 @@ def main():
         rows = manager.list_invalid_facts(args.status)
         emit_success(rows, message="invalid_list")
 
+    elif args.command == "aggregate-review-results":
+        data = load_json_arg(args.data)
+        aggregated = _aggregate_checker_results(args.chapter, data)
+        emit_success(aggregated, message="review_results_aggregated", chapter=args.chapter)
+
+    elif args.command == "materialize-review-metrics":
+        data = load_json_arg(args.data)
+        aggregated = ReviewAggregateResult(**_aggregate_checker_results(args.chapter, data))
+        metrics = aggregated.to_review_metrics(report_file=args.report_file)
+        emit_success(asdict(metrics), message="review_metrics_materialized", chapter=args.chapter)
+
     elif args.command == "save-review-metrics":
         data = load_json_arg(args.data)
         metrics = ReviewMetrics(

+ 4 - 0
webnovel-writer/scripts/data_modules/schemas.py

@@ -109,6 +109,8 @@ class MemoryFacts(BaseModel):
     reader_promises: List[ReaderPromise] = Field(default_factory=list)
 
 
+
+
 class DataAgentOutput(BaseModel):
     model_config = ConfigDict(extra="allow")
 
@@ -181,4 +183,6 @@ def normalize_data_agent_output(payload: Dict[str, Any]) -> Dict[str, Any]:
                 memory_facts[key] = [value]
 
     payload.setdefault("scenes_chunked", 0)
+
+
     return payload

+ 2 - 1
webnovel-writer/scripts/data_modules/state_manager.py

@@ -32,6 +32,7 @@ from .observability import safe_append_perf_timing, safe_log_tool_call
 
 logger = logging.getLogger(__name__)
 
+
 try:
     # 当 scripts 目录在 sys.path 中(常见:从 scripts/ 运行)
     from security_utils import atomic_write_json, read_json_safe
@@ -1079,6 +1080,7 @@ class StateManager:
         # 处理消歧不确定项(不影响实体写入,但必须对 Writer 可见)
         warnings.extend(self._record_disambiguation(chapter, result.get("uncertain", [])))
 
+
         # 写入 chapter_meta(钩子/模式/结束状态)
         chapter_meta = result.get("chapter_meta")
         if isinstance(chapter_meta, dict):
@@ -1086,7 +1088,6 @@ class StateManager:
             self._state.setdefault("chapter_meta", {})
             self._state["chapter_meta"][meta_key] = chapter_meta
             self._pending_chapter_meta[meta_key] = chapter_meta
-
         # 更新进度
         self.update_progress(chapter)
 

+ 39 - 0
webnovel-writer/scripts/data_modules/state_validator.py

@@ -51,6 +51,28 @@ _PATTERN_FIELDS = [
 _PATTERN_SPLIT_RE = re.compile(r"[、,,/|+;;。]+")
 
 
+_PLOT_LIST_FIELDS = ("cpns", "mandatory_nodes", "prohibitions")
+
+
+def _normalize_string_list(raw_value: Any) -> List[str]:
+    items: List[str] = []
+    if isinstance(raw_value, list):
+        source = raw_value
+    elif isinstance(raw_value, str):
+        source = split_patterns(raw_value)
+    else:
+        return []
+
+    seen = set()
+    for item in source:
+        text = str(item).strip()
+        if not text or text in seen:
+            continue
+        seen.add(text)
+        items.append(text)
+    return items
+
+
 def to_positive_int(value: Any) -> Optional[int]:
     if value is None or isinstance(value, bool):
         return None
@@ -203,6 +225,23 @@ def normalize_chapter_meta_entry(entry: Mapping[str, Any]) -> Dict[str, Any]:
     if merged_patterns:
         normalized["coolpoint_patterns"] = merged_patterns
 
+    plot_structure = entry.get("plot_structure")
+    if isinstance(plot_structure, Mapping):
+        normalized_plot_structure = dict(plot_structure)
+        cbn = str(plot_structure.get("cbn") or "").strip()
+        cen = str(plot_structure.get("cen") or "").strip()
+        if cbn:
+            normalized_plot_structure["cbn"] = cbn
+        if cen:
+            normalized_plot_structure["cen"] = cen
+        for field_name in _PLOT_LIST_FIELDS:
+            normalized_values = _normalize_string_list(plot_structure.get(field_name))
+            if normalized_values:
+                normalized_plot_structure[field_name] = normalized_values
+            elif field_name in normalized_plot_structure:
+                normalized_plot_structure[field_name] = []
+        normalized["plot_structure"] = normalized_plot_structure
+
     return normalized
 
 

+ 34 - 0
webnovel-writer/scripts/data_modules/tests/test_context_manager.py

@@ -699,3 +699,37 @@ def test_context_manager_genre_profile_prefers_project_over_project_info(temp_pr
 
     assert profile.get("genre_raw") == "xuanhuan"
     assert profile.get("genre") == "xuanhuan"
+
+
+def test_context_manager_includes_plot_structure_when_outline_has_nodes(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",
+    )
+
+    manager = ContextManager(temp_project)
+    payload = manager.build_context(4, use_snapshot=False, save_snapshot=False)
+
+    plot_structure = payload["sections"]["plot_structure"]["content"]
+    assert plot_structure.get("cbn") == "进入试炼场"
+    assert plot_structure.get("cpns") == ["观察规则", "发现陷阱"]
+    assert plot_structure.get("cen") == "决定将计就计"
+    assert plot_structure.get("mandatory_nodes") == ["发现陷阱"]
+    assert plot_structure.get("prohibitions") == ["不能直接翻脸"]

+ 182 - 0
webnovel-writer/scripts/data_modules/tests/test_data_modules.py

@@ -42,9 +42,19 @@ def temp_project():
     with tempfile.TemporaryDirectory() as tmpdir:
         config = DataModulesConfig.from_project_root(tmpdir)
         config.ensure_dirs()
+        _ensure_project_state(config)
         yield config
 
 
+
+
+def _ensure_project_state(config):
+    config.ensure_dirs()
+    state_path = config.webnovel_dir / "state.json"
+    if not state_path.exists():
+        state_path.write_text(json.dumps({}, ensure_ascii=False), encoding="utf-8")
+
+
 class TestEntityLinker:
     """实体链接器测试"""
 
@@ -1412,3 +1422,175 @@ class TestRAGAdapter:
 
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])
+
+
+
+def test_aggregate_checker_results_cli(temp_project, monkeypatch, capsys):
+    def run_cli(args):
+        monkeypatch.setattr(sys, "argv", args)
+        import data_modules.index_manager as im
+        im.main()
+        out = capsys.readouterr().out
+        return json.loads(out)
+
+    payload = {
+        "consistency-checker": {
+            "agent": "consistency-checker",
+            "chapter": 12,
+            "overall_score": 90,
+            "pass": True,
+            "issues": [],
+            "metrics": {"violated_prohibitions": []},
+            "summary": "设定稳定",
+        },
+        "continuity-checker": {
+            "agent": "continuity-checker",
+            "chapter": 12,
+            "overall_score": 70,
+            "pass": True,
+            "issues": [
+                {
+                    "id": "PLOT_001",
+                    "type": "PLOT_NODE_MISSING",
+                    "severity": "high",
+                    "location": "第8段",
+                    "description": "缺少关键节点“决定隐忍”",
+                    "suggestion": "补一段主角压下冲动的决策过程",
+                }
+            ],
+            "metrics": {
+                "hit_nodes": ["发现异常"],
+                "missing_nodes": ["决定隐忍"],
+                "coverage_grade": "C",
+            },
+            "summary": "节点覆盖不足",
+        },
+    }
+
+    out = run_cli([
+        "index_manager",
+        "--project-root",
+        str(temp_project.project_root),
+        "aggregate-review-results",
+        "--chapter",
+        "12",
+        "--data",
+        json.dumps(payload, ensure_ascii=False),
+    ])
+
+    assert out["status"] == "success"
+    data = out["data"]
+    assert data["overall_score"] == 80.0
+    assert data["severity_counts"]["high"] == 1
+    assert data["dimension_scores"]["consistency"] == 90.0
+    assert data["dimension_scores"]["continuity"] == 70.0
+    assert data["selected_checkers"] == ["consistency-checker", "continuity-checker"]
+    assert data["issues"][0]["agent"] == "continuity-checker"
+
+
+
+def test_materialize_review_metrics_cli(temp_project, monkeypatch, capsys):
+    def run_cli(args):
+        monkeypatch.setattr(sys, "argv", args)
+        import data_modules.index_manager as im
+        im.main()
+        out = capsys.readouterr().out
+        return json.loads(out)
+
+    payload = {
+        "continuity-checker": {
+            "agent": "continuity-checker",
+            "chapter": 18,
+            "overall_score": 72,
+            "pass": True,
+            "issues": [
+                {
+                    "id": "TIME_001",
+                    "type": "TIMELINE_ISSUE",
+                    "severity": "high",
+                    "location": "第3段",
+                    "description": "时间线回跳",
+                    "suggestion": "补明确闪回标记",
+                }
+            ],
+            "metrics": {},
+            "summary": "存在时间线风险",
+        },
+        "consistency-checker": {
+            "agent": "consistency-checker",
+            "chapter": 18,
+            "overall_score": 88,
+            "pass": True,
+            "issues": [],
+            "metrics": {},
+            "summary": "设定稳定",
+        },
+    }
+
+    out = run_cli([
+        "index_manager",
+        "--project-root",
+        str(temp_project.project_root),
+        "materialize-review-metrics",
+        "--chapter",
+        "18",
+        "--report-file",
+        "审查报告/第18章.md",
+        "--data",
+        json.dumps(payload, ensure_ascii=False),
+    ])
+
+    assert out["status"] == "success"
+    data = out["data"]
+    assert data["start_chapter"] == 18
+    assert data["end_chapter"] == 18
+    assert data["overall_score"] == 80.0
+    assert data["severity_counts"]["high"] == 1
+    assert data["report_file"] == "审查报告/第18章.md"
+    assert data["notes"] == "selected_checkers=continuity-checker,consistency-checker | timeline_gate:block=True;count=1"
+
+
+def test_aggregate_checker_results_blocks_overall_pass_for_high_timeline_issue(temp_project, monkeypatch, capsys):
+    def run_cli(args):
+        monkeypatch.setattr(sys, "argv", args)
+        import data_modules.index_manager as im
+        im.main()
+        out = capsys.readouterr().out
+        return json.loads(out)
+
+    payload = {
+        "continuity-checker": {
+            "agent": "continuity-checker",
+            "chapter": 21,
+            "overall_score": 78,
+            "pass": True,
+            "issues": [
+                {
+                    "id": "TIME_002",
+                    "type": "TIMELINE_ISSUE",
+                    "severity": "high",
+                    "location": "第6段",
+                    "description": "事件时间回跳",
+                    "suggestion": "补时间锚点",
+                }
+            ],
+            "metrics": {},
+            "summary": "时间线阻断",
+        }
+    }
+
+    out = run_cli([
+        "index_manager",
+        "--project-root",
+        str(temp_project.project_root),
+        "aggregate-review-results",
+        "--chapter",
+        "21",
+        "--data",
+        json.dumps(payload, ensure_ascii=False),
+    ])
+
+    data = out["data"]
+    assert data["timeline_gate"]["blocked"] is True
+    assert data["overall"]["pass"] is False
+    assert data["overall"]["can_proceed"] is False

+ 77 - 0
webnovel-writer/scripts/data_modules/tests/test_extract_chapter_context.py

@@ -272,3 +272,80 @@ def test_render_text_contains_rag_assist_section_when_hits_exist(tmp_path):
     assert "- 模式: auto" in text
     assert "[graph_hybrid]" in text
     assert "萧炎与药老" in text
+
+
+def test_build_chapter_context_payload_includes_plot_structure(tmp_path):
+    scripts_dir = Path(__file__).resolve().parents[2]
+    if str(scripts_dir) not in sys.path:
+        sys.path.insert(0, str(scripts_dir))
+
+    from extract_chapter_context import build_chapter_context_payload
+
+    webnovel_dir = tmp_path / ".webnovel"
+    webnovel_dir.mkdir(parents=True, exist_ok=True)
+    state = {
+        "project": {"genre": "xuanhuan"},
+        "progress": {"current_chapter": 5, "total_words": 15000},
+        "protagonist_state": {},
+        "chapter_meta": {},
+        "disambiguation_warnings": [],
+        "disambiguation_pending": [],
+    }
+    (webnovel_dir / "state.json").write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
+
+    outline_dir = tmp_path / "大纲"
+    outline_dir.mkdir(parents=True, exist_ok=True)
+    (outline_dir / "第1卷-详细大纲.md").write_text(
+        """### 第5章:试炼开局
+CBN:萧炎抵达外院试炼场
+CPNs:
+- 导师宣布试炼规则
+- 萧炎发现规则被人做了手脚
+CEN:萧炎决定先隐忍观察
+必须覆盖节点:规则异常暴露、决定隐忍
+本章禁区:不能直接揭穿黑手
+""",
+        encoding="utf-8",
+    )
+
+    payload = build_chapter_context_payload(tmp_path, 5)
+    plot_structure = payload.get("plot_structure") or {}
+    assert plot_structure.get("cbn") == "萧炎抵达外院试炼场"
+    assert plot_structure.get("cpns") == ["导师宣布试炼规则", "萧炎发现规则被人做了手脚"]
+    assert plot_structure.get("cen") == "萧炎决定先隐忍观察"
+    assert plot_structure.get("mandatory_nodes") == ["规则异常暴露", "决定隐忍"]
+    assert plot_structure.get("prohibitions") == ["不能直接揭穿黑手"]
+
+
+def test_render_text_contains_plot_structure_section(tmp_path):
+    scripts_dir = Path(__file__).resolve().parents[2]
+    if str(scripts_dir) not in sys.path:
+        sys.path.insert(0, str(scripts_dir))
+
+    from extract_chapter_context import _render_text
+
+    payload = {
+        "chapter": 8,
+        "outline": "测试大纲",
+        "previous_summaries": [],
+        "state_summary": "状态",
+        "context_contract_version": "v2",
+        "plot_structure": {
+            "cbn": "主角进入遗迹",
+            "cpns": ["发现石碑异常", "与守卫短暂交锋"],
+            "cen": "决定深入遗迹核心",
+            "mandatory_nodes": ["发现石碑异常"],
+            "prohibitions": ["不能提前拿到终极传承"],
+        },
+        "reader_signal": {},
+        "genre_profile": {},
+        "writing_guidance": {},
+        "rag_assist": {"invoked": False, "hits": []},
+    }
+
+    text = _render_text(payload)
+    assert "## 情节结构" in text
+    assert "- CBN: 主角进入遗迹" in text
+    assert "- CPN1: 发现石碑异常" in text
+    assert "- CEN: 决定深入遗迹核心" in text
+    assert "- 本章禁区: 不能提前拿到终极传承" in text

+ 0 - 1
webnovel-writer/scripts/data_modules/tests/test_state_validator.py

@@ -104,4 +104,3 @@ def test_normalize_state_runtime_sections():
     chapter_meta = normalize_chapter_meta(normalized["chapter_meta"])
     assert "1" in chapter_meta
     assert chapter_meta["1"]["coolpoint_patterns"] == ["打脸", "翻车"]
-

+ 223 - 0
webnovel-writer/scripts/data_modules/tests/test_webnovel_unified_cli.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
+import json
 import sys
 from pathlib import Path
 
@@ -166,3 +167,225 @@ def test_quality_trend_report_writes_to_book_root_when_input_is_workspace_root(t
     assert output_path.is_file()
     assert (book_root / ".webnovel" / "index.db").is_file()
     assert not (workspace_root / ".webnovel" / "index.db").exists()
+
+
+
+def test_index_aggregate_review_results_forwards_with_resolved_project_root(monkeypatch, tmp_path):
+    module = _load_webnovel_module()
+
+    book_root = (tmp_path / "book").resolve()
+    called = {}
+
+    def _fake_resolve(explicit_project_root=None):
+        return book_root
+
+    def _fake_run_data_module(module_name, argv):
+        called["module_name"] = module_name
+        called["argv"] = list(argv)
+        return 0
+
+    monkeypatch.setattr(module, "_resolve_root", _fake_resolve)
+    monkeypatch.setattr(module, "_run_data_module", _fake_run_data_module)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "webnovel",
+            "--project-root",
+            str(tmp_path),
+            "index",
+            "aggregate-review-results",
+            "--chapter",
+            "12",
+            "--data",
+            '{"continuity-checker":{"overall_score":80}}',
+        ],
+    )
+
+    with pytest.raises(SystemExit) as exc:
+        module.main()
+
+    assert int(exc.value.code or 0) == 0
+    assert called["module_name"] == "index_manager"
+    assert called["argv"] == [
+        "--project-root",
+        str(book_root),
+        "aggregate-review-results",
+        "--chapter",
+        "12",
+        "--data",
+        '{"continuity-checker":{"overall_score":80}}',
+    ]
+
+
+
+
+
+def test_review_pipeline_builds_artifacts(tmp_path):
+    _ensure_scripts_on_path()
+    import review_pipeline as review_pipeline_module
+
+    project_root = (tmp_path / "book").resolve()
+    (project_root / ".webnovel").mkdir(parents=True, exist_ok=True)
+    (project_root / ".webnovel" / "state.json").write_text("{}", encoding="utf-8")
+
+    review_results_path = tmp_path / "review_results.json"
+    review_results_path.write_text(
+        json.dumps(
+            {
+                "continuity-checker": {
+                    "agent": "continuity-checker",
+                    "chapter": 20,
+                    "overall_score": 76,
+                    "pass": True,
+                    "issues": [
+                        {
+                            "id": "TIME_001",
+                            "type": "TIMELINE_ISSUE",
+                            "severity": "high",
+                            "location": "第2段",
+                            "description": "时间线回跳",
+                            "suggestion": "补时间锚点",
+                        }
+                    ],
+                    "metrics": {},
+                    "summary": "存在时间线问题",
+                },
+                "consistency-checker": {
+                    "agent": "consistency-checker",
+                    "chapter": 20,
+                    "overall_score": 84,
+                    "pass": True,
+                    "issues": [],
+                    "metrics": {},
+                    "summary": "设定稳定",
+                },
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+
+    payload = review_pipeline_module.build_review_artifacts(
+        project_root=project_root,
+        chapter=20,
+        review_results_path=review_results_path,
+        report_file="审查报告/第20章.md",
+    )
+
+    assert payload["aggregated"]["overall_score"] == 80.0
+    assert payload["aggregated"]["timeline_gate"]["blocked"] is True
+    assert payload["review_metrics"]["report_file"] == "审查报告/第20章.md"
+    assert "timeline_gate:block=True;count=1" in payload["review_metrics"]["notes"]
+
+
+def test_review_pipeline_forwards_with_resolved_project_root(monkeypatch, tmp_path):
+    module = _load_webnovel_module()
+
+    book_root = (tmp_path / "book").resolve()
+    review_results = (tmp_path / "review_results.json").resolve()
+    called = {}
+
+    def _fake_resolve(explicit_project_root=None):
+        return book_root
+
+    def _fake_run_script(script_name, argv):
+        called["script_name"] = script_name
+        called["argv"] = list(argv)
+        return 0
+
+    monkeypatch.setattr(module, "_resolve_root", _fake_resolve)
+    monkeypatch.setattr(module, "_run_script", _fake_run_script)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "webnovel",
+            "--project-root",
+            str(tmp_path),
+            "review-pipeline",
+            "--chapter",
+            "18",
+            "--review-results",
+            str(review_results),
+            "--aggregated-out",
+            str(tmp_path / "aggregated.json"),
+            "--review-metrics-out",
+            str(tmp_path / "review_metrics.json"),
+            "--report-file",
+            "审查报告/第18章.md",
+        ],
+    )
+
+    with pytest.raises(SystemExit) as exc:
+        module.main()
+
+    assert int(exc.value.code or 0) == 0
+    assert called["script_name"] == "review_pipeline.py"
+    assert called["argv"] == [
+        "--project-root",
+        str(book_root),
+        "--chapter",
+        "18",
+        "--review-results",
+        str(review_results),
+        "--aggregated-out",
+        str(tmp_path / "aggregated.json"),
+        "--review-metrics-out",
+        str(tmp_path / "review_metrics.json"),
+        "--report-file",
+        "审查报告/第18章.md",
+    ]
+
+
+def test_review_pipeline_main_creates_output_directories(tmp_path):
+    _ensure_scripts_on_path()
+    import review_pipeline as review_pipeline_module
+
+    project_root = (tmp_path / "book").resolve()
+    (project_root / ".webnovel").mkdir(parents=True, exist_ok=True)
+    (project_root / ".webnovel" / "state.json").write_text("{}", encoding="utf-8")
+
+    review_results_path = tmp_path / "review_results.json"
+    review_results_path.write_text(
+        json.dumps(
+            {
+                "consistency-checker": {
+                    "agent": "consistency-checker",
+                    "chapter": 9,
+                    "overall_score": 88,
+                    "pass": True,
+                    "issues": [],
+                    "metrics": {},
+                    "summary": "稳定",
+                }
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+
+    aggregated_out = project_root / ".webnovel" / "tmp" / "review" / "aggregated.json"
+    metrics_out = project_root / ".webnovel" / "tmp" / "review" / "metrics.json"
+
+    old_argv = sys.argv
+    sys.argv = [
+        "review_pipeline",
+        "--project-root",
+        str(project_root),
+        "--chapter",
+        "9",
+        "--review-results",
+        str(review_results_path),
+        "--aggregated-out",
+        str(aggregated_out),
+        "--review-metrics-out",
+        str(metrics_out),
+    ]
+    try:
+        review_pipeline_module.main()
+    finally:
+        sys.argv = old_argv
+
+    assert aggregated_out.is_file()
+    assert metrics_out.is_file()

+ 21 - 0
webnovel-writer/scripts/data_modules/tests/test_workflow_manager.py

@@ -202,3 +202,24 @@ def test_cleanup_artifacts_confirm_deletes_with_backup(tmp_path, monkeypatch):
     backup_dir = tmp_path / ".webnovel" / "recovery_backups"
     backups = list(backup_dir.glob("ch0008-*"))
     assert backups
+
+
+def test_workflow_step_zero_point_five_is_registered(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 0.5") == "webnovel-write-skill"
+    assert module.get_pending_steps("webnovel-write")[0] == "Step 0.5"
+
+    module.start_task("webnovel-write", {"chapter_num": 15})
+    module.start_step("Step 0.5", "节点预检")
+    module.complete_step("Step 0.5")
+    module.start_step("Step 1", "Context")
+
+    trace_path = module.get_call_trace_path()
+    rows = [json.loads(line) for line in trace_path.read_text(encoding="utf-8").splitlines() if line.strip()]
+    violation_events = [row for row in rows if row.get("event") == "step_order_violation"]
+    assert not violation_events

+ 20 - 0
webnovel-writer/scripts/data_modules/webnovel.py

@@ -252,6 +252,13 @@ def main() -> None:
     p_extract_context.add_argument("--chapter", type=int, required=True, help="目标章节号")
     p_extract_context.add_argument("--format", choices=["text", "json"], default="text", help="输出格式")
 
+    p_review_pipeline = sub.add_parser("review-pipeline", help="转发到 review_pipeline.py")
+    p_review_pipeline.add_argument("--chapter", type=int, required=True, help="目标章节号")
+    p_review_pipeline.add_argument("--review-results", required=True, help="checker 原始结果 JSON 文件")
+    p_review_pipeline.add_argument("--aggregated-out", help="聚合结果输出文件")
+    p_review_pipeline.add_argument("--review-metrics-out", help="review_metrics 输出文件")
+    p_review_pipeline.add_argument("--report-file", default="", help="审查报告路径")
+
     # 兼容:允许 `--project-root` 出现在任意位置(减少 agents/skills 拼命令的出错率)
     from .cli_args import normalize_global_project_root
 
@@ -308,6 +315,19 @@ def main() -> None:
     if tool == "extract-context":
         return_args = [*forward_args, "--chapter", str(args.chapter), "--format", str(args.format)]
         raise SystemExit(_run_script("extract_chapter_context.py", return_args))
+    if tool == "review-pipeline":
+        return_args = [
+            *forward_args,
+            "--chapter", str(args.chapter),
+            "--review-results", str(args.review_results),
+        ]
+        if args.aggregated_out:
+            return_args.extend(["--aggregated-out", str(args.aggregated_out)])
+        if args.review_metrics_out:
+            return_args.extend(["--review-metrics-out", str(args.review_metrics_out)])
+        if args.report_file:
+            return_args.extend(["--report-file", str(args.report_file)])
+        raise SystemExit(_run_script("review_pipeline.py", return_args))
 
     raise SystemExit(2)
 

+ 24 - 1
webnovel-writer/scripts/extract_chapter_context.py

@@ -20,7 +20,7 @@ import sys
 from pathlib import Path
 from typing import Any, Dict, List
 
-from chapter_outline_loader import load_chapter_outline
+from chapter_outline_loader import load_chapter_outline, load_chapter_plot_structure
 
 from runtime_compat import enable_windows_utf8_stdio
 
@@ -314,6 +314,7 @@ def _load_contract_context(project_root: Path, chapter_num: int) -> Dict[str, An
         "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", {}),
     }
 
@@ -329,6 +330,7 @@ def build_chapter_context_payload(project_root: Path, chapter_num: int) -> Dict[
 
     state_summary = extract_state_summary(project_root)
     contract_context = _load_contract_context(project_root, chapter_num)
+    plot_structure = contract_context.get("plot_structure") or load_chapter_plot_structure(project_root, chapter_num)
     rag_assist = _load_rag_assist(project_root, chapter_num, outline)
 
     return {
@@ -341,6 +343,7 @@ def build_chapter_context_payload(project_root: Path, chapter_num: int) -> Dict[
         "reader_signal": contract_context.get("reader_signal", {}),
         "genre_profile": contract_context.get("genre_profile", {}),
         "writing_guidance": contract_context.get("writing_guidance", {}),
+        "plot_structure": plot_structure,
         "long_term_memory": contract_context.get("long_term_memory", {}),
         "rag_assist": rag_assist,
     }
@@ -382,6 +385,26 @@ def _render_text(payload: Dict[str, Any]) -> str:
             lines.append(f"- 上下文阶段权重: {stage}")
             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 []

+ 86 - 0
webnovel-writer/scripts/review_pipeline.py

@@ -0,0 +1,86 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+review_pipeline.py - Step 3 审查结果文件流辅助脚本
+
+职责:
+- 读取 checker 原始结果文件
+- 调用 index_manager 聚合审查结果
+- 物化为 review_metrics 结构
+- 可选写入 review_metrics 到 index.db
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import Any, Dict
+
+from runtime_compat import enable_windows_utf8_stdio
+
+
+def _ensure_scripts_path() -> None:
+    scripts_dir = Path(__file__).resolve().parent
+    if str(scripts_dir) not in sys.path:
+        sys.path.insert(0, str(scripts_dir))
+
+
+def _load_json(path: Path) -> Dict[str, Any]:
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def _write_json(path: Path, payload: Dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+
+def build_review_artifacts(project_root: Path, chapter: int, review_results_path: Path, report_file: str = "") -> Dict[str, Any]:
+    _ensure_scripts_path()
+    from data_modules.config import DataModulesConfig
+    from data_modules.index_manager import _aggregate_checker_results, ReviewAggregateResult
+
+    _ = DataModulesConfig.from_project_root(project_root)
+    raw_data = _load_json(review_results_path)
+    aggregated = ReviewAggregateResult(**_aggregate_checker_results(chapter, raw_data))
+    review_metrics = aggregated.to_review_metrics(report_file=report_file)
+    return {
+        "chapter": chapter,
+        "review_results": raw_data,
+        "aggregated": aggregated.__dict__,
+        "review_metrics": review_metrics.__dict__,
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="构建 Step 3 审查中间产物")
+    parser.add_argument("--project-root", required=True, help="项目根目录")
+    parser.add_argument("--chapter", type=int, required=True, help="章节号")
+    parser.add_argument("--review-results", required=True, help="checker 原始结果 JSON 文件")
+    parser.add_argument("--aggregated-out", help="聚合结果输出文件")
+    parser.add_argument("--review-metrics-out", help="review_metrics 输出文件")
+    parser.add_argument("--report-file", default="", help="审查报告路径")
+    args = parser.parse_args()
+
+    project_root = Path(args.project_root).resolve()
+    review_results_path = Path(args.review_results).resolve()
+    payload = build_review_artifacts(
+        project_root=project_root,
+        chapter=args.chapter,
+        review_results_path=review_results_path,
+        report_file=args.report_file,
+    )
+
+    if args.aggregated_out:
+        _write_json(Path(args.aggregated_out), payload["aggregated"])
+
+    if args.review_metrics_out:
+        _write_json(Path(args.review_metrics_out), payload["review_metrics"])
+
+    print(json.dumps(payload, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    if sys.platform == "win32":
+        enable_windows_utf8_stdio()
+    main()

+ 13 - 2
webnovel-writer/scripts/workflow_manager.py

@@ -42,6 +42,16 @@ STEP_STATUS_RUNNING = "running"
 STEP_STATUS_COMPLETED = "completed"
 STEP_STATUS_FAILED = "failed"
 
+WRITE_WORKFLOW_STEPS = [
+    "Step 0.5",
+    "Step 1",
+    "Step 2A",
+    "Step 2B",
+    "Step 3",
+    "Step 4",
+    "Step 5",
+    "Step 6",
+]
 
 def now_iso() -> str:
     return datetime.now().isoformat()
@@ -110,6 +120,7 @@ def expected_step_owner(command: str, step_id: str) -> str:
     """
     if command == "webnovel-write":
         mapping = {
+            "Step 0.5": "webnovel-write-skill",
             "Step 1": "context-agent",
             "Step 1.5": "webnovel-write-skill",
             "Step 2A": "writer-draft",
@@ -423,7 +434,7 @@ def analyze_recovery_options(interrupt_info):
 
     step_id = current_step["id"]
 
-    if step_id in {"Step 1", "Step 1.5"}:
+    if step_id in {"Step 0.5", "Step 1", "Step 1.5"}:
         return [
             {
                 "option": "A",
@@ -715,7 +726,7 @@ def get_pending_steps(command):
     """Get command pending step list."""
     if command == "webnovel-write":
         # v2: Step 1 内置 Contract v2,不再单独记录 Step 1.5,避免产生 step_order_violation 噪声。
-        return ["Step 1", "Step 2A", "Step 2B", "Step 3", "Step 4", "Step 5", "Step 6"]
+        return list(WRITE_WORKFLOW_STEPS)
     if command == "webnovel-review":
         return ["Step 1", "Step 2", "Step 3", "Step 4", "Step 5", "Step 6", "Step 7", "Step 8"]
     return []

+ 50 - 4
webnovel-writer/skills/webnovel-plan/SKILL.md

@@ -10,6 +10,7 @@ description: 基于总纲生成卷纲、时间线和章纲,并把新增设定
 - 基于总纲细化卷纲、时间线与章纲,不重做全局故事。
 - 先补齐设定基线,再产出可直接进入写作的章纲。
 - 卷纲完成后,把新增设定增量写回现有设定集。
+- 将详细大纲升级为“结构化详细大纲”,为下游写作提供中层情节结构。
 
 ## 执行原则
 
@@ -17,6 +18,7 @@ description: 基于总纲生成卷纲、时间线和章纲,并把新增设定
 2. 先锁定卷级节奏,再批量拆章。
 3. 时间线是硬约束,所有章纲都必须带时间字段。
 4. 若发现总纲与设定冲突,先阻断,再等用户裁决。
+5. 结构化节点服务于写作执行,不追求语法学上的严格 SVO 抽取。
 
 ## 环境准备
 
@@ -139,10 +141,10 @@ cat "$PROJECT_ROOT/.webnovel/idea_bank.json"
 ### Step 7:批量生成章纲
 
 批次规则:
-- `<=20` 章:1 批
-- `21-40` 章:2 批
-- `41-60` 章:3 批
-- `>60` 章:4 批及以上
+- 默认按 `10章/批`
+- 复杂题材或多线并进时可降到 `8章/批`
+- 简单升级流可放宽到 `12章/批`
+- 不建议单批超过 `12章`
 
 按需加载:
 
@@ -167,6 +169,48 @@ cat "${SKILL_ROOT}/references/outlining/chapter-planning.md"
 - 本章变化
 - 章末未闭合问题
 - 钩子
+- `章节起点(CBN)`
+- `推进节点(CPNs)`
+- `章节终点(CEN)`
+- `必须覆盖节点`
+- `本章禁区`
+
+#### 结构化节点规范
+
+节点格式统一为:
+
+`主体 | 动作/变化 | 对象/结果`
+
+说明:
+- 这里的节点是写作执行骨架,不追求严格语法学 SVO。
+- `动作/变化` 可以表示行动、判断、意识变化或状态转移。
+- `对象/结果` 可以是人、物、地点,也可以是结果状态。
+
+示例:
+- `萧炎 | 抵达 | 迦南学院入口`
+- `萧炎 | 展示 | 异火控制力`
+- `药老 | 对萧炎产生 | 明确兴趣`
+- `萧炎 | 意识到 | 学院考核远比预想更严苛`
+
+结构规则:
+- 每章固定 `1 个 CBN`
+- 每章 `2-4 个 CPN`
+- 每章固定 `1 个 CEN`
+- 相邻章节 `CEN -> 下一章 CBN` 必须逻辑承接(首章和末章除外)
+- `CPNs` 必须按时间顺序排列
+
+必须覆盖规则:
+- 每章必须覆盖节点最多 `4` 个
+- 建议为:`CBN + CEN + 1~2 个核心 CPN`
+- 可选节点只作为写作建议,不得作为 fail 主依据
+
+本章禁区规则:
+- 不超过 `5` 条
+- 只写本章绝对不能发生的硬禁区
+- 不写风格类建议,不写空泛表述
+
+向后兼容:
+- 若旧项目章纲缺失 `CBN/CPNs/CEN/必须覆盖节点/本章禁区` 字段,下游流程正常执行,仅跳过结构化检查
 
 输出文件:`大纲/第{volume_id}卷-详细大纲.md`
 
@@ -198,6 +242,8 @@ cat "${SKILL_ROOT}/references/outlining/chapter-planning.md"
 - 倒计时推进正确
 - 新设定已回写到现有设定集
 - `BLOCKER=0`
+- 有节点时,相邻章节 `CEN -> CBN` 无明显逻辑冲突
+- 有节点时,每章必须覆盖节点不超过 `4` 个
 
 更新状态:
 

+ 59 - 13
webnovel-writer/skills/webnovel-write/SKILL.md

@@ -12,6 +12,7 @@ allowed-tools: Read Write Edit Grep Bash Task
 - 默认目标字数 2000-2500;若用户或大纲另有要求,以用户和大纲为准。
 - 保证审查、润色、数据回写、长期记忆提取全部闭环。
 - 输出内容必须能被下一章直接消费。
+- 若章纲包含结构化节点,则正文必须围绕节点骨架展开。
 
 ## 执行原则
 
@@ -23,9 +24,9 @@ allowed-tools: Read Write Edit Grep Bash Task
 
 ## 模式定义
 
-- `/webnovel-write`:Step 1 → Step 2A → Step 2B → Step 3 → Step 4 → Step 5 → Step 6
-- `/webnovel-write --fast`:Step 1 → Step 2A → Step 3 → Step 4 → Step 5 → Step 6
-- `/webnovel-write --minimal`:Step 1 → Step 2A → Step 2B → Step 3(仅核心 3 个审查器)→ Step 4 → Step 5 → Step 6
+- `/webnovel-write`:Step 0.5 → Step 1 → Step 2A → Step 2B → Step 3 → Step 4 → Step 5 → Step 6
+- `/webnovel-write --fast`:Step 0.5 → Step 1 → Step 2A → Step 3 → Step 4 → Step 5 → Step 6
+- `/webnovel-write --minimal`:Step 0.5 → Step 1 → Step 2A → Step 2B → Step 3(仅核心 3 个审查器)→ Step 4 → Step 5 → Step 6
 
 最小产物:
 - 章节正文文件
@@ -41,7 +42,7 @@ allowed-tools: Read Write Edit Grep Bash Task
 - 禁止改名:标准产物文件名和格式不得私自改写。
 - 禁止伪造审查:Step 3 必须由 Task 子代理执行。
 - 禁止源码探测:CLI 调用方式以本文档和 agent 文档为准,命令失败优先查日志。
-- Workflow step-id 必须使用实现侧真实编号:`Step 1`、`Step 2A`、`Step 2B`、`Step 3`、`Step 4`、`Step 5`、`Step 6`。
+- Workflow step-id 必须使用实现侧真实编号:`Step 0.5`、`Step 1`、`Step 2A`、`Step 2B`、`Step 3`、`Step 4`、`Step 5`、`Step 6`。
 
 ## 引用加载等级
 
@@ -113,6 +114,35 @@ export PROJECT_ROOT="$(python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-roo
 python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "${PROJECT_ROOT}" workflow start-task --command webnovel-write --chapter {chapter_num} || true
 ```
 
+### Step 0.5:轻量节点预检
+
+目的:在不阻断流程的前提下,对章纲中的结构化节点做轻量一致性提醒。
+
+执行前记录:
+
+```bash
+python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "${PROJECT_ROOT}" workflow start-step --step-id "Step 0.5" --step-name "节点预检" || true
+```
+
+规则:
+- 只在当前章详细大纲存在 `CBN/CEN` 时执行。
+- 只检查主角或 POV 角色相关节点。
+- 第一版仅检查:
+  - `CBN` 中地点是否与 `protagonist_state.location` 明显冲突
+  - `CBN/CEN` 中主角境界或能力要求是否与 `protagonist_state.power` 明显冲突
+- 检查结果仅作为警告注入给 `context-agent`,不得阻断流程。
+- 若无节点字段,直接跳过。
+
+警告示例:
+- `[NODE_WARNING] CBN 地点与当前状态不一致: 章纲=迦南学院入口, 实际=乌坦城`
+- `[NODE_WARNING] CBN 强度要求与当前境界不一致: 章纲=斗师级压制, 实际=斗者三星`
+
+完成后记录:
+
+```bash
+python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "${PROJECT_ROOT}" workflow complete-step --step-id "Step 0.5" --artifacts '{"node_precheck":true}' || true
+```
+
 ### Step 1:调用 Context Agent 生成执行包
 
 记录开始:
@@ -126,10 +156,12 @@ python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "${PROJECT_ROOT}" wor
 - `project_root`
 - `storage_path=.webnovel/`
 - `state_file=.webnovel/state.json`
+- 若存在 `NODE_WARNING`,一并传入
 
 硬要求:
 - 输出必须包含任务书、Context Contract、Step 2A 直写提示词。
 - 执行包中必须纳入长期记忆约束与时间约束。
+- 若章纲提供结构化节点,执行包中必须包含“情节结构”板块与节拍映射。
 
 记录完成:
 
@@ -151,6 +183,7 @@ python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "${PROJECT_ROOT}" wor
 - 不得出现 `[TODO]`、`[待补充]` 等占位符。
 - 若上章存在明确钩子,本章必须回应。
 - 中文思维写作,不使用英文框架骨架驱动正文。
+- 若存在结构化节点:正文必须围绕 `CBN -> CPNs -> CEN` 展开,不得跳过必须节点。
 
 完成后记录:
 
@@ -170,6 +203,7 @@ python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "${PROJECT_ROOT}" wor
 硬要求:
 - 只改表达,不改事实、事件顺序、人物行为结果、设定规则。
 - 重点消除模板腔、说明腔、机械腔。
+- 若存在结构化节点,不得因风格改写破坏节点顺序和收束方向。
 
 完成后记录:
 
@@ -204,16 +238,25 @@ python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "${PROJECT_ROOT}" wor
 - 标准模式与 `--fast`:核心 3 个 + auto 命中的条件审查器
 - `--minimal`:只跑核心 3 个
 
-审查指标落库:
+Step 3 中间产物约定:
+- checker 原始结果:`${PROJECT_ROOT}/.webnovel/tmp/review_results.json`
+- 聚合结果:`${PROJECT_ROOT}/.webnovel/tmp/review_aggregated.json`
+- 落库指标:`${PROJECT_ROOT}/.webnovel/tmp/review_metrics.json`
+
+标准文件流:
 
 ```bash
+python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "${PROJECT_ROOT}" review-pipeline   --chapter {chapter_num}   --review-results "${PROJECT_ROOT}/.webnovel/tmp/review_results.json"   --aggregated-out "${PROJECT_ROOT}/.webnovel/tmp/review_aggregated.json"   --review-metrics-out "${PROJECT_ROOT}/.webnovel/tmp/review_metrics.json"   --report-file "审查报告/第{chapter_num}章审查报告.md"
+
 python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "${PROJECT_ROOT}" index save-review-metrics --data "@${PROJECT_ROOT}/.webnovel/tmp/review_metrics.json"
 ```
 
 硬要求:
-- 必须产出 `overall_score`。
-- `notes` 必须是单个字符串。
-- 未落库 `review_metrics` 不得进入 Step 4。
+- `review_results.json` 必须保留各 checker 原始结构化结果。
+- `review_aggregated.json` 必须包含 `overall_score`、`issues` 与 `timeline_gate`。
+- `review_metrics.json` 的 `notes` 必须是单个字符串。
+- 若 `review_aggregated.json.timeline_gate.blocked=true`,不得进入 Step 4/5。
+- Step 4 必须直接消费 `review_results.json` 与 `review_aggregated.json`。
 
 完成后记录:
 
@@ -239,6 +282,7 @@ python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "${PROJECT_ROOT}" wor
 硬要求:
 - 必须输出 `anti_ai_force_check=pass/fail`
 - `fail` 时不得进入 Step 5
+- 有节点时,不得在润色中删除必须节点对应的情节落点
 
 完成后记录:
 
@@ -321,11 +365,13 @@ python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "${PROJECT_ROOT}" wor
 未满足以下条件前,不得结束流程:
 
 1. 章节正文文件存在且非空。
-2. Step 3 已产出 `overall_score`,且 `review_metrics` 已落库。
-3. Step 4 的 `anti_ai_force_check=pass`。
-4. Step 5 已更新 `state.json`、`index.db`、`summaries/ch{chapter_padded}.md`。
-5. Step 5 已写入 `.webnovel/memory_scratchpad.json`。
-6. 若启用观测,已读取最新 timing 记录并给出结论。
+2. Step 3 已产出 `${PROJECT_ROOT}/.webnovel/tmp/review_aggregated.json`,其中包含 `overall_score`。
+3. Step 3 的 `review_metrics` 已落库。
+4. 若存在 `timeline_gate.blocked=true`,流程必须停在 Step 3。
+5. Step 4 的 `anti_ai_force_check=pass`。
+6. Step 5 已更新 `state.json`、`index.db`、`summaries/ch{chapter_padded}.md`。
+7. Step 5 已写入 `.webnovel/memory_scratchpad.json`。
+8. 若启用观测,已读取最新 timing 记录并给出结论。
 
 ## 验证与交付