框架: B4-S3 出卡权上移窗级——章级候选线索+窗级聚类母卡+间隔机械计算+嵌入判重

- parse_llm 拆 chapters/cards 两命令:章级一遍正文产细纲+实体+范式候选线索;窗级按大纲窗行聚类归并出母卡+多实例章号
- parse_ingest 新增 cards 窗级入库:实例域校验/间隔章号差机械追加(伪精确灭绝)/嵌入判重0.85归并终判/越合同字段裁剪降级/占位符机械检测
- parse_outline 幂等键改 from_order(流水号重跑漂移)+书末残窗必成窗+--to 限域
- 出卡环加拒卡带因重试+0卡可疑重试(M3 输出方差实测)
- review_cards CRITERIA 补实测间隔可信例外+溯源豁免区说明(防审核假阳)
- 章级判重名录随窗级聚类废除(重跑自噬两补丁退役)

.claude/skills/parse-book/SKILL.md

@@ -31,18 +31,22 @@ disable-model-invocation: true
 - **顺序性只来自增量判重**(新实体要对着已积累实体判重合并),细纲逆推本身章间独立——先顺序跑保正确,并行化留作后续优化;
 - 进度每 10 章报一行(章号/新实体数/累计分型统计)。
 
-**M3 直调形态（创始人 2026-07-13 拍板,现行）**：循环体不再派 opus/haiku 子代理,改为 `scripts/parse_llm.py` 直调 New-API `MiniMax-M3`（经 llm skill）——脚本自己取数（正文/前文实体/已积累卡名录内联进 prompt）→ M3 两 pass（脚手架→范式）→ `parse_ingest` 机械校验入库（比例硬顶/归型/字段越合同/15连字泄漏）。断点续跑按 example_parse_task 状态机逐章跳过已 done 的 pass。
+**M3 直调形态（创始人 2026-07-13 拍板,现行）**：循环体不再派 opus/haiku 子代理,改为 `scripts/parse_llm.py` 直调 New-API `MiniMax-M3`（经 llm skill）。
 
