框架: 升格串行铁律——失败当场重试不跳窗+迟到覆写闸+缺键宽容

创始人拍板:窗串行生长,失败不得跳窗后补(断层+覆盖风险)。
- run 窗循环改 while:失败当场撤销半写入同窗重试一次,仍败停书(断点=本窗)
- merge_card 加水位闸:win_no<卡水位的覆写只留审计不动卡体(防补跑时间倒流)
- 水位更新改 GREATEST 防倒退
- 更新批缺'更新'键宽容为空批(prompt 明确空批输出+代码兜底,6窗实测死因)

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

@@ -200,7 +200,7 @@ def update_prompt(contracts, title, a, b, text, cards_with_obs):
 【相关型字段合同】
 {render_entity_contracts(contracts, sorted({p.get("type") for _, p, _ in cards_with_obs} & set(ENTITY_TYPES)))}
 
-【输出规则（只输出一个 JSON 对象）】
+【输出规则（只输出一个 JSON 对象；即使本批所有卡都无变化，也必须输出 {{"更新": []}}，不得省略"更新"键）】
 {{"更新": [{{"draft_id": 数字, "变更字段": {{"字段key": "新值或新增条目数组"}}, "别名新增": []}}]}}
 
 ━━━ 本窗材料 ━━━
@@ -272,6 +272,12 @@ def merge_card(conn, draft_id, win_no, changes, alias_new, valid_keys=None):
     payload = conn.execute("SELECT draft_payload FROM muse_knowledge_draft WHERE id=%s",
                            (draft_id,)).fetchone()[0]
     fields = payload.setdefault("字段", {})
+    # 卡水位（该卡最后一次被更新的窗号）：补跑迟到窗（如窗41在窗80后补跑）的覆写类字段
+    # 若直接落卡，会把书末态倒写回中期态（时间倒流污染，2026-07-15 补跑实测坐实）。
+    # 闸门：win_no < 水位 ⇒ 覆写只留审计不动卡体；追加类带窗号标签乱序无害，不拦。
+    wm_row = conn.execute("SELECT watermark_window FROM example_upgrade_card_state WHERE draft_id=%s",
+                          (draft_id,)).fetchone()
+    watermark = wm_row[0] if wm_row else 0
     changes = dict(changes or {})
     # 模型偶把「别名新增」混进变更字段（窗2实测）：摘出来并入别名流程，不落卡体字段
     alias_new = list(alias_new or []) + \
@@ -301,6 +307,15 @@ def merge_card(conn, draft_id, win_no, changes, alias_new, valid_keys=None):
                     old.append(f"[窗{win_no}] {core}")
                     seen.add(core)
             fields[k] = old
