框架: 升格管线三处韧性修复(模型输出str防御/failed窗重跑先撤销/连续2窗失败才停书)

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

@@ -408,10 +408,19 @@ def run(work_id, max_windows, max_calls, model, redo_window):
                 undo_window(conn, work_id, redo_window)
         else:
             wins = conn.execute(
-                """SELECT window_no, from_chapter, to_chapter FROM example_upgrade_window
+                """SELECT window_no, from_chapter, to_chapter, status FROM example_upgrade_window
                    WHERE tenant_id=%s AND work_id=%s AND status!='done' AND deleted=FALSE
                    ORDER BY from_chapter""", (TENANT, work_id)).fetchall()
+            wins = [(r[0], r[1], r[2]) for r in wins]
+            # failed 窗重跑前必须先撤销旧写入（部分写入直接重跑会 double-append 追加字段）
+            for r0 in conn.execute(
+                    """SELECT window_no FROM example_upgrade_window
+                       WHERE tenant_id=%s AND work_id=%s AND status='failed' AND deleted=FALSE""",
+                    (TENANT, work_id)).fetchall():
+                click.echo(f"[撤销] failed 窗{r0[0]} 旧写入回滚后重跑")
+                undo_window(conn, work_id, r0[0])
     done_n = 0
+    fail_streak = 0   # 连续失败计数：单窗偶发失败（网络瞬断）continue，连续 2 窗停书（系统性问题）
     for win_no, a, b in wins:
         if max_windows and done_n >= max_windows:
             break
@@ -426,6 +435,9 @@ def run(work_id, max_windows, max_calls, model, redo_window):
                 # ① 观察
                 obs, usage = call(observe_prompt(contracts, title, a, b, text, onstage),
                                   ("新名字", "已知实体新信息", "纯出场"))
+                # 模型输出防御（深空窗5实测：列表元素偶为裸字符串，.get 直接炸）——统一只留 dict 元素
+                for k in ("新名字", "已知实体新信息", "纯出场"):
+                    obs[k] = [x for x in (obs.get(k) or []) if isinstance(x, dict)]
                 # ② 判重+立卡门槛（机械）
                 to_update = {}   # draft_id -> [观察点…]
                 for ent in obs.get("新名字", []):
@@ -480,7 +492,7 @@ def run(work_id, max_windows, max_calls, model, redo_window):
                         cards.append((did, p, obs_pts))
                     upd, _ = call(update_prompt(contracts, title, a, b, text, cards), ("更新",))
                     valid = {d for d, _, _ in cards}
-                    for u in upd.get("更新", []):
+                    for u in [x for x in (upd.get("更新") or []) if isinstance(x, dict)]:
                         if u.get("draft_id") in valid and (u.get("变更字段") or u.get("别名新增")):
                             merge_card(conn, u["draft_id"], win_no,
                                        u.get("变更字段"), u.get("别名新增"))
@@ -512,7 +524,7 @@ def run(work_id, max_windows, max_calls, model, redo_window):
                     exist = {tuple(sorted((r.get("甲方draft"), r.get("乙方draft")))): (rid, r)
                              for rid, r in rels
                              if r.get("甲方draft") and r.get("乙方draft")}
-                    for r in rel_out.get("关系", []):
+                    for r in [x for x in (rel_out.get("关系") or []) if isinstance(x, dict)]:
                         ja, yi = r.get("甲方"), r.get("乙方")
                         if ja not in id2name or yi not in id2name or ja == yi:
                             continue
@@ -564,6 +576,7 @@ 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
         except SensitiveHardStop as e:
             with psycopg.connect(DSN) as conn:
                 conn.execute(
@@ -580,8 +593,14 @@ 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()
-            click.echo(f"  ✗ 窗{win_no} 失败（已记录，续跑会重试）: {str(e)[:200]}")
+            fail_streak += 1
-            return
+            # 偶发失败（网络瞬断/单窗输出畸形）跳去下一窗，failed 窗留给续跑（自动先撤销）；
+            # 连续 2 窗失败=系统性问题，停书防连环烧额度
+            if fail_streak >= 2:
+                click.echo(f"  ⛔ 连续 {fail_streak} 窗失败，本书升格停（末窗{win_no}）: {str(e)[:200]}")
+                return
+            click.echo(f"  ✗ 窗{win_no} 失败（已记录，续跑重试；连续第{fail_streak}次）: {str(e)[:200]}")
+            continue
     click.echo(f"《{title}》本次完成 {done_n} 窗，LLM 调用 {calls['n']} 次")