-**放量全流程（每书四步,均 M3）**：
+**出卡权上移窗级（B4-S3 重构,现行）**：章级逐章出卡有三同根病（同功撞车/单章证不成跨章公式/间隔数字伪精确）,治法=章级只产「范式候选线索」（并入脚手架 pass,正文只过一遍）,出卡在窗级聚类归并——同一手法多章多次出现归并成一张母卡+实例章号,**间隔章数由实例章号差机械计算**（M3 禁自报数字）。跨窗/跨书同手法靠**嵌入判重**（初筛 ≥0.85 → M3 归并终判 merge/keep,拿不准保留）。大纲窗行（example_parse_outline,幂等键=窗起始章）是出卡窗的唯一切分依据。
+
+**放量全流程（每书五步,均 M3）**：
 
 ```bash
-# ① 逐章两 pass（脚手架→范式;断点续跑,重跑自动补失败章）
+# ① 章级 pass（细纲+实体+范式候选线索;断点续跑,重跑自动补失败章）
-.venv/bin/python .claude/skills/parse-book/scripts/parse_llm.py --work-id 4 --from 1 --to 50
+.venv/bin/python .claude/skills/parse-book/scripts/parse_llm.py chapters --work-id 4 --from 1 --to 50
-# ② 窗级大纲聚合（每 5–10 万字:多章细纲+正文→阶段大纲;单章对大纲层可能零贡献,不做章内比例强制）
+# ② 窗级大纲聚合（每 5–10 万字:多章细纲+正文→阶段大纲;书末残窗无论大小必成窗）
 .venv/bin/python .claude/skills/parse-book/scripts/parse_outline.py window --work-id 4
-# ③ 全书拆完:终检（逐窗细纲对账大纲 + 跨段连贯性纵览）
+# ③ 窗级聚类出卡（窗=②的窗行;线索+细纲+阶段大纲→母卡;守卫+判重在 parse_ingest cards）
+.venv/bin/python .claude/skills/parse-book/scripts/parse_llm.py cards --work-id 4
+# ④ 全书拆完:终检（逐窗细纲对账大纲 + 跨段连贯性纵览）
 .venv/bin/python .claude/skills/parse-book/scripts/parse_outline.py check --work-id 4
-# ④ 公共卡三角色审核（番茄作家/起点作家/主编,M3 常设步骤;见 review-cards skill）
+# ⑤ 公共卡三角色审核（番茄作家/起点作家/主编,M3 常设步骤;见 review-cards skill）
 .venv/bin/python .claude/skills/review-cards/scripts/review_cards.py review --batch <批次> --work-id 4
 # 进度
 .venv/bin/python .claude/skills/parse-book/scripts/parse_ingest.py progress
@@ -50,7 +54,7 @@ disable-model-invocation: true
 
 审核纪律：常设审核=M3（已用 opus 金标准校准,偏差 0.45 达标）；fable/opus 只做起量前校准与起量后一次总审核（门禁与优化,不进流程循环）。
 
-试拆实测病理档（放量前须知）：M3 细纲按比例会写超（治=绝对字数上限+压缩重试）；type 一律标 craft 且串型混填（prompt 矫正无效,治=B4 环按金标准清单改造 prompt/schema）；重跑自噬（治=判重名录排除本章+0 卡不软删）；source 偷懒（治=机械回填）。
+试拆实测病理档（放量前须知）：M3 细纲按比例会写超（治=绝对字数上限+压缩重试）；type 一律标 craft 且串型混填（治=合同动态渲染+字段指纹改型+窗级聚类）；重跑自噬（治=窗级幂等软删+0 卡不软删,章级判重名录已随 S3 废除）；source 偷懒（治=出处由实例机械生成）。parse_ingest 的 patterns 命令是 S3 前章级出卡入口,仅留作回滚保险。
 
 ## 步骤(自底向上,与创作期规划的自顶向下互为镜像)
 

.claude/skills/parse-book/scripts/parse_ingest.py

@@ -1,9 +1,14 @@
 #!/usr/bin/env python3
-"""parse-book 配套确定性脚本：拆书产物校验入库 + 任务状态机（B2）。
+"""parse-book 配套确定性脚本：拆书产物校验入库 + 任务状态机（B2/B4-S3）。
 
 职责边界：extractor(LLM) 只产结构化 JSON 文件，不碰库；本脚本做机械校验后写库——
 字段 key 合法性（对库内字段合同）、五型归型、出处必填、**脱敏红线 15 连字检测**（硬阻断）。
-状态全在库（example_parse_task），断点续跑与幂等按章。
+状态全在库（example_parse_task），断点续跑与幂等按章/按窗。
+
+B4-S3 起管线两级：scaffold（章级：细纲+实体+范式候选线索）→ cards（窗级：聚类母卡）。
+本脚本仅在两个受控点调用 LLM/嵌入服务（不产内容）：
+  ① 新卡嵌入（判重与检索基座共用）；② 相似度 ≥0.85 时的归并判定（输入两卡 JSON，输出 merge/keep）。
+patterns 命令是 S3 前的章级出卡入口，保留作回滚保险，新管线不再使用。
 """
 import hashlib
 import json
@@ -15,11 +20,20 @@ import click
 import psycopg
 from psycopg.types.json import Jsonb
 
+# 受控点依赖：嵌入走 embed skill 同一实现（同模型同维），归并判定走 llm skill 统一入口
+sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2] / "embed" / "scripts"))
+sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2] / "llm" / "scripts"))
+from embed_drafts import _session, build_embed_text, embed_texts  # noqa: E402
+from llm import chat, extract_json  # noqa: E402
+
 DSN = ("postgresql://root:f6710e2d0294eb1c10e26a805a64bc54@100.64.0.8:5433/muse-example"
        "?keepalives=1&keepalives_idle=15&keepalives_interval=5&keepalives_count=3")
 TENANT, ACTOR = 1, "1"
 PATTERN_TYPES = {"craft", "combat", "emotion", "scene_pattern", "trope"}  # 拍板①：首轮只拆五型
 NGRAM = 15  # 脱敏红线：≥15 连续字与原文重合=违规（parse-book skill）
+EMBED_MODEL = "Qwen/Qwen3-Embedding-8B"
+SIM_MERGE = 0.85   # 嵌入判重：≥该值触发 M3 归并判定（阈值未校准，实战收集中）
+SIM_MARK = 0.75    # [MARK, MERGE) 区间只标记不判定（校准带宽，审核环可见）
 
 
 def norm_scaffold(data: dict) -> dict:
@@ -29,15 +43,27 @@ def norm_scaffold(data: dict) -> dict:
     for e in data.get("实体") or data.get("entities") or []:
         ents.append({"型": e.get("型") or e.get("type"), "名称": e.get("名称") or e.get("name"),
                      "一句话摘要": e.get("一句话摘要") or e.get("brief")})
-    return {"细纲": data.get("细纲") or data.get("outline") or "", "实体": ents}
+    hints = []
+    for h in data.get("线索") or data.get("hints") or []:
+        hints.append({"型": h.get("型") or h.get("type"), "短名": h.get("短名") or h.get("name"),
+                      "线索": h.get("线索") or h.get("clue"), "证据": h.get("证据") or h.get("evidence")})
+    return {"细纲": data.get("细纲") or data.get("outline") or "", "实体": ents, "线索": hints}
 
 
 def norm_card(c: dict) -> dict:
-    """范式卡键名归一（ASCII→中文）。"""
+    """范式卡键名归一（ASCII→中文）；窗级卡带实例数组 [{章,定位}]。"""
     src = c.get("出处") or c.get("source") or {}
+    inss = []
+    for ins in c.get("实例") or c.get("instances") or []:
+        try:
+            ch_no = int(ins.get("章") or ins.get("ch"))
+        except (TypeError, ValueError):
+            continue  # 章号非数字=编造，丢弃该实例
+        inss.append({"章": ch_no, "定位": ins.get("定位") or ins.get("anchor") or ""})
     return {"型": c.get("型") or c.get("type"), "名称": c.get("名称") or c.get("name"),
             "一句话摘要": c.get("一句话摘要") or c.get("brief"),
             "字段": c.get("字段") or c.get("fields") or {},
+            "实例": inss,
             "出处": {"书名": src.get("书名") or src.get("book"),
                      "回目": src.get("回目") or src.get("chapter"),
                      "定位": src.get("定位") or src.get("anchor")}}
@@ -66,6 +92,16 @@ def field_contract(conn, ttype):
     return {r[0] for r in rows}
 
 
+def load_entity_names(conn, work_id):
+    """本书专名词典：scaffold 实体名就是现成专名表（短专名泄漏 15 连字抓不到，
+    金标准抓到 4 张「海拉/千亿夸克」级泄漏）；出处/实例定位字段豁免（设计允许专名只进出处）。"""
+    names = {e.get("名称") or e.get("name") for (ents,) in conn.execute(
+        """SELECT entities FROM example_parse_scaffold
+           WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE""",
+        (TENANT, work_id)).fetchall() for e in ents}
+    return {n for n in names if n and len(n) >= 2}
+
+
 def leak_check(card_texts, source_text):
     """脱敏机械检查：卡内任一文本值含与原文 ≥NGRAM 连续字重合 → 返回违规片段。"""
     src = re.sub(r'\s', '', source_text)
@@ -78,6 +114,68 @@ def leak_check(card_texts, source_text):
     return None
 
 
+def validate_card(c, contracts, entity_names, src_text):
+    """卡片通用机械守卫（章级 patterns 与窗级 cards 共用——守卫同源，禁两处各写一套）。
+    返回 (拒卡原因列表, 改型审计|None)；c 可能被就地改型。"""
+    t = c.get("型")
+    reasons, retyped = [], None
+    if t not in PATTERN_TYPES:
+        reasons.append(f"型不合法:{t}（首轮只拆五型）")
+    if not c.get("名称"):
+        reasons.append("缺名称")
+    fields = c.get("字段") or {}
+    if t in contracts and fields:
+        illegal = set(fields) - contracts[t]
+        if illegal:
+            # 字段指纹改型：LLM 的 type 标注不可靠（实测 M3 系统性全标 craft），
+            # 五型字段 key 互不重叠——字段集合全命中唯一他型合同时，以字段为准改型
+            fits = [t2 for t2 in PATTERN_TYPES
+                    if t2 != t and fields and set(fields) <= contracts[t2]]
+            if len(fits) == 1:
+                retyped = f"{t}→{fits[0]}(字段指纹改型)"
+                c["型"] = t = fits[0]
+            else:
+                # 裁剪降级（B4-S3 实测：M3 往 trope 塞「读者收益」类他型增益字段屡教不改）：
+                # 越界字段剥离（内容在本型无处安放，不该陪葬整卡），审计留痕；
+                # 裁掉必填字段的情况由下方 craft 双模板必填校验兜底拒卡
+                for k in illegal:
+                    fields.pop(k)
+                c["裁剪字段"] = sorted(illegal)
+    # 守卫（B4-S2）：craft 双模板条件必填/禁填 + 装置类型闭合枚举 + 禁复合标签
+    if t == "craft" and not reasons:
+        form = (fields.get("装置形态") or "").strip()
+        dtype = (fields.get("装置类型") or "").strip()
+        if form not in ("结构装置", "场景手法"):
+            reasons.append(f"装置形态非法:{form!r}（须为 结构装置/场景手法）")
+        elif form == "结构装置":
+            miss = [k for k in ("埋设手法", "回收点", "记忆维持", "间隔纪律") if not fields.get(k)]
+            if miss:
+                reasons.append(f"结构装置必填缺失:{miss}")
+        else:  # 场景手法：禁填间隔纪律（同场景闭环没有间隔，硬填必出伪纪律）
+            miss = [k for k in ("复用节奏", "异质化要求", "单章上限") if not fields.get(k)]
+            if miss:
+                reasons.append(f"场景手法必填缺失:{miss}")
+            if fields.get("间隔纪律"):
+                reasons.append("场景手法禁填间隔纪律（伪纪律来源）")
+        if dtype:
+            if any(sep in dtype for sep in ("+", "、", "/", "，")):
+                reasons.append(f"装置类型禁复合标签:{dtype!r}（只填最主要的一个）")
+            elif dtype not in ("伏笔", "契诃夫之枪", "信息差", "重复意象", "倒计时", "身份错认", "非装置"):
+                reasons.append(f"装置类型不在闭合枚举:{dtype!r}")
+        # 原理句式占位符检测（实测病：M3 把模板「当X时做Y因为读者会Z」的占位符字面抄进卡）
+        if re.search(r"[做当会][XYZ]", str(fields.get("原理") or "")):
+            reasons.append("原理含未展开的模板占位符（X/Y/Z 须替换为具体内容）")
+    # 脱敏红线 + 专名词典（实例定位/出处豁免——设计允许专名只进定位）
+    texts = [x for x in (c.get("名称"), c.get("一句话摘要"), *fields.values()) if x]
+    leak = leak_check(texts, src_text)
+    if leak:
+        reasons.append(f"脱敏违规(≥{NGRAM}连字重合):「{leak}」")
+    hit_names = [n for n in entity_names if any(n in str(x) for x in texts)]
+    if hit_names:
+        reasons.append(f"专名泄漏(本书实体名):{hit_names[:3]}")
+    return reasons, retyped
+
+
 def set_task(conn, work_id, chapter_id, **cols):
     """推进任务状态机（attempt 自增）。"""
     sets = ", ".join(f"{k}=%s" for k in cols)
@@ -124,7 +222,7 @@ def init_tasks(work_id, from_, to):
 @click.option("--chapter-order", type=int, required=True)
 @click.option("--file", "file_", type=click.Path(exists=True), required=True)
 def scaffold(work_id, chapter_order, file_):
-    """脚手架入库：{细纲, 实体:[{型,名称,一句话摘要,备注?}]}；比例约束校验（3–5%，超标拒绝）。"""
+    """章级入库：{细纲, 实体:[{型,名称,一句话摘要}], 线索:[{型,短名,线索,证据}]}；比例约束校验。"""
     data = norm_scaffold(json.loads(pathlib.Path(file_).read_text()))
     with psycopg.connect(DSN) as conn:
         ch_id, src = chapter_of(conn, work_id, chapter_order)
@@ -141,17 +239,25 @@ def scaffold(work_id, chapter_order, file_):
         for e in ents:
             if not e.get("名称") or not e.get("型"):
                 raise click.ClickException(f"实体缺 名称/型: {e}")
+        # 范式候选线索（B4-S3）：候选层宽容处置——坏行丢弃计数，不整批拒
+        hints, n_bad = [], 0
+        for h in data.get("线索") or []:
+            if h.get("短名") and h.get("线索") and (h.get("型") in PATTERN_TYPES or h.get("型") == "?"):
+                hints.append(h)
+            else:
+                n_bad += 1
         conn.execute(
             """INSERT INTO example_parse_scaffold (work_id, chapter_id, outline_text, entities,
-                   creator, updater, tenant_id)
+                   pattern_hints, creator, updater, tenant_id)
-               VALUES (%s,%s,%s,%s,%s,%s,%s)
+               VALUES (%s,%s,%s,%s,%s,%s,%s,%s)
                ON CONFLICT (tenant_id, chapter_id)
                DO UPDATE SET outline_text=EXCLUDED.outline_text, entities=EXCLUDED.entities,
-                             deleted=FALSE, updater=EXCLUDED.updater""",
+                             pattern_hints=EXCLUDED.pattern_hints, deleted=FALSE, updater=EXCLUDED.updater""",
-            (work_id, ch_id, outline, Jsonb(ents), ACTOR, ACTOR, TENANT))
+            (work_id, ch_id, outline, Jsonb(ents), Jsonb(hints), ACTOR, ACTOR, TENANT))
         set_task(conn, work_id, ch_id, scaffold_status="done", error_message=None)
         conn.commit()
-        click.echo(f"scaffold✓ work={work_id} ch#{chapter_order}: 细纲{len(outline)}字({ratio:.1%}) 实体{len(ents)}")
+        click.echo(f"scaffold✓ work={work_id} ch#{chapter_order}: 细纲{len(outline)}字({ratio:.1%}) "
+                   f"实体{len(ents)} 线索{len(hints)}" + (f"(丢弃坏行{n_bad})" if n_bad else ""))
 
 
 @cli.command()
@@ -159,8 +265,7 @@ def scaffold(work_id, chapter_order, file_):
 @click.option("--chapter-order", type=int, required=True)
 @click.option("--file", "file_", type=click.Path(exists=True), required=True)
 def patterns(work_id, chapter_order, file_):
-    """范式卡入库：[{型∈五型, 名称, 一句话摘要, 字段{…}, 出处{书名,回目,定位}}] → draft(pending)。
+    """【旧管线·回滚保险】章级范式卡入库；B4-S3 后由窗级 cards 替代，新管线勿用。"""
-    机械硬阻断：归型合法、字段 key 合法（库内合同）、出处必填、15 连字脱敏检测。"""
     raw = json.loads(pathlib.Path(file_).read_text())
     if isinstance(raw, dict):  # 兼容 {cards:[…]}/{卡:[…]} 包装
         raw = raw.get("cards") or raw.get("卡") or []
@@ -171,16 +276,9 @@ def patterns(work_id, chapter_order, file_):
         ch_id, src = chapter_of(conn, work_id, chapter_order)
         contracts = {t: field_contract(conn, t) for t in PATTERN_TYPES}
         book = conn.execute("SELECT title FROM muse_content_work WHERE id=%s", (work_id,)).fetchone()[0]
-        # 专名词典（B4-S2）：本书 scaffold 实体名就是现成专名表——短专名泄漏 15 连字抓不到
+        entity_names = load_entity_names(conn, work_id)
-        #（金标准抓到 4 张「海拉/千亿夸克」级泄漏），出处字段豁免（设计允许专名只进出处）
-        entity_names = {e.get("名称") or e.get("name") for (ents,) in conn.execute(
-            """SELECT entities FROM example_parse_scaffold
-               WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE""",
-            (TENANT, work_id)).fetchall() for e in ents}
-        entity_names = {n for n in entity_names if n and len(n) >= 2}
         # 幂等：重跑本章 = 软删本章旧 draft。
         # 但新轮 0 卡时保留旧卡——LLM 判卡有随机性，重跑「0 卡」不应清掉上轮已验证的产出
-        #（实测组合病：判重名录+随机性+软删 → 每次重跑都损耗卡）
         if cards:
             conn.execute(
                 """UPDATE muse_knowledge_draft SET deleted=TRUE, updater=%s
@@ -189,67 +287,20 @@ def patterns(work_id, chapter_order, file_):
                 (ACTOR, TENANT, work_id, str(chapter_order)))
         ok, rejected = 0, []
         for i, c in enumerate(cards):
-            t = c.get("型")
+            reasons, retyped = validate_card(c, contracts, entity_names, src)
-            reasons = []
-            if t not in PATTERN_TYPES:
-                reasons.append(f"型不合法:{t}（首轮只拆五型）")
-            if not c.get("名称"):
-                reasons.append("缺名称")
             src_ref = c.get("出处") or {}
             if not (src_ref.get("书名") and src_ref.get("回目")):
                 reasons.append("出处不完整（需书名+回目）")
-            fields = c.get("字段") or {}
-            retyped = None
-            if t in contracts and fields:
-                illegal = set(fields) - contracts[t]
-                if illegal:
-                    # 字段指纹改型：LLM 的 type 标注不可靠（实测 M3 系统性全标 craft），
-                    # 五型字段 key 互不重叠——字段集合全命中唯一他型合同时，以字段为准改型
-                    fits = [t2 for t2 in PATTERN_TYPES
-                            if t2 != t and fields and set(fields) <= contracts[t2]]
-                    if len(fits) == 1:
-                        retyped = f"{t}→{fits[0]}(字段指纹改型)"
-                        c["型"] = t = fits[0]
-                    else:
-                        reasons.append(f"字段越合同:{sorted(illegal)}")
-            # 守卫（B4-S2）：craft 双模板条件必填/禁填 + 装置类型闭合枚举 + 禁复合标签
-            if t == "craft" and not reasons:
-                form = (fields.get("装置形态") or "").strip()
-                dtype = (fields.get("装置类型") or "").strip()
-                if form not in ("结构装置", "场景手法"):
-                    reasons.append(f"装置形态非法:{form!r}（须为 结构装置/场景手法）")
-                elif form == "结构装置":
-                    miss = [k for k in ("埋设手法", "回收点", "记忆维持", "间隔纪律") if not fields.get(k)]
-                    if miss:
-                        reasons.append(f"结构装置必填缺失:{miss}")
-                else:  # 场景手法：禁填间隔纪律（同场景闭环没有间隔，硬填必出伪纪律）
-                    miss = [k for k in ("复用节奏", "异质化要求", "单章上限") if not fields.get(k)]
-                    if miss:
-                        reasons.append(f"场景手法必填缺失:{miss}")
-                    if fields.get("间隔纪律"):
-                        reasons.append("场景手法禁填间隔纪律（伪纪律来源）")
-                if dtype and t == "craft":
-                    if any(sep in dtype for sep in ("+", "、", "/", "，")):
-                        reasons.append(f"装置类型禁复合标签:{dtype!r}（只填最主要的一个）")
-                    elif dtype not in ("伏笔", "契诃夫之枪", "信息差", "重复意象", "倒计时", "身份错认", "非装置"):
-                        reasons.append(f"装置类型不在闭合枚举:{dtype!r}")
-            texts = [c.get("名称"), c.get("一句话摘要"), *fields.values()]
-            leak = leak_check([x for x in texts if x], src)
-            if leak:
-                reasons.append(f"脱敏违规(≥{NGRAM}连字重合):「{leak}」")
-            # 专名词典拒卡：卡文本（不含出处）命中本书实体名 → 拒
-            hit_names = [n for n in entity_names
-                         if any(n in str(x) for x in texts if x)]
-            if hit_names:
-                reasons.append(f"专名泄漏(本书实体名):{hit_names[:3]}")
             if reasons:
                 rejected.append({"卡": c.get("名称") or f"#{i}", "原因": reasons})
                 continue
-            payload = {"型": t, "名称": c["名称"], "一句话摘要": c.get("一句话摘要", ""),
+            payload = {"型": c["型"], "名称": c["名称"], "一句话摘要": c.get("一句话摘要", ""),
-                       "字段": fields, "出处": src_ref, "目标库": "公共范式库",
+                       "字段": c.get("字段") or {}, "出处": src_ref, "目标库": "公共范式库",
                        "章序": chapter_order, "来源": f"拆书@{book}", "状态": "草稿"}
             if retyped:
                 payload["改型"] = retyped  # 审计：机械改型可追溯
+            if c.get("裁剪字段"):
+                payload["裁剪字段"] = c["裁剪字段"]  # 审计：越合同字段被剥离入库
             cid = f"parse-{work_id}-{chapter_order}-{i}-" + hashlib.sha256(
                 json.dumps(payload, ensure_ascii=False, sort_keys=True).encode()).hexdigest()[:8]
             conn.execute(
@@ -259,7 +310,7 @@ def patterns(work_id, chapter_order, file_):
                    ON CONFLICT (tenant_id, command_id) WHERE command_id IS NOT NULL DO NOTHING""",
                 (Jsonb(payload), work_id, cid, ACTOR, ACTOR, TENANT))
             ok += 1
-        set_task(conn, work_id, ch_id, pattern_status="done" if not rejected else "done",
+        set_task(conn, work_id, ch_id, pattern_status="done",
                  error_message=None if not rejected else f"拒卡{len(rejected)}: " + json.dumps(rejected, ensure_ascii=False)[:900])
         conn.commit()
         click.echo(f"patterns✓ work={work_id} ch#{chapter_order}: 入库{ok} 拒{len(rejected)}")
@@ -267,6 +318,165 @@ def patterns(work_id, chapter_order, file_):
             click.echo(f"  [拒] {r['卡']}: {'; '.join(r['原因'])}")
 
 
+def merge_judge(new_payload, old_payload, model="MiniMax-M3"):
+    """受控 LLM 点②：嵌入初筛 ≥SIM_MERGE 后的归并终判（输入两卡 JSON，输出 merge/keep）。
+    双保险设计：embedding 只做初筛，是否同一手法由 LLM 判——单靠阈值必产生错误合并。"""
+    prompt = ("两张「公共范式卡」由不同窗/不同书各自独立归纳（JSON 附后）。判断它们是否为同一写作手法：\n"
+              "- 运作机制相同（哪怕载体/题材/叫法不同）→ merge；\n"
+              "- 机制不同、或适用场景本质不同 → keep。拿不准时选 keep（错误合并比重复卡更伤）。\n"
+              '输出规则（只输出一个 JSON 对象）：{"verdict": "merge"或"keep", "reason": "一句话依据"}\n\n'
+              f"【卡A（已入库）】\n{json.dumps(old_payload, ensure_ascii=False)}\n\n"
+              f"【卡B（新卡）】\n{json.dumps(new_payload, ensure_ascii=False)}")
+    try:
+        content, _ = chat(prompt, model=model)
+        data = extract_json(content)
+        return data.get("verdict"), data.get("reason", "")
+    except Exception as e:  # 判定失败=keep（不阻断入库，宁重复不误并）
+        return "keep", f"归并判定调用失败({type(e).__name__})，默认保留"
+
+
+@cli.command()
+@click.option("--work-id", type=int, required=True)
+@click.option("--from-order", "from_order", type=int, required=True, help="窗起始章（对应大纲窗行）")
+@click.option("--file", "file_", type=click.Path(exists=True), required=True)
+@click.option("--model", default="MiniMax-M3", show_default=True, help="归并判定用模型")
+def cards(work_id, from_order, file_, model):
+    """窗级母卡入库（B4-S3）：{cards:[{型,名称,摘要,字段,实例[{章,定位}]}]}。
+    守卫=章级全部 + 窗级新增：实例域校验 / 间隔章数机械计算（伪精确灭绝）/ 嵌入判重→归并判定。"""
+    raw = json.loads(pathlib.Path(file_).read_text())
+    if isinstance(raw, dict):
+        raw = raw.get("cards") or raw.get("卡") or []
+    cards_in = [norm_card(c) for c in raw]
+    with psycopg.connect(DSN) as conn:
+        win = conn.execute(
+            """SELECT to_order FROM example_parse_outline
+               WHERE tenant_id=%s AND work_id=%s AND from_order=%s AND deleted=FALSE""",
+            (TENANT, work_id, from_order)).fetchone()
+        if not win:
+            raise click.ClickException(f"窗不存在: work={work_id} from={from_order}——先跑 parse_outline window")
+        to_order = win[0]
+        book = conn.execute("SELECT title FROM muse_content_work WHERE id=%s", (work_id,)).fetchone()[0]
+        # 窗内正文拼接（15 连字红线对照源：卡可能引到窗内任何一章）
+        src = "".join(t for (t,) in conn.execute(
+            """SELECT b.content_text FROM muse_content_chapter c
+               JOIN muse_content_block b ON b.chapter_id=c.id AND b.deleted=FALSE
+               WHERE c.tenant_id=%s AND c.work_id=%s AND c.order_no BETWEEN %s AND %s
+                 AND c.deleted=FALSE ORDER BY c.order_no""",
+            (TENANT, work_id, from_order, to_order)).fetchall())
+        contracts = {t: field_contract(conn, t) for t in PATTERN_TYPES}
+        entity_names = load_entity_names(conn, work_id)
+        # 幂等：重出本窗 = 软删本窗旧卡（0 卡保留旧卡，纪律同章级）
+        if cards_in:
+            conn.execute(
+                """UPDATE muse_knowledge_draft SET deleted=TRUE, updater=%s
+                   WHERE tenant_id=%s AND source_type='parse_book' AND source_id=%s
+                     AND draft_payload->>'窗起'=%s AND deleted=FALSE""",
+                (ACTOR, TENANT, work_id, str(from_order)))
+        sess = _session()  # 嵌入通道（判重+新卡即时嵌入共用）
+        ok, merged, rejected = 0, [], []
+        for i, c in enumerate(cards_in):
+            reasons, retyped = validate_card(c, contracts, entity_names, src)
+            # 实例域校验：章号必须落在本窗内——越窗章号=编造（跨窗同手法靠嵌入判重归并）
+            inss = [ins for ins in c.get("实例") or [] if from_order <= ins["章"] <= to_order]
+            n_drop = len(c.get("实例") or []) - len(inss)
+            if not inss:
+                reasons.append("实例为空或章号全部越窗（窗级卡必须带窗内实例）")
+            fields = c.get("字段") or {}
+            # 间隔章数机械计算（伪精确灭绝）：结构装置实例≥2 → 章号差为权威值追加；
+            # 单实例结构装置=本窗证据不全（可能收在后续窗），标「跨窗待证」入库不拒
+            cross_pend = False
+            if not reasons and c.get("型") == "craft" and fields.get("装置形态") == "结构装置":
+                ch_nos = sorted({ins["章"] for ins in inss})
+                if len(ch_nos) >= 2:
+                    fields["间隔纪律"] = ((fields.get("间隔纪律") or "").strip()
+                                          + f"（实测：#{ch_nos[0]}埋→#{ch_nos[-1]}收，隔{ch_nos[-1] - ch_nos[0]}章）").strip()
+                else:
+                    cross_pend = True
+            if reasons:
+                rejected.append({"卡": c.get("名称") or f"#{i}", "原因": reasons})
+                continue
+            payload = {"型": c["型"], "名称": c["名称"], "一句话摘要": c.get("一句话摘要", ""),
+                       "字段": fields, "实例": inss,
+                       "出处": {"书名": book, "回目": f"第{inss[0]['章']}章", "定位": inss[0]["定位"]},
+                       "目标库": "公共范式库", "窗起": from_order, "来源": f"拆书@{book}", "状态": "草稿"}
+            if retyped:
+                payload["改型"] = retyped
+            if c.get("裁剪字段"):
+                payload["裁剪字段"] = c["裁剪字段"]  # 审计：越合同字段被剥离入库
+            if cross_pend:
+                payload["跨窗待证"] = True  # 审核环可见：结构装置只有单端实例
+            if n_drop:
+                payload["越窗实例弃"] = n_drop  # 审计：M3 报了窗外章号
+            # ── 受控点①②：嵌入判重（失败不阻断——判重是增强不是红线，卡不能因通道故障丢）──
+            embed_text = build_embed_text(payload)
+            vec = None
+            try:
+                vecs, bad = embed_texts(sess, [embed_text])
+                vec = None if bad else vecs[0]
+            except Exception:
+                pass
+            if vec is not None:
+                top = conn.execute(
+                    """SELECT d.id, d.draft_payload, 1-(e.embedding<=>%s::vector) AS score
+                       FROM example_knowledge_embedding e
+                       JOIN muse_knowledge_draft d ON d.id=e.draft_id
+                       WHERE e.tenant_id=%s AND e.deleted=FALSE AND d.deleted=FALSE
+                         AND d.source_type='parse_book'
+                       ORDER BY score DESC LIMIT 1""", (json.dumps(vec), TENANT)).fetchone()
+                if top and top[2] >= SIM_MERGE:
+                    old_id, old_payload, score = top
+                    verdict, why = merge_judge(payload, old_payload, model)
+                    if verdict == "merge":
+                        # 归并=旧卡追加实例（跨书时带书名）+完整审计（含被归并卡全文，可逆）
+                        add = [dict(ins, 书=book) for ins in inss]
+                        old_payload["实例"] = (old_payload.get("实例") or []) + add
+                        old_payload.setdefault("归并审计", []).append(
+                            {"相似度": round(float(score), 4), "判定依据": why, "被归并卡": payload})
+                        conn.execute(
+                            "UPDATE muse_knowledge_draft SET draft_payload=%s, updater=%s WHERE id=%s",
+                            (Jsonb(old_payload), ACTOR, old_id))
+                        merged.append(f"《{payload['名称']}》并入已有卡#{old_id}《{old_payload.get('名称')}》({score:.2f}) {why}")
+                        continue
+                    payload["判重"] = {"相似卡": old_payload.get("名称"), "相似度": round(float(score), 4),
+                                       "判定": "keep", "依据": why}
+                elif top and top[2] >= SIM_MARK:
+                    payload["判重"] = {"相似卡": top[1].get("名称"), "相似度": round(float(top[2]), 4),
+                                       "判定": "未达判定线"}
+            cid = f"parse-{work_id}-w{from_order}-{i}-" + hashlib.sha256(
+                json.dumps(payload, ensure_ascii=False, sort_keys=True).encode()).hexdigest()[:8]
+            row = conn.execute(
+                """INSERT INTO muse_knowledge_draft (work_id, draft_type, draft_payload, status,
+                       source_type, source_id, command_id, creator, updater, tenant_id)
+                   VALUES (0,'entity',%s,'pending','parse_book',%s,%s,%s,%s,%s)
+                   ON CONFLICT (tenant_id, command_id) WHERE command_id IS NOT NULL DO NOTHING
+                   RETURNING id""",
+                (Jsonb(payload), work_id, cid, ACTOR, ACTOR, TENANT)).fetchone()
+            # 新卡即时嵌入（下一窗/下一书判重立即可见；失败留给 embed skill 批量补）
+            if row and vec is not None:
+                h = hashlib.sha256(f"{embed_text}|{EMBED_MODEL}".encode()).hexdigest()
+                conn.execute(
+                    """INSERT INTO example_knowledge_embedding
+                           (draft_id, content_hash, embed_text, model, dimensions, embedding,
+                            creator, updater, tenant_id)
+                       VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
+                       ON CONFLICT (tenant_id, content_hash, model) DO NOTHING""",
+                    (row[0], h, embed_text, EMBED_MODEL, 1024, json.dumps(vec), ACTOR, ACTOR, TENANT))
+            ok += 1
+        # 窗内章 pattern_status 批量置 done（窗级出卡完成=这些章的范式阶段完成）
+        conn.execute(
+            """UPDATE example_parse_task SET pattern_status='done', updater=%s
+               WHERE tenant_id=%s AND work_id=%s AND chapter_id IN (
+                   SELECT id FROM muse_content_chapter
+                   WHERE tenant_id=%s AND work_id=%s AND order_no BETWEEN %s AND %s AND deleted=FALSE)""",
+            (ACTOR, TENANT, work_id, TENANT, work_id, from_order, to_order))
+        conn.commit()
+        click.echo(f"cards✓ work={work_id} win#{from_order}–{to_order}: 入库{ok} 归并{len(merged)} 拒{len(rejected)}")
+        for m in merged:
+            click.echo(f"  [归并] {m}")
+        for r in rejected:
+            click.echo(f"  [拒] {r['卡']}: {'; '.join(r['原因'])}")
+
+
 @cli.command()
 @click.option("--work-id", type=int)
 def progress(work_id):

.claude/skills/parse-book/scripts/parse_llm.py

@@ -1,12 +1,18 @@
 #!/usr/bin/env python3
-"""parse-book skill：M3 直调拆书执行器——逐章两 pass（脚手架→范式），入库走 parse_ingest 守卫。
+"""parse-book skill：M3 直调拆书执行器（B4-S3 重构：出卡权上移窗级）。
 
 创始人拍板（2026-07-13）：拆书内容生产 LLM=New-API MiniMax-M3（经 llm skill），
-不再派 opus/haiku 子代理。本脚本把 workflow 版的提示词资产（身份段/实体判据/五型合同）
+不再派 opus/haiku 子代理。B4 fable 审查裁决（2026-07-13）：章级逐章出卡有三同根病
-固化为直调版：脚本自己取数（正文/前文实体/已积累卡名录内联进 prompt）、调 M3、
+（同功 family 撞车/单章证不成跨章公式/间隔数字伪精确），治法=章级只产「范式候选线索」，
-容错解析 JSON、经 parse_ingest 机械校验入库（比例硬顶/归型/字段越合同/15连字泄漏全在那边）。
+出卡在窗级聚类归并（复用大纲窗切分）——同一手法多章多次出现归并为一张母卡+实例章号，
+间隔数字由实例章号差机械计算。
 
-断点续跑：按 example_parse_task 状态机逐章跳过已 done 的 pass；失败记录不阻断后续章。
+管线三步（顺序依赖）：
+  chapters  逐章一次 M3：细纲+实体+范式候选线索 → example_parse_scaffold（正文只过一遍）
+  （parse_outline window：窗级大纲聚合 → example_parse_outline，窗=出卡窗的 SoT）
+  cards     逐窗一次 M3：窗内线索+细纲+阶段大纲 → 聚类出母卡 → parse_ingest cards 守卫入库
+
+断点续跑：chapters 按 example_parse_task.scaffold_status 跳过；cards 按窗内是否已有活卡跳过。
 """
 import json
 import pathlib
@@ -27,7 +33,7 @@ TENANT = 1
 HERE = pathlib.Path(__file__).resolve().parent
 TMP = pathlib.Path("/tmp/muse-parse")
 
-# ── 提示词资产（自 workflow 试拆版移植，与 A3 库内合同同源） ──────────────
+# ── 提示词资产（合同一律 load_contracts 动态渲染，禁手写——CONTRACTS 漂移冤案教训） ──
 
 IDENTITY = """你是知识抽取员(extractor)，分析槽位的默认绑定件。产出全部是草稿。
 元数据纪律：schema 有什么字段你就抽什么，schema 没有的不抽——字段合同就是抽取 checklist，不自造结构；归型走各型「判据」；归不进任何型的候选=枚举缺口，如实报不硬塞；每字段要有正文证据，置信度低标「?」。
@@ -44,6 +50,9 @@ PATTERN_TYPES = ("craft", "combat", "emotion", "scene_pattern", "trope")
 # 基础公共字段（yudao 惯例列，出卡时由 ingest/payload 承载，不进抽取字段表）
 BASE_KEYS = {"名称", "别名", "一句话摘要", "标签", "来源", "状态", "例证出处"}
 
+# 抽象指代白名单（金标准结论：M3 自造代号如"A角色"破坏可读性；专名只允许进实例定位）
+PLACEHOLDERS = "主角/对手/强敌/导师/盟友/队友/配角/反派/长辈/宝物/装备/机关/势力/秘密/危机"
+
 
 def load_contracts(conn):
     """从库内 schema 版本快照动态渲染五型合同——prompt 与 ingest 守卫同源。
@@ -54,11 +63,12 @@ def load_contracts(conn):
     """
     contracts = {}
     for t in PATTERN_TYPES:
+        # 走 active_version_id（激活版本=治理权威），与 ingest 守卫同版本语义——
+        # 「最新版本行」在铺了新版未激活时会与守卫劈叉
         snap = conn.execute(
             """SELECT v.field_contract_snapshot FROM muse_meta_schema_version v
-               JOIN muse_meta_schema s ON s.id=v.schema_id
+               JOIN muse_meta_schema s ON s.active_version_id=v.id
-               WHERE s.tenant_id=%s AND s.schema_key=%s AND v.deleted=FALSE
+               WHERE s.tenant_id=%s AND s.schema_key=%s""", (TENANT, t)).fetchone()[0]
-               ORDER BY v.id DESC LIMIT 1""", (TENANT, t)).fetchone()[0]
         fields = [f for f in snap.get("特有字段", []) if f.get("key") not in BASE_KEYS]
         contracts[t] = {"中文名": snap.get("中文名", t), "判据": snap.get("判据", ""),
                         "字段": fields}
@@ -80,47 +90,55 @@ def scaffold_prompt(title, ch, ch_title, text, prev_entities):
     # 绝对字数上限比抽象比例对 LLM 更可执行（首轮实测 M3 按比例会写到 9.8%–19%）
     wc = len(text.replace("\n", "").replace(" ", ""))
     cap = max(60, int(wc * 0.05))
-    return f"""【功能指令(parse-book 逐章内环)】
+    return f"""【功能指令(parse-book 章级 pass：细纲+实体+范式候选线索)】
-对参考书《{title}》第 {ch} 章《{ch_title}》（正文附后，约 {wc} 字）：
+对参考书《{title}》第 {ch} 章《{ch_title}》（正文附后，约 {wc} 字）做三件事：
 1) 逆推本章细纲：章目标/关键事件/出场角色/伏笔动作(埋·推·收)/章末钩子。**细纲全文不得超过 {cap} 字**（章正文的 5%；超标会被校验脚本机械退回）。细纲是结构骨架，不是缩写复述——用短语与分号，不写完整句子。
 2) 抽实体增量（脚手架级索引）：{ENTITY_CRITERIA}
 判重：下方是前文已收录实体，不重报（除非本章给出新身份，则在一句话摘要里并入）。
+3) 报范式候选线索（**只报线索不出卡**，出卡由窗级聚类另做）：本章表现突出、疑似可跨书复用的写法。五型候选：craft=单点叙事装置(删去它场景仍成立)；combat=整场武力对抗的打法；emotion=整场情绪戏的推进；scene_pattern=拍卖/谈判/审讯等场景公式；trope=跨章复用的情节公式。
+每条线索给四项：type（五型之一，拿不准填"?"）、name（2–8 字短名，作者口头会说的话，禁书内专名禁修饰堆叠）、clue（一句话：这个写法怎么运作，用{PLACEHOLDERS}等抽象指代）、evidence（一句话：本章哪里这样写、为何突出，可用专名）。
+0–5 条/章，平庸章 0 条正常；门槛比出卡低——拿不准的报上来，窗级聚类会过滤。
 
 【前文实体索引】
 {ents}
 
 【输出规则（只输出一个 JSON 对象，禁止任何其他文字）】
-{{"outline": "细纲文本", "entities": [{{"type": "六型之一", "name": "名称", "brief": "一句话摘要"}}]}}
+{{"outline": "细纲文本", "entities": [{{"type": "六型之一", "name": "名称", "brief": "一句话摘要"}}], "hints": [{{"type": "五型之一或?", "name": "短名", "clue": "运作机制一句话", "evidence": "本章证据一句话"}}]}}
 
 【本章正文】
 {text}"""
 
 
-def patterns_prompt(title, ch, ch_title, text, outline, existing_cards, contracts):
+def window_cards_prompt(title, a, b, stage_outline, ch_outlines, ch_hints, contracts):
-    cards = json.dumps(existing_cards, ensure_ascii=False) if existing_cards else "（尚无）"
+    """窗级聚类出卡 prompt（B4-S3 核心资产；金标准七条纪律全落于此）。"""
-    return f"""【功能指令(parse-book 范式拆取+脱敏红线)】
+    return f"""【功能指令(parse-book 窗级聚类出卡+脱敏红线)】
-对参考书《{title}》第 {ch} 章《{ch_title}》（正文附后），在脚手架细纲之上拆「范式五型」卡：
+《{title}》第 {a}–{b} 章已完成章级分析。下面给你三层材料：阶段大纲、逐章细纲、逐章范式候选线索。
+任务：把候选线索**聚类归并**成「公共范式卡」——同一手法在多章的多次出现归并为一张母卡，带全部实例章号；线索不足处可从细纲的伏笔账（埋·推·收）补证跨章装置。
 
 {render_contracts(contracts)}
-纪律：
-- 宁缺毋滥：一章 0–3 张为常态，只收本章表现突出、可跨书复用的写法；平庸章可出 0 张。
-- 归型走判据；五型之外一律不出卡。**先按判据定 type，再只用该 type 字段表里的中文 key 填 fields——混用他型字段=机械拒卡**（如 trope 的「公式步骤」不得出现在 craft 卡里）。
-- **五型都是候选，不要把一切归成 craft**：整场武力对抗→combat；整场情绪戏→emotion；拍卖/谈判/审讯等场景公式→scene_pattern；跨章复用的情节公式→trope；craft 只留给「单点装置」（删去它场景仍成立）。type 值必须与 fields 所用字段表同型。
-- **脱敏红线**：只写抽象结构与手法归纳，严禁抄录原文；≥15 连续字与原文重合=校验脚本机械拒卡。卡名与字段值用「主角/强敌/导师」等抽象指代，**不得出现书内专名**（人名/地名/机甲名/组织名）——专名只允许出现在 source.anchor 里。
-- fields 按该型字段合同的中文 key 填；没证据的 key 省略，不编造。
-- 判重：下方已积累卡名录，不与之重复立同义卡。
 
-【本章细纲(脚手架 pass 产出)】
+出卡纪律：
-{outline}
+- 聚类优先：先把线索按「同一运作机制」分组（不同章的同类线索=同一卡的多个实例），再逐组判断是否值得出卡。一窗 0–6 张为常态，宁缺毋滥——聚不成组又不够突出的线索直接丢弃。
+- 跨章证据：结构装置（伏笔类）尽量给出埋与收两端的实例章号；只有单章实例的会被标「跨窗待证」降低可信度。
+- 命名：卡名=作者口头会说的话（2–8 字，如「先抑后扬」「借刀杀人」），禁「XX式YY化ZZ」修饰语堆叠、禁书内专名、禁自造黑话。
+- 摘要：一句话**通用手法陈述**（抹掉本书信息后换任何题材仍成立），禁复述本书剧情、禁与实例定位雷同。
+- 抽象指代白名单：{PLACEHOLDERS}——白名单外的自造代号（如"A角色""X道具"）不要用。
+- 可迁移性是卡的本体：说不清「作者这样做、读者会怎样」的因果=不是范式，整张卡不出；带此类字段（如原理）的型按其字段说明的句式写——**句式里的 X/Y/Z 是占位符，必须替换成本卡的具体内容**，字面保留「做Y」「读者会Z」=机械拒卡。
+- 实例只报章号+一句话定位；**严禁在任何字段里自报间隔章数**——间隔由校验脚本按实例章号差机械计算，自己编数字=伪精确。
+- fields 只允许使用该卡 type 在上方合同表里列出的中文 key，没证据的 key 省略不编造。**先按判据定 type，再对照该型的表格填字段**——把别的型才有的 key（哪怕内容再有价值）塞进本型=机械拒卡；上一批实测拒卡主因就是给非 craft 卡塞了 craft 才有的字段。
+- 脱敏红线：只写抽象结构与手法归纳，严禁抄录原文（≥15 连续字与原文重合=机械拒卡）；书内专名只允许出现在实例定位(anchor)里。
 
-【已积累范式卡名录】
+【阶段大纲】
-{cards}
+{stage_outline}
+
+【逐章细纲】
+{ch_outlines}
+
+【逐章候选线索】
+{ch_hints}
 
 【输出规则（只输出一个 JSON 对象，禁止任何其他文字；没有值得立的卡时 cards 给空数组）】
-{{"cards": [{{"type": "craft|combat|emotion|scene_pattern|trope", "name": "抽象手法名(无书内专名)", "brief": "一句话摘要", "fields": {{"中文合同key": "值"}}, "source": {{"book": "{title}", "chapter": "第{ch}章 {ch_title}", "anchor": "一句话情节定位(抽象指代)"}}}}]}}
+{{"cards": [{{"type": "craft|combat|emotion|scene_pattern|trope", "name": "短名", "brief": "一句话通用摘要", "fields": {{"中文合同key": "值"}}, "instances": [{{"ch": 章号数字, "anchor": "一句话情节定位(可用专名)"}}]}}]}}"""
-
-【本章正文】
-{text}"""
 
 
 def m3_json(prompt, model, need_keys, system=IDENTITY):
@@ -141,35 +159,40 @@ def m3_json(prompt, model, need_keys, system=IDENTITY):
     raise RuntimeError(f"JSON 形状重试仍失败: {err}")
 
 
-def ingest(kind, work_id, ch, payload):
+def ingest(kind, work_id, key, payload, keyflag="--chapter-order"):
     """写临时文件 → parse_ingest 机械校验入库；返回 (是否成功, 输出文本)。"""
     TMP.mkdir(parents=True, exist_ok=True)
-    f = TMP / f"{work_id}-{ch}-{kind}.json"
+    f = TMP / f"{work_id}-{key}-{kind}.json"
     f.write_text(json.dumps(payload, ensure_ascii=False, indent=1))
     r = subprocess.run([sys.executable, str(HERE / "parse_ingest.py"), kind,
-                        "--work-id", str(work_id), "--chapter-order", str(ch), "--file", str(f)],
+                        "--work-id", str(work_id), keyflag, str(key), "--file", str(f)],
                        capture_output=True, text=True)
     return r.returncode == 0, (r.stdout + r.stderr).strip()
 
 
-@click.command()
+@click.group()
+def cli():
+    """M3 直调拆书（章级线索 → 窗级出卡）"""
+
+
+@cli.command()
 @click.option("--work-id", type=int, required=True)
 @click.option("--from", "from_", type=int, required=True)
 @click.option("--to", type=int, required=True)
 @click.option("--model", default="MiniMax-M3", show_default=True)
-def main(work_id, from_, to, model):
+def chapters(work_id, from_, to, model):
+    """章级 pass：逐章一次 M3（细纲+实体+范式候选线索）。正文只过这一遍。"""
     # 任务行就绪（幂等）
     subprocess.run([sys.executable, str(HERE / "parse_ingest.py"), "init-tasks",
                     "--work-id", str(work_id), "--from", str(from_), "--to", str(to)],
                    capture_output=True, text=True)
     with psycopg.connect(DSN) as conn:
         title = conn.execute("SELECT title FROM muse_content_work WHERE id=%s", (work_id,)).fetchone()[0]
-        contracts = load_contracts(conn)  # prompt 与 ingest 守卫同源（库内 schema 快照）
     total_in = total_out = 0
     for ch in range(from_, to + 1):
         with psycopg.connect(DSN) as conn:
             row = conn.execute(
-                """SELECT c.id, c.title, b.content_text, t.scaffold_status, t.pattern_status
+                """SELECT c.id, c.title, b.content_text, t.scaffold_status
                    FROM muse_content_chapter c
                    JOIN muse_content_block b ON b.chapter_id=c.id AND b.deleted=FALSE
                    JOIN example_parse_task t ON t.chapter_id=c.id AND t.tenant_id=c.tenant_id
@@ -178,32 +201,19 @@ def main(work_id, from_, to, model):
             if not row:
                 click.echo(f"#{ch} 章或任务行不存在，跳过")
                 continue
-            ch_id, ch_title, text, s_st, p_st = row
+            ch_id, ch_title, text, s_st = row
+            if s_st == "done":
+                click.echo(f"#{ch} 脚手架已完成，跳过")
+                continue
             # 前文实体索引（紧凑：型/名称/摘要），判重用
             prev = [e for (ents,) in conn.execute(
                 """SELECT s.entities FROM example_parse_scaffold s
                    JOIN muse_content_chapter c ON c.id=s.chapter_id
                    WHERE s.tenant_id=%s AND s.work_id=%s AND c.order_no<%s AND s.deleted=FALSE""",
                 (TENANT, work_id, ch)).fetchall() for e in ents]
-            # 判重名录排除本章自己的卡：否则重跑时 M3 对着自己首轮的卡判重不出卡，
-            # ingest 幂等软删旧卡后插 0 张——卡被自己删了（重跑自噬,实测病）
-            cards = conn.execute(
-                """SELECT draft_payload->>'型', draft_payload->>'名称' FROM muse_knowledge_draft
-                   WHERE tenant_id=%s AND source_type='parse_book' AND deleted=FALSE
-                     AND NOT (source_id=%s AND (draft_payload->>'章序')::int=%s)""",
-                (TENANT, work_id, ch)).fetchall()
-            outline = None
-            if s_st == "done":
-                r = conn.execute(
-                    "SELECT outline_text FROM example_parse_scaffold WHERE tenant_id=%s AND chapter_id=%s AND deleted=FALSE",
-                    (TENANT, ch_id)).fetchone()
-                outline = r[0] if r else None
-
-        # pass1 脚手架（断点续跑：done 跳过）
-        if s_st != "done" or outline is None:
         try:
             p1 = scaffold_prompt(title, ch, ch_title, text, prev)
-                data, usage = m3_json(p1, model, ("outline", "entities"))
+            data, usage = m3_json(p1, model, ("outline", "entities", "hints"))
             total_in += usage.get("prompt_tokens", 0)
             total_out += usage.get("completion_tokens", 0)
             ok, out = ingest("scaffold", work_id, ch, data)
@@ -213,42 +223,96 @@ def main(work_id, from_, to, model):
                 data, usage = m3_json(
                     p1 + f"\n\n【重试】上次细纲 {len(data['outline'])} 字超标被退回。"
                          f"压缩到 {cap} 字以内：只留章目标/关键事件/伏笔动作/钩子，删掉一切修饰与过程描述。",
-                        model, ("outline", "entities"))
+                    model, ("outline", "entities", "hints"))
                 total_in += usage.get("prompt_tokens", 0)
                 total_out += usage.get("completion_tokens", 0)
                 ok, out = ingest("scaffold", work_id, ch, data)
             click.echo(f"  {out}")
-                if not ok:
-                    continue  # 已记 task failed，重跑本命令补
-                outline = data["outline"]
         except RuntimeError as e:
-                click.echo(f"  #{ch} 脚手架 M3 失败: {e}")
+            click.echo(f"  #{ch} 章级 M3 失败: {e}")
+    click.echo(f"《{title}》{from_}–{to} 章级完成；token in={total_in:,} out={total_out:,}")
+
+
+@cli.command()
+@click.option("--work-id", type=int, required=True)
+@click.option("--from-order", "from_order", type=int, help="只出该窗（窗起始章）；不给则全部窗")
+@click.option("--model", default="MiniMax-M3", show_default=True)
+@click.option("--redo", is_flag=True, help="窗内已有活卡也重出（默认跳过=断点续跑）")
+def cards(work_id, from_order, model, redo):
+    """窗级出卡：逐窗一次 M3 聚类归并（窗=example_parse_outline 行，先跑 parse_outline window）。"""
+    with psycopg.connect(DSN) as conn:
+        title = conn.execute("SELECT title FROM muse_content_work WHERE id=%s", (work_id,)).fetchone()[0]
+        contracts = load_contracts(conn)  # prompt 与 ingest 守卫同源（库内 schema 快照）
+        sql = """SELECT from_order, to_order, outline_text, window_no FROM example_parse_outline
+                 WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE"""
+        args = [TENANT, work_id]
+        if from_order:
+            sql += " AND from_order=%s"
+            args.append(from_order)
+        wins = conn.execute(sql + " ORDER BY from_order", args).fetchall()
+    if not wins:
+        click.echo("无大纲窗行——先跑 parse_outline.py window（窗是出卡的切分依据）")
+        return
+    total_in = total_out = 0
+    for a, b, stage_ol, wno in wins:
+        with psycopg.connect(DSN) as conn:
+            # 断点续跑：窗内已有活卡（本窗出的）则跳过
+            if not redo and conn.execute(
+                    """SELECT 1 FROM muse_knowledge_draft WHERE tenant_id=%s AND source_type='parse_book'
+                       AND source_id=%s AND draft_payload->>'窗起'=%s AND deleted=FALSE LIMIT 1""",
+                    (TENANT, work_id, str(a))).fetchone():
+                click.echo(f"win#{a}–{b} 已有活卡，跳过（--redo 强制重出）")
                 continue
-        # pass2 范式（done 跳过；卡被守卫拒属正常，不算失败）
+            rows = conn.execute(
-        if p_st != "done":
+                """SELECT c.order_no, c.title, s.outline_text, s.pattern_hints
+                   FROM muse_content_chapter c
+                   JOIN example_parse_scaffold s ON s.chapter_id=c.id AND s.deleted=FALSE
+                   WHERE c.tenant_id=%s AND c.work_id=%s AND c.order_no BETWEEN %s AND %s
+                     AND c.deleted=FALSE ORDER BY c.order_no""",
+                (TENANT, work_id, a, b)).fetchall()
+        if len(rows) < (b - a + 1):
+            have = {r[0] for r in rows}
+            click.echo(f"win#{a}–{b} 缺章级脚手架 {[n for n in range(a, b + 1) if n not in have][:10]}，先补 chapters")
+            continue
+        ch_outlines = "\n".join(f"第{no}章《{ct}》：{o}" for no, ct, o, _ in rows)
+        hints = [{"章": no, "线索": h} for no, _, _, hs in rows for h in (hs or [])]
+        n_hint = len(hints)
+        ch_hints = (json.dumps(hints, ensure_ascii=False, indent=1) if hints
+                    else "（各章均未报线索——只依据细纲伏笔账判断，出不了卡就给空数组）")
         try:
+            p = window_cards_prompt(title, a, b, stage_ol, ch_outlines, ch_hints, contracts)
+            data, usage = m3_json(p, model, ("cards",))
+            total_in += usage.get("prompt_tokens", 0)
+            total_out += usage.get("completion_tokens", 0)
+            # 0 卡可疑重试（实测病：M3 输出方差大，同材料一次 10 卡一次秒回空数组）：
+            # 线索 ≥3 条却 0 卡=大概率分叉到偷懒路径，重试一次；仍 0 卡则接受（真无卡）
+            if not (data.get("cards") or []) and n_hint >= 3:
                 data, usage = m3_json(
-                    patterns_prompt(title, ch, ch_title, text, outline, cards, contracts),
+                    p + "\n\n【重试】上次输出了空数组，但本窗有 %d 条候选线索——请认真逐条聚类评估后再输出；"
+                        "确实全部不够格才允许空数组。" % n_hint, model, ("cards",))
+                total_in += usage.get("prompt_tokens", 0)
+                total_out += usage.get("completion_tokens", 0)
+            _, out = ingest("cards", work_id, a, data, keyflag="--from-order")
+            # 拒卡带原因重试一次：拒卡原因机械明确（占位符/越合同），带回 prompt 大概率可救；
+            # 放量时每窗调用贵，救回率优先于省一次调用
+            rej_lines = [ln for ln in out.splitlines() if ln.strip().startswith("[拒]")]
+            if rej_lines:
+                data, usage = m3_json(
+                    p + "\n\n【重试】上次输出有以下卡被机械校验拒收，逐条按原因修复后**重新输出全部卡**"
+                        "（被拒的修好，没被拒的原样保留）：\n" + "\n".join(rej_lines),
                     model, ("cards",))
                 total_in += usage.get("prompt_tokens", 0)
                 total_out += usage.get("completion_tokens", 0)
-                # source 机械回填：book/chapter 是调用侧确定数据，不依赖 LLM 自觉
+                _, out = ingest("cards", work_id, a, data, keyflag="--from-order")
-                #（实测 M3 偶发 source=null 整批被拒）；anchor 是内容性字段，缺时用 brief 兜底
+            click.echo(f"win#{a}–{b}（线索 {n_hint} 条）\n  {out}")
-                for c in data.get("cards") or []:
-                    src = c.get("source") or {}
-                    c["source"] = {"book": src.get("book") or title,
-                                   "chapter": src.get("chapter") or f"第{ch}章 {ch_title}",
-                                   "anchor": src.get("anchor") or c.get("brief") or ""}
-                _, out = ingest("patterns", work_id, ch, data)
-                click.echo(f"  {out}")
         except RuntimeError as e:
-                click.echo(f"  #{ch} 范式 M3 失败: {e}")
+            click.echo(f"win#{a}–{b} 窗级 M3 失败: {e}")
-    click.echo(f"《{title}》{from_}–{to} 章完成；token in={total_in:,} out={total_out:,}")
+    click.echo(f"《{title}》窗级出卡完成；token in={total_in:,} out={total_out:,}")
 
 
 if __name__ == "__main__":
     try:
-        main()
+        cli()
     except (psycopg.Error, RuntimeError) as e:
         click.echo(f"[错误] {type(e).__name__}: {e}", err=True)
         sys.exit(1)

.claude/skills/parse-book/scripts/parse_outline.py

@@ -77,16 +77,22 @@ def cli():
 @click.option("--window", type=int, default=80000, show_default=True, help="每窗目标字数（5–10万区间）")
 @click.option("--model", default="MiniMax-M3", show_default=True)
 @click.option("--from", "from_", type=int, default=1, help="起始章（断点续跑给上次结束章+1）")
-def window(work_id, window, model, from_):
+@click.option("--to", "to_", type=int, help="结束章（限定已拆域，防止切窗吞进没有细纲的章；默认全书）")
+def window(work_id, window, model, from_, to_):
     """按字数切窗聚合大纲（窗内章须已有细纲；已有窗大纲的窗跳过=断点续跑）。"""
     with psycopg.connect(DSN) as conn:
         title = conn.execute("SELECT title FROM muse_content_work WHERE id=%s", (work_id,)).fetchone()[0]
-        rows = [r for r in load_chapters(conn, work_id) if r[0] >= from_]
+        all_rows = load_chapters(conn, work_id)
-        done = {r[0] for r in conn.execute(
+        last_order = all_rows[-1][0] if all_rows else 0  # 全书末章（书末残窗判断用）
-            "SELECT window_no FROM example_parse_outline WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE",
+        rows = [r for r in all_rows if r[0] >= from_ and (to_ is None or r[0] <= to_)]
-            (TENANT, work_id)).fetchall()}
+        # 幂等键=from_order（内容锚定，B4-S3/fable 审查 E2）：window_no 是流水号，
-        # 章序切窗（与清洗窗同思路：章对齐）
+        # 断点续跑或参数变化时会漂移导致覆盖错窗——只作展示序号，不作跳过依据
-        win_no, cur, cur_len = max(done, default=0), [], 0
+        prior = conn.execute(
+            "SELECT from_order, window_no FROM example_parse_outline WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE",
+            (TENANT, work_id)).fetchall()
+        done = {r[0] for r in prior}
+        # 章序切窗（与清洗窗同思路：章对齐）；全量重跑序号从头计，续跑接上次
+        win_no, cur, cur_len = (0 if from_ <= 1 else max((r[1] for r in prior), default=0)), [], 0
         for no, ct, text, outline in rows:
             cur.append((no, ct, text, outline))
             cur_len += len(text)
@@ -94,16 +100,18 @@ def window(work_id, window, model, from_):
                 win_no += 1
                 _do_window(conn, work_id, title, win_no, cur, model, done)
                 cur, cur_len = [], 0
-        if cur and cur_len >= window * 0.4:  # 尾窗过小并入上窗的策略：足够大才单独成窗
+        # 尾段：已到全书末章 → 无论大小独立成窗（全书拆完时尾段必须落窗，否则永远留白）；
+        # 未到书末 → 足够大才成窗，不足留待与后续章合窗
+        if cur and (cur[-1][0] == last_order or cur_len >= window * 0.4):
             win_no += 1
             _do_window(conn, work_id, title, win_no, cur, model, done)
         elif cur:
-            click.echo(f"尾段 {len(cur)} 章（{cur_len:,} 字）不足半窗，留待与后续章合窗")
+            click.echo(f"尾段 {len(cur)} 章（{cur_len:,} 字）不足半窗且未到书末，留待与后续章合窗")
 
 
 def _do_window(conn, work_id, title, win_no, chs, model, done):
-    if win_no in done:
+    if chs[0][0] in done:  # 幂等按窗起始章判断（不按流水号）
-        click.echo(f"win-{win_no:02d} 已有大纲，跳过")
+        click.echo(f"win-{win_no:02d}（第{chs[0][0]}章起）已有大纲，跳过")
         return
     missing = [no for no, _, _, o in chs if not o]
     if missing:
@@ -124,8 +132,8 @@ def _do_window(conn, work_id, title, win_no, chs, model, done):
         """INSERT INTO example_parse_outline (work_id, window_no, from_order, to_order, outline_text,
                creator, updater, tenant_id)
            VALUES (%s,%s,%s,%s,%s,%s,%s,%s)
-           ON CONFLICT (tenant_id, work_id, window_no)
+           ON CONFLICT (tenant_id, work_id, from_order)
-           DO UPDATE SET outline_text=EXCLUDED.outline_text, from_order=EXCLUDED.from_order,
+           DO UPDATE SET outline_text=EXCLUDED.outline_text, window_no=EXCLUDED.window_no,
                          to_order=EXCLUDED.to_order, deleted=FALSE, check_status='pending', updater=EXCLUDED.updater""",
         (work_id, win_no, chs[0][0], chs[-1][0], ol, ACTOR, ACTOR, TENANT))
     conn.commit()

.claude/skills/review-cards/scripts/review_cards.py

@@ -37,9 +37,11 @@ CRITERIA = """审查判定要点（每张卡都过一遍）：
 2. 命名黑话检测：卡名是"作者会说的话"还是"XX式YY化ZZ"的 AI 修饰语堆叠？摘要是通用手法陈述还是原书剧情复述？
 3. 标签判定：装置类型是否乱贴（信息差=必须存在角色A知道而B/读者不知道的落差；契诃夫之枪=预先可见放置+后续引爆；身份错认=必须有人把A当成B）；
 4. 伪精确检测："隔两章以上"这类数字是从原文数出来的还是编的？编的=危险的 false precision；
+   例外：带「实测：#a埋→#b收，隔n章」标注的数字是校验脚本按实例章号机械计算的权威值，可信；
 5. 同功重复：与本批其他卡是否同一手法换皮（先知降压/面瘫外化这类 family）；
 6. 干货位检查：失败模式字段是否比手法本身更值钱（是→说明手法总结空洞）；
 7. 专名泄漏：字段里出现原书人名/地名/数字/梗=破坏跨书复用；
+   豁免区：「实例」定位与「出处」里的专名是设计允许的（溯源用），不算泄漏——只审名称/摘要/字段；
 8. 场景走位判定：间隔纪律/复用节奏写明「埋收同场景闭环」的，不是跨章装置而是场景内走位，
    可参考性上限 2 分（反例：某"借道具看直播"卡把单场走位包装成范式——金标准 2.67 分）。"""
 

README.md

@@ -230,9 +230,10 @@ flowchart LR
 - **B2 试拆收口（2026-07-13，门②已呈报）**：5 本×前 3 章 M3 直调完成——scaffold 15/15 全过（绝对字数上限+压缩重试对症后，细纲 2.9%–7.8%）；范式卡三轮迭代后活卡 19 张（样张 `docs/试拆-门②样张.md`，opus 同章基准 `docs/试拆-opus基准-机动风暴.md`）。**M3 三病理已档**：①归型偏科——内容有型区分但 type 一律标 craft，prompt 两轮矫正无效，字段指纹改型又被「串型混填」挫败（一卡混两型字段）；②重跑自噬——判重名录含本章旧卡+幂等软删+LLM 随机性组合损耗（已修：名录排除本章+0 卡不软删）；③偶发脱敏违规/source 偷懒（守卫拦对+机械回填治）。
 - **拆书流程增补（2026-07-13 创始人拍板+落地）**：①**窗级大纲聚合**——大纲不从单章抽（单章对大纲层可能零贡献），5–10 万字窗（多章细纲+正文）聚合一次，整本完后拿全体细纲终检（parse_outline.py + 93 表）；②**三角色审核常设步骤**——番茄作家/起点作家/主编审核公共卡（成立性/AI 可用性/可参考性），**M3 执行**（review-cards skill）；fable/opus 只做起量前校准+起量后总审核。校准已过：opus 三评审金标准入 golden/，M3 对照平均绝对偏差 0.45（合格线 0.5），四大系统病独立复现。首批 19 卡 M3 判定 pass 5 / revise 10 / reject 4。
 - **金标准评审结论（拆书 prompt 的 B4 改造清单,见 review-cards/SKILL.md）**：同功 family 撞车（19 卡实为 10–12 个独立技法）、命名黑话+专名泄漏、伪精确数字、伏笔框架硬套场景手法、失败模式才是干货位、采样偏科开局——七条改进已档,B4 环实施。
-- **B4 收敛环（2026-07-13 创始人拍板「先收敛质量」,进行中）**：fable 独立审查完成（报告=`docs/2026-07-13-B4-fable独立审查报告.md`,S1-S6 实施序）。**头号发现=CONTRACTS 漂移冤案**：parse_llm 手写合同与库内字段名漂移,M3 正确产出的 11 张非 craft 卡全被守卫冤杀——「craft 偏科」一半是冤案。**已完成**：S1 修地基（合同库内动态渲染 prompt 与守卫同源+system 角色分离+审核 idx 匹配;验证=重跑 15 章拒卡 14→0、scene_pattern 首次存活 8 张）;S2 schema 批（craft 双模板「装置形态」条件必填禁填+装置类型闭合枚举判定测试+原理/触发条件/读者收益/回收点/记忆维持/滥用反例/迁移用例新字段+emotion 目标情绪+**pacing 启用 8 字段**「黄金三章画像/开局策略/上架节奏切换/张力曲线/爽点间隔…」=创始人抽取分层议题的书级归属;ingest 三守卫已验）;S4 审核环（CRITERIA#8 场景走位+机械降档回测零误伤+分页 ≤30）。**待做**：S3 管线重构（scaffold 扩候选 hints+窗级聚类出卡+embedding 判重+间隔机械计算——出卡权上移到判据能证成的窗口）→S5 参数 A/B（范式线 temp 0.2vs1.0）→复跑校准。
+- **B4 收敛环（2026-07-13 创始人拍板「先收敛质量」,S1-S4 全部完成）**：fable 独立审查完成（报告=`docs/2026-07-13-B4-fable独立审查报告.md`,S1-S6 实施序）。**头号发现=CONTRACTS 漂移冤案**：parse_llm 手写合同与库内字段名漂移,M3 正确产出的 11 张非 craft 卡全被守卫冤杀——「craft 偏科」一半是冤案。**已完成**：S1 修地基（合同库内动态渲染 prompt 与守卫同源+system 角色分离+审核 idx 匹配;验证=重跑 15 章拒卡 14→0、scene_pattern 首次存活 8 张）;S2 schema 批（craft 双模板「装置形态」条件必填禁填+装置类型闭合枚举判定测试+原理/触发条件/读者收益/回收点/记忆维持/滥用反例/迁移用例新字段+emotion 目标情绪+**pacing 启用 8 字段**「黄金三章画像/开局策略/上架节奏切换/张力曲线/爽点间隔…」=创始人抽取分层议题的书级归属;ingest 三守卫已验）;S4 审核环（CRITERIA#8 场景走位+机械降档回测零误伤+分页 ≤30+豁免区说明防假阳）;**S3 管线重构（本轮完成并验收）**——见下条。**待做**：S5 参数 A/B（范式线 temp 0.2vs1.0;M3 输出方差的根治口）→S6 pacing 书级抽取（等整本拆完）。
+- **S3 管线重构收口（2026-07-13,验收报告=`docs/2026-07-13-S3窗级出卡验收报告.md`）**：出卡权上移窗级——章级只产「范式候选线索」（并入 scaffold pass,正文只过一遍,放量每书省 ~8M 输入 token）,窗级（=大纲窗行,幂等键改 from_order+书末残窗必成窗）聚类归并出母卡+多实例章号,**间隔章数由实例章号差机械计算**（伪精确灭绝）,入库前嵌入判重（≥0.85→M3 归并终判,0.75-0.85 标记）。**验收（5 书×3 章重拆对比）**：型偏科痊愈（craft 27/32→14/39,trope 0→15）;命名全短名;摘要通用化;间隔全机械溯源;跨书近亲 3 对被判重标记;审核 pass 11/39,top 4.33×2 超旧金标准最高 4.0。**本轮实战修四病**（各带机械防线,见验收报告）：纪律措辞串型→裁剪降级;占位符字面抄写→机械检测+拒卡重试环;M3 输出方差（同料 0 卡vs10 卡）→0 卡可疑重试;审核豁免区假阳→CRITERIA 补说明。管线定版=五步（chapters→window→cards→check→review,SKILL.md §放量全流程）。
 - **抽取窗口×型映射（创始人议题定版）**：单章=6 实体型+范式候选 hint;窗级（5-10万字）=五型出卡聚类+trope+character_relation;书级=pacing/style/终检。原理:**抽取窗口必须 ≥ 该型判据的证成窗口**（trope 判据是跨章公式却曾在单章抽=永远证不成）。
-- **未决**：①B2 放量（硬停,等 B4 收敛完+创始人指令）②机动残留变体与感言章处理方式③S5 温度 A/B 依赖 S3。
+- **未决**：①B2 放量（硬停,等创始人指令;B4 质量收敛主体已完成,S5 参数实验可与放量并行或先行,由创始人定）②机动残留变体与感言章处理方式③跨书同功母卡归并（「设定投喂」家族 3 书各一张,相似度 0.6-0.7 不到预警线）=放量后公共面课题。
 - **教训入档**：StructuredOutput 工具 schema 属性名仅限 ASCII（中文键 API 400）——schema ASCII 键+入库脚本键名归一；Tailscale 长事务需 keepalive+批量写（逐行两万往返曾半死 16 分钟）。
 
 ## 九·旧（2026-07-10 快照，留档）

db/ddl/91-example实验私货.sql

@@ -58,6 +58,8 @@ CREATE TABLE example_parse_scaffold (
     chapter_id BIGINT NOT NULL,                       -- → muse_content_chapter.id
     outline_text TEXT,                                -- 逆推细纲（对原文 3–5% 字数）
     entities JSONB,                                   -- 实体清单 [{type,name,brief}]
+    -- B4-S3（2026-07-13）：出卡权上移窗级——章级只产「范式候选线索」，窗级聚类归并出卡
+    pattern_hints JSONB NOT NULL DEFAULT '[]',        -- 候选线索 [{type,name,clue,evidence}]
     creator VARCHAR(64) NOT NULL DEFAULT '',
     create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
     updater VARCHAR(64) NOT NULL DEFAULT '',

db/ddl/93-example大纲聚合.sql

@@ -17,7 +17,9 @@ CREATE TABLE IF NOT EXISTS example_parse_outline (
     update_time  TIMESTAMP    NOT NULL DEFAULT now(),
     deleted      BOOLEAN      NOT NULL DEFAULT FALSE,
     tenant_id    BIGINT       NOT NULL DEFAULT 0,
-    CONSTRAINT uk_example_parse_outline UNIQUE (tenant_id, work_id, window_no)
+    -- B4-S3（fable 审查 E2）：幂等键用 from_order（内容锚定）而非 window_no（流水号，
+    -- 断点续跑/参数变化时会漂移导致覆盖错窗）；window_no 只作展示序号
+    CONSTRAINT uk_example_parse_outline UNIQUE (tenant_id, work_id, from_order)
 );
 
 COMMENT ON TABLE example_parse_outline IS 'example 私货：拆书窗级大纲聚合（5–10万字/窗，细纲+正文→大纲；终检=全体细纲对账）';