+        elif win_no < watermark:
+            # 迟到覆写弃用：本窗时序早于卡已生长到的窗位，覆写会让卡态倒流——
+            # 只留审计（标记可查），卡体保持高窗态不动
+            conn.execute(
+                """INSERT INTO example_upgrade_audit
+                       (draft_id, window_no, field_name, old_value, new_value, tenant_id)
+                   VALUES (%s,%s,%s,NULL,%s,%s)""",
+                (draft_id, win_no, ("迟到覆写弃用:" + str(k))[:100],
+                 json.dumps(v, ensure_ascii=False)[:2000], TENANT))
         else:
             conn.execute(
                 """INSERT INTO example_upgrade_audit
@@ -329,7 +344,9 @@ def merge_card(conn, draft_id, win_no, changes, alias_new, valid_keys=None):
     conn.execute(
         """INSERT INTO example_upgrade_card_state (draft_id, work_id, watermark_window, tenant_id)
            VALUES (%s,%s,%s,%s)
-           ON CONFLICT (draft_id) DO UPDATE SET watermark_window=EXCLUDED.watermark_window,
+           ON CONFLICT (draft_id) DO UPDATE
+           SET watermark_window=GREATEST(example_upgrade_card_state.watermark_window,
+                                         EXCLUDED.watermark_window),
                update_time=now()""",
         (draft_id, payload.get("_work_id") or 0, win_no, TENANT))
 
@@ -478,8 +495,14 @@ def run(work_id, max_windows, max_calls, model, redo_window):
                 click.echo(f"[撤销] failed 窗{r0[0]} 旧写入回滚后重跑")
                 undo_window(conn, work_id, r0[0])
     done_n = 0
-    fail_streak = 0   # 连续失败计数：单窗偶发失败（网络瞬断）continue，连续 2 窗停书（系统性问题）
+    # 串行铁律（创始人 2026-07-15 拍板）：窗与窗是串行生长——后窗的预扫/判重/更新
-    for win_no, a, b in wins:
+    # 全依赖前窗长成的卡。窗失败绝不跳窗（跳窗=知识断层+事后补跑有覆盖风险），
+    # 而是当场撤销半写入→整窗重试一次（挡偶发病）；仍败→停书，断点就在本窗，
+    # 下次启动从本窗续跑（failed 先撤销机制），串行语义天然无损。
+    wi = 0
+    retried = False   # 当前窗是否已当场重试过
+    while wi < len(wins):
+        win_no, a, b = wins[wi]
         if max_windows and done_n >= max_windows:
             break
         if max_calls and calls["n"] >= max_calls:
@@ -552,7 +575,16 @@ def run(work_id, max_windows, max_calls, model, redo_window):
                         p = conn.execute("SELECT draft_payload FROM muse_knowledge_draft WHERE id=%s",
                                          (did,)).fetchone()[0]
                         cards.append((did, p, obs_pts))
-                    upd, _ = call(update_prompt(contracts, title, a, b, text, cards), ("更新",))
+                    # 缺"更新"键宽容为空批：prompt 教"无变化的卡不输出"，某批恰好全无变化时
+                    # 模型会顺势连键一起省（批7实测 6 窗全死于此）。语义上缺键≈空批，按空批放行
+                    # 并留警告日志可审计；其余格式错误（乱码/解析失败）仍原样抛、窗照 fail。
+                    try:
+                        upd, _ = call(update_prompt(contracts, title, a, b, text, cards), ("更新",))
+                    except RuntimeError as e:
+                        if "缺少必需键" not in str(e):
+                            raise
+                        print(f"[宽容] 窗{win_no} 更新批缺键按空批放行: {str(e)[:80]}", file=sys.stderr)
+                        upd = {"更新": []}
                     valid = {d for d, _, _ in cards}
                     # 每卡按其型的合同 key 集校验（+一句话摘要），越合同 key 裁剪留审计
                     did2keys = {d: {f["key"] for f in contracts.get(p.get("type"), {}).get("字段", [])}
@@ -665,7 +697,8 @@ def run(work_id, max_windows, max_calls, model, redo_window):
                 click.echo(f"  窗{win_no}✓ ({a}-{b}章) 在场{len(onstage)} 新名字{new_n} "
                            f"更新卡{len(to_update)} 调用累计{calls['n']}")
                 done_n += 1
-                fail_streak = 0
+                wi += 1
+                retried = False
         except SensitiveHardStop as e:
             with psycopg.connect(DSN) as conn:
                 conn.execute(
@@ -682,14 +715,17 @@ def run(work_id, max_windows, max_calls, model, redo_window):
                        WHERE tenant_id=%s AND work_id=%s AND window_no=%s""",
                     (str(e)[:500], TENANT, work_id, win_no))
                 conn.commit()
-            fail_streak += 1
+            if not retried:
-            # 偶发失败（网络瞬断/单窗输出畸形）跳去下一窗，failed 窗留给续跑（自动先撤销）；
+                # 串行铁律第一层：当场撤销半写入→同窗立即重试（挡网络瞬断/偶发格式病）
-            # 连续 2 窗失败=系统性问题，停书防连环烧额度
+                with psycopg.connect(DSN) as conn:
-            if fail_streak >= 2:
+                    undo_window(conn, work_id, win_no)
-                click.echo(f"  ⛔ 连续 {fail_streak} 窗失败，本书升格停（末窗{win_no}）: {str(e)[:200]}")
+                    conn.commit()
-                return
+                retried = True
-            click.echo(f"  ✗ 窗{win_no} 失败（已记录，续跑重试；连续第{fail_streak}次）: {str(e)[:200]}")
+                click.echo(f"  ↻ 窗{win_no} 失败，撤销后当场重试（串行铁律不跳窗）: {str(e)[:150]}")
-            continue
+                continue
+            # 第二层重试仍败：停书，断点=本窗；下次启动 failed 先撤销、从本窗续跑
+            click.echo(f"  ⛔ 窗{win_no} 当场重试仍失败，本书升格停（断点=本窗，续跑从此接续）: {str(e)[:200]}")
+            return
     click.echo(f"《{title}》本次完成 {done_n} 窗，LLM 调用 {calls['n']} 次")