#!/usr/bin/env python3 """parse_upgrade 升格拆卡执行器·纯逻辑离线自测（升格卡改造 3 洞补齐）。 红线：**不连库、不发任何网络/嵌入/LLM 调用**——只覆盖纯函数与提示词构造器： · 机械判重分类 _classify_new_name（各分支） · 登场兜底 _debut_milestone（有/无登场、无出场章、摘要超长） · 里程碑台阶守卫 _clean_milestone（超 80 字截断） · 归并材料 _merge_material（演变历程完整不截、其余字段截断） · embed_drafts.build_embed_text 对 type 键的型修正 · observe/update/relation 三个提示词含关键纪律语句 连库/联网路径（prejudge_semantic / recall_neighbors / semantic_dedup / embed_touched_cards） 本测试**不触碰**——它们需真连接，属放量前零落库小样阶段的活体验收，不在离线自测范围。 跑法：仓根 `.venv/bin/python .claude/skills/parse-book/scripts/test_parse_upgrade_offline.py` """ import json import pathlib import sys from copy import deepcopy from unittest.mock import patch from click.testing import CliRunner # 与 parse_upgrade 同目录：直接 import 触发其 sys.path 装配（含 embed/llm scripts），随后可导 embed_drafts sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) import parse_upgrade as pu # noqa: E402 import embed_drafts # noqa: E402 （型修正验证在 embed skill 本体） _passed = 0 def check(name, cond, detail=""): """单项断言：通过打 [PASS]，失败抛 AssertionError（带上下文，令 CI/人工一眼定位）。""" global _passed assert cond, f"[FAIL] {name} :: {detail}" _passed += 1 print(f"[PASS] {name}") # ── ① 机械判重分类 _classify_new_name（洞①：预判段与写段共用判据）── def test_classify_new_name(): name_map = { "张三": (10, "character", "主角"), "生物机甲": (20, "power_system", "机甲体系"), "开心果子": (30, "item", "道具"), } presence = {("character", "李四"): {5, 12}} # 李四靠留档跨章 check("classify-empty", pu._classify_new_name({"名称": " ", "型": "character"}, name_map, presence) == ("empty", None)) check("classify-merge(在册直接归并)", pu._classify_new_name({"名称": "张三", "型": "character"}, name_map, presence) == ("merge", "张三")) check("classify-alias(疑似别名指向在册)", pu._classify_new_name({"名称": "阿三", "型": "character", "疑似别名指向": "张三"}, name_map, presence) == ("alias", "张三")) check("classify-substr(同型互为子串)", pu._classify_new_name({"名称": "果子", "型": "item", "出场章": [1]}, name_map, presence) == ("substr", "开心果子")) # 子串判据须同型：果子(character) 不应命中 开心果子(item)，单章 → presence check("classify-substr跨型不命中", pu._classify_new_name({"名称": "果子", "型": "character", "出场章": [1]}, name_map, presence) == ("presence", None)) check("classify-new(本窗跨章立卡)", pu._classify_new_name({"名称": "王五", "型": "character", "出场章": [3, 8]}, name_map, presence) == ("new", None)) check("classify-new(留档补足跨章)", pu._classify_new_name({"名称": "李四", "型": "character", "出场章": [20]}, name_map, presence) == ("new", None)) check("classify-presence(单章龙套留档)", pu._classify_new_name({"名称": "路人甲", "型": "character", "出场章": [7]}, name_map, presence) == ("presence", None)) # ── ② 登场兜底 _debut_milestone（洞②机械那一保险）── def test_debut_milestone(): # 已有登场 → 原样返回，不重复补 has = [{"章": 10, "台阶": "训练机登场", "周期": "登场"}, {"章": 50, "台阶": "进化", "周期": "成长"}] out = pu._debut_milestone(has, "摘要", [8], 3) check("debut-已有登场不补", len(out) == 2 and out[0]["台阶"] == "训练机登场") # 无登场 + 有出场章 → 头部补登场，章=min(出场章)，_win 盖窗号 no_debut = [{"章": 50, "台阶": "V代编队", "周期": "成长"}] out = pu._debut_milestone(no_debut, "4级训练机", [15, 3, 8], 7) check("debut-无登场则补一条", len(out) == 2 and out[0]["周期"] == "登场" and out[0]["章"] == 3 and out[0]["_win"] == 7) check("debut-台阶取摘要", out[0]["台阶"].startswith("登场：4级训练机")) check("debut-进化台阶仍在尾部", out[1]["台阶"] == "V代编队") # 无正文实证出场章 → 不补登场里程碑（真实性优先，不能生成无证据台阶） out = pu._debut_milestone([], "无章摘要", [], 5) check("debut-无真实章不补", out == []) # milestones=None 也当空处理 out = pu._debut_milestone(None, "空列表摘要", [2], 5) check("debut-None当空补登场", len(out) == 1 and out[0]["周期"] == "登场" and out[0]["章"] == 2) # 摘要超长 → 台阶截 40 字摘要 + 过 STEP_MAX 守卫（总长 ≤ STEP_MAX） out = pu._debut_milestone([], "甲" * 100, [2], 9) step = out[0]["台阶"] check("debut-摘要超长台阶受守卫", len(step) <= pu.STEP_MAX and step.startswith("登场：") and "甲" * 40 in step) check("debut-摘要只取前40字", "甲" * 41 not in step) # ── ③ 里程碑台阶守卫 _clean_milestone（洞③：超 80 字截断）── def test_clean_milestone_guard(): # 对象入参：台阶 90 字 → 截到 80 m = pu._clean_milestone({"章": 100, "台阶": "阶" * 90, "周期": "成长"}, 1) check("clean-对象台阶截至80", m is not None and len(m["台阶"]) == 80 and m["台阶"] == "阶" * 80) check("clean-对象保真章周期", m["章"] == 100 and m["周期"] == "成长" and m["_win"] == 1) # 字符串入参：整串当台阶，同样受守卫 m = pu._clean_milestone("步" * 120, 2) check("clean-字符串台阶截至80", m is not None and len(m["台阶"]) == 80) # 正常短台阶不动 m = pu._clean_milestone({"章": 5, "台阶": "短台阶", "周期": "高光"}, 3) check("clean-短台阶不截", m["台阶"] == "短台阶" and len(m["台阶"]) == 3) # 截断顺序回归（主代理核验修复）：跑飞长文缺章缺周期时，藏在 80 字外的内嵌章号/周期关键词 # 必须先用**全文**抽取/推断、抽完再截——若先截后抽，兜底线索被截丢（章=None、周期误判成长）。 m = pu._clean_milestone({"台阶": "阶" * 85 + "突破至V代（见第432章）"}, 4) check("clean-截断前先抽全文章号", m is not None and m["章"] == 432) check("clean-截断前先推全文周期", m["周期"] == "高光") check("clean-抽后仍截至80", len(m["台阶"]) == 80) # 新抽取里程碑必须用短原文证据把章号绑定到对应正文；证据只用于校验，不写入卡体。 chapter_texts = { 488: "安若雪率领亲卫继续突进，环星防御火力全面展开。", 489: "安若雪与加特朗激战，同步率飙至100%，击碎能量屏障。", } verified = pu._clean_milestone( { "章": 489, "台阶": "同步率飙至100%击碎加特朗屏障", "周期": "高光", "证据": "同步率飙至100%，击碎能量屏障", }, 84, chapter_texts=chapter_texts, require_evidence=True, ) check( "clean-证据绑定正确章", verified is not None and verified["章"] == 489 and "证据" not in verified, ) wrong_chapter = pu._clean_milestone( { "章": 488, "台阶": "同步率飙至100%击碎加特朗屏障", "周期": "高光", "证据": "同步率飙至100%，击碎能量屏障", }, 84, chapter_texts=chapter_texts, require_evidence=True, ) check("clean-错章证据拒收", wrong_chapter is None) missing_evidence = pu._clean_milestone( {"章": 489, "台阶": "同步率飙至100%", "周期": "高光"}, 84, chapter_texts=chapter_texts, require_evidence=True, ) check("clean-缺证据拒收", missing_evidence is None) short_evidence = {"章": 488, "台阶": "继续突进", "周期": "成长", "证据": "安若雪率领亲卫"} check( "clean-不足8字证据拒收", not pu._milestone_evidence_matches(short_evidence, chapter_texts), ) long_quote = "证" * 31 long_evidence = {"章": 489, "台阶": "长证据", "周期": "成长", "证据": long_quote} check( "clean-超过30字证据拒收", not pu._milestone_evidence_matches(long_evidence, {489: f"前缀{long_quote}后缀"}), ) # ── ④ 归并材料 _merge_material（洞②：演变历程完整不截、其余字段截断）── def test_merge_material(): milestones = [{"章": i, "台阶": f"台阶第{i}条", "周期": "成长"} for i in range(1, 21)] ent = {"型": "item", "名称": "影杀者", "一句话摘要": "IV代机甲", "字段": {"演变历程": milestones, "能力与限制": "填充" * 500}} # 其余字段 1000 字 mat = pu._merge_material(ent, rest_limit=600) # 演变历程每条台阶完整保留（尤其末条不被截断丢失） check("material-演变历程首条在", "台阶第1条" in mat) check("material-演变历程末条在", "台阶第20条" in mat) check("material-摘要在", "IV代机甲" in mat and "影杀者" in mat) # 其余字段被截断（1000 字巨型字段不应整段进材料） check("material-其余字段被截断", ("填充" * 500) not in mat) # 无字段时不崩 check("material-空字段不崩", pu._merge_material({"型": "character", "名称": "甲", "一句话摘要": "s"})) # ── ⑤ build_embed_text 型键修正（洞①配套；改在 embed skill 本体）── def test_build_embed_text_type_fix(): # 升格卡用 type 键存型：修正后应认出（修正前会产「【】名称…」丢型） txt = embed_drafts.build_embed_text({"type": "power_system", "名称": "生物机甲", "一句话摘要": "体系", "字段": {"境界阶梯": "一代<二代"}}) check("embed-type键认出型", txt.startswith("【power_system】生物机甲")) check("embed-非空型不再丢", not txt.startswith("【】")) # 型 键仍优先（既有行为不动） check("embed-型键优先", embed_drafts.build_embed_text({"型": "item", "名称": "剑"}).startswith("【item】")) # target_type 回退仍在 check("embed-target_type回退", embed_drafts.build_embed_text({"target_type": "faction", "名称": "军团"}).startswith("【faction】")) # 三者优先级：型 > type > target_type check("embed-型优先于type与target_type", embed_drafts.build_embed_text({"型": "A", "type": "B", "target_type": "C", "名称": "x"}).startswith("【A】")) def _fake_contracts(): """离线假合同（覆盖六实体型 + 关系型），供三个提示词构造器渲染，不连库。""" c = {} for t in pu.ENTITY_TYPES + (pu.RELATION_TYPE,): c[t] = {"中文名": t, "判据": f"{t}判据", "字段": [{"key": "演变历程", "说明": "里程碑数组"}, {"key": "一句话摘要", "说明": "摘要"}]} return c # ── ⑦ 出场章归一化 _int_chaps（窗113 实证 bug：int/str 混排炸 + 并集虚增计数）── def test_int_chaps(): # 核心复现：模型给字符串章号 "508" 与库内 int 507 混合，旧代码 sorted 直接炸 mixed = [507, "508", 509] got = pu._int_chaps(mixed) check("intchaps-混类型归一为int", got == {507, 508, 509}) check("intchaps-归一后可安全排序", sorted(got) == [507, 508, 509]) # 并集不再虚增：{"508"} 与 {508} 旧代码算 2 个（误判跨章立卡），归一后算 1 个 union = pu._int_chaps(["508"]) | {508} check("intchaps-并集不虚增计数", union == {508} and len(union) == 1) # 脏值/区间/bool 丢弃（出场章只应是单章整数） check("intchaps-丢非数字与区间", pu._int_chaps(["420-423", "abc", None, 12]) == {12}) check("intchaps-排除bool", pu._int_chaps([True, False, 5]) == {5}) check("intchaps-空输入", pu._int_chaps([]) == set() and pu._int_chaps(None) == set()) class _CardResult: """为立卡与归并离线测试提供最小查询结果对象。""" def __init__(self, row=None, rows=None): self.row = row self.rows = rows or [] def fetchone(self): """返回预置的单行结果。""" return self.row def fetchall(self): """返回预置的多行结果。""" return self.rows class _CardConn: """只模拟卡片写入所需 SQL，并保留最终 payload 供机械断言。""" def __init__(self, payload=None): self.payload = payload self.audits = [] def execute(self, query, params): """按 SQL 用途返回最小结果，或捕获立卡、更新后的 payload。""" normalized = " ".join(query.split()) if normalized.startswith("SELECT draft_payload"): return _CardResult((self.payload,)) if normalized.startswith("SELECT watermark_window"): return _CardResult((0,)) if normalized.startswith("SELECT a.draft_id, a.field_name, a.old_value"): rows = [(did, field_name, old_value) for did, _, field_name, old_value in reversed(self.audits)] return _CardResult(rows=rows) if normalized.startswith("SELECT id, draft_payload FROM muse_knowledge_draft"): return _CardResult(rows=[(101, self.payload)]) if "INSERT INTO muse_knowledge_draft" in query: self.payload = json.loads(params[1]) return _CardResult((101,)) if "INSERT INTO example_upgrade_audit" in query and len(params) >= 4 \ and params[2] == "顶层:出场章": self.audits.append((params[0], params[1], params[2], params[3])) if normalized.startswith("DELETE FROM example_upgrade_audit"): self.audits = [] if "UPDATE muse_knowledge_draft SET draft_payload" in query: self.payload = json.loads(params[0]) return _CardResult() def commit(self): """模拟事务提交；离线测试中的状态已在内存立即生效。""" def test_entity_chapter_evidence_filter(): """模型出场章只能保留规范名或合法别名在对应正文真实出现的章节。""" chapter_texts = { 488: "环星防线开启，亲卫继续突进。", 489: "安若雪同步率飙至100%，击碎能量屏障。", 490: "众人称雪姐已经抵达核心区。", 491: "备忘录提到雪（指挥官）这个带注释称呼。", } got = pu._filter_entity_chapters( "安若雪", ["雪姐", "雪（指挥官）", "新"], [488, "489", 490, 491], chapter_texts, ) check("chapter-evidence-错488过滤为489与合法别名章", got == {489, 490}) parenthetical = pu._filter_entity_chapters( "白色游魂（无名侦察兵）", [], [492], {492: "白色游魂从破损舱门后现身。"}, ) check("chapter-evidence-括号注名称按规范名命中", parenthetical == {492}) false_positive_texts = { 493: "大小姐转身离开大厅。", 494: "小姐姐转身离开大厅。", 495: "雪姐已经抵达核心区。", 496: "安若雪已经抵达核心区。", 497: "小姐转身离开大厅。", } check( "chapter-evidence-单字规范名禁作证据", pu._filter_entity_chapters("雪", [], [495, 496], false_positive_texts) == set(), ) check( "chapter-evidence-通用称谓别名完全禁作证据", pu._filter_entity_chapters("安若雪", ["小姐"], [493, 494, 497], false_positive_texts) == set(), ) generic_titles = ( "队长", "舰长", "指挥官", "司令", "统领", "院长", "校长", "会长", "团长", "主任", "长老", "领主", "城主", "陛下", "殿下", ) titles = {500 + index: f"{title}下令立刻行动。" for index, title in enumerate(generic_titles)} check( "chapter-evidence-规范名常见职务称谓全部禁作证据", all( pu._filter_entity_chapters(title, [], [500 + index], titles) == set() for index, title in enumerate(generic_titles) ), ) check( "chapter-evidence-别名常见职务称谓全部禁作证据", all( pu._filter_entity_chapters("安若雪", [title], [500 + index], titles) == set() for index, title in enumerate(generic_titles) ), ) proper_names = { 530: "银河指挥官越过了环星防线。", 531: "青云院长打开密室入口。", 532: "玄天宗主唤醒护山大阵。", 533: "司令塔发出低沉警报。", } check( "chapter-evidence-完整专名不因职务片段误杀", pu._filter_entity_chapters( "银河指挥官", ["青云院长", "玄天宗主", "司令塔"], ["530", 531, 532, 533], proper_names, ) == {530, 531, 532, 533}, ) check( "chapter-evidence-常规二至四字专名不受损", pu._filter_entity_chapters("安若雪", ["雪姐"], [495, 496], false_positive_texts) == {495, 496}, ) def test_new_card_chapter_evidence_chain(): """串联复现：错章里程碑被拒后，初卡只能用正文实证章补登场。""" conn = _CardConn() pu.new_card( conn, 8, 84, { "型": "character", "名称": "安若雪", "别名": ["雪姐"], "一句话摘要": "联邦战士", "字段": { "演变历程": [ { "章": 488, "台阶": "同步率飙至100%", "周期": "高光", "证据": "同步率飙至100%，击碎能量屏障", } ] }, "出场章": [488, 489], }, {"character"}, chapter_texts={ 488: "环星防线开启，亲卫继续突进。", 489: "安若雪同步率飙至100%，击碎能量屏障。", }, ) milestones = conn.payload["字段"]["演变历程"] check("new-card-错488过滤为489", conn.payload["出场章"] == [489]) check( "new-card-拒错里程碑后仅补实证登场", len(milestones) == 1 and milestones[0]["章"] == 489 and milestones[0]["周期"] == "登场" and milestones[0]["_win"] == 84, ) no_evidence_conn = _CardConn() pu.new_card( no_evidence_conn, 8, 85, { "型": "character", "名称": "安若雪", "别名": [], "一句话摘要": "联邦战士", "字段": {"演变历程": []}, "出场章": [488], }, {"character"}, chapter_texts={488: "环星防线开启，亲卫继续突进。"}, ) check("new-card-无真实章不落出场章", no_evidence_conn.payload["出场章"] == []) check("new-card-无真实章不补里程碑", no_evidence_conn.payload["字段"]["演变历程"] == []) history_conn = _CardConn() pu.new_card( history_conn, 8, 86, { "型": "character", "名称": "安若雪", "别名": [], "一句话摘要": "联邦战士", "字段": {"演变历程": []}, "出场章": [489], }, {"character"}, chapter_texts={489: "安若雪抵达核心区。"}, known_chapters=[480], ) check("new-card-历史留档章不被当前窗过滤", history_conn.payload["出场章"] == [480, 489]) def test_merge_card_chapter_evidence(): """既有卡顶层出场章追加必须复用正文实体命中过滤。""" conn = _CardConn( { "type": "character", "名称": "安若雪", "别名": ["雪姐"], "一句话摘要": "联邦战士", "字段": {}, "出场章": [487], "_work_id": 8, } ) pu.merge_card( conn, 101, 84, {}, [], chapter_texts={ 488: "环星防线开启，亲卫继续突进。", 489: "雪姐同步率飙至100%，击碎能量屏障。", }, appearance_chapters=[488, 489], ) check("merge-card-顶层错488过滤为489", conn.payload["出场章"] == [487, 489]) # 顶层出场章必须随窗可撤销；正文变化后重跑不得残留上一轮已失效的章。 pu.undo_window(conn, 8, 84) check("merge-card-undo精确恢复旧出场章", conn.payload["出场章"] == [487]) pu.merge_card( conn, 101, 84, {}, [], chapter_texts={ 489: "同步率飙至100%，击碎能量屏障。", 490: "安若雪已经抵达核心区。", }, appearance_chapters=[489, 490], ) check("merge-card-正文变化重跑删除旧错章", conn.payload["出场章"] == [487, 490]) def test_redo_cleans_legacy_appearance_chapters(): """显式 redo 清理历史无审计章域，并可由失败撤销或本窗全文重新建立。""" # 修复前历史 payload 没有顶层审计：首次 redo 必须先清空指定窗域，再只重建正文真实出现章。 conn = _CardConn( { "type": "character", "名称": "安若雪", "别名": [], "字段": {}, "出场章": [487, 489], "_work_id": 8, } ) pu.undo_window(conn, 8, 84, from_chapter=487, to_chapter=489) check("redo-历史无审计窗内章清空", conn.payload["出场章"] == []) # 首次失败后的即时重试走“恢复旧值后重新清理”，第二次尝试前窗内仍必须为空。 pu.undo_window(conn, 8, 84, from_chapter=487, to_chapter=489) check("redo-首次失败重试前重新清理", conn.payload["出场章"] == []) # 清理本身带旧值审计：模拟后续处理失败时的既有失败撤销，必须恢复清理前历史值。 pu.undo_window(conn, 8, 84) check("redo-后续失败可恢复历史章", conn.payload["出场章"] == [487, 489]) pu.undo_window(conn, 8, 84, from_chapter=487, to_chapter=489) pu.merge_card( conn, 101, 84, {}, [], chapter_texts={ 487: "环星防线开启，亲卫继续突进。", 488: "同步率继续上升。", 489: "安若雪击碎能量屏障。", }, appearance_chapters=[487, 488, 489], ) check("redo-全文重建只落真实489", conn.payload["出场章"] == [489]) outside_conn = _CardConn( { "type": "character", "名称": "安若雪", "别名": [], "字段": {}, "出场章": [486, 487, 489, 490], "_work_id": 8, } ) pu.undo_window(outside_conn, 8, 84, from_chapter=487, to_chapter=489) check("redo-窗外章完整保留", outside_conn.payload["出场章"] == [486, 490]) normal_conn = _CardConn( { "type": "character", "名称": "安若雪", "别名": [], "字段": {}, "出场章": [487, 489], "_work_id": 8, } ) pu.undo_window(normal_conn, 8, 84) check("redo-非显式redo不做章域清理", normal_conn.payload["出场章"] == [487, 489]) class _RunConn: """run 级离线夹具：只模拟控制流所需 SQL，绝不建立真实连接。""" def __init__(self, state): self.state = state self.writes = [] self.commits = 0 def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def execute(self, query, params=()): normalized = " ".join(query.split()) if normalized.startswith("SELECT title FROM muse_content_work"): return _CardResult(("离线书",)) if normalized.startswith("SELECT window_no, from_chapter, to_chapter FROM example_upgrade_window"): return _CardResult(rows=self.state.get("windows", [(7, 70, 71)])) if normalized.startswith("SELECT window_no, from_chapter, to_chapter, status FROM example_upgrade_window"): return _CardResult(rows=[(7, 70, 71, "pending")]) if normalized.startswith("SELECT window_no FROM example_upgrade_window"): return _CardResult(rows=[]) if normalized.startswith("SELECT draft_payload FROM muse_knowledge_draft"): return _CardResult((self.state["payload"],)) if normalized.startswith("UPDATE muse_knowledge_draft SET draft_payload=%s"): self.state["payload"] = json.loads(params[0]) if normalized.startswith("UPDATE example_upgrade_window SET status='failed'"): self.state["window_status"] = "failed" if normalized.startswith("UPDATE example_upgrade_window SET status='done'"): self.state["window_status"] = "done" if not normalized.startswith("SELECT"): self.writes.append((normalized, params)) return _CardResult() def commit(self): """内存状态即时生效。""" self.commits += 1 def test_run_redo_preflight_rejects_unsafe_targets(): """所有 redo 非法输入都必须在快照、撤销与写入前失败。""" cases = [ ("历史窗", [(1, 1, 2), (2, 3, 4)], 1, {1: "甲", 2: "乙"}, "仅允许重跑当前末窗"), ("不存在", [(1, 1, 2)], 2, {}, "不存在"), ("窗口号断裂", [(1, 1, 2), (3, 3, 4)], 3, {}, "窗口号不连续"), ("章域断裂", [(1, 1, 2), (2, 4, 5)], 2, {}, "窗章域不连续"), ("正文缺章", [(1, 70, 71)], 1, {70: "正文"}, "正文不完整"), ("正文为空", [(1, 70, 71)], 1, {70: "正文", 71: " "}, "正文不完整"), ] for name, windows, target, chapter_texts, expected_error in cases: state = {"windows": windows} snapshot_calls, undo_calls, connections = [], [], [] def fake_connect(*_): conn = _RunConn(state) connections.append(conn) return conn def fake_material(conn, work_id, a, b): text = "\n".join(str(chapter_texts.get(chapter, "")) for chapter in range(a, b + 1)) return text, chapter_texts with patch.object(pu.psycopg, "connect", side_effect=fake_connect), \ patch.object(pu, "load_entity_contracts", return_value=_fake_contracts()), \ patch.object(pu, "load_window_material", side_effect=fake_material), \ patch.object(pu, "_snapshot_redo_window", side_effect=lambda *args: snapshot_calls.append(args)), \ patch.object(pu, "undo_window", side_effect=lambda *args, **kwargs: undo_calls.append((args, kwargs))): result = CliRunner().invoke( pu.cli, ["run", "--work-id", "8", "--redo-window", str(target)], ) check( f"redo-preflight-{name}-非零拒绝", result.exit_code != 0 and expected_error in result.output, detail=f"exit={result.exit_code}, output={result.output!r}", ) check(f"redo-preflight-{name}-无副作用", not snapshot_calls and not undo_calls and not any(conn.writes or conn.commits for conn in connections)) check(f"redo-preflight-{name}-无假完成", "完成 0 窗" not in result.output) def test_run_rejects_negative_limits_before_redo_side_effects(): """负调用闸必须由 Click 在函数入口拒绝，即使同时请求合法末窗 redo 也不得接触数据库。""" for option in ("--max-windows", "--max-calls"): with patch.object(pu.psycopg, "connect") as connect, \ patch.object(pu, "_snapshot_redo_window") as snapshot, \ patch.object(pu, "undo_window") as undo: result = CliRunner().invoke( pu.cli, ["run", "--work-id", "8", "--redo-window", "2", option, "-1"], ) check( f"negative-limit-{option}-入口非零拒绝", result.exit_code != 0, detail=f"exit={result.exit_code}, output={result.output!r}", ) check( f"negative-limit-{option}-无snapshot与undo", not snapshot.called and not undo.called, ) check(f"negative-limit-{option}-无DB调用", not connect.called) def _run_with_failures(*, max_calls, failures): """执行单个显式 redo 窗；前 failures 次观察调用失败，返回状态与调用数。""" old_state = { "windows": [(1, 1, 69), (2, 70, 71)], "window_status": "done", "fields": {"阵营": "旧阵营", "经历": ["[窗2] 旧经历"]}, "aliases": ["旧别名"], "presence": [(2, 70, "旧龙套")], "initial_cards": [701], } state = deepcopy(old_state) calls = {"n": 0} undo_calls = [] def fake_undo(conn, work_id, win_no, *, from_chapter=None, to_chapter=None): undo_calls.append((from_chapter, to_chapter)) # 复现旧实现边界：章域 undo 会清完整旧窗；无章域 undo 只能恢复顶层章，救不回其余旧窗状态。 state["fields"] = {} state["aliases"] = [] state["presence"] = [] state["initial_cards"] = [] def fake_m3_json(prompt, model, need_keys, system=None): calls["n"] += 1 if calls["n"] <= failures: raise RuntimeError(f"离线失败{calls['n']}") return ({"新名字": [], "已知实体新信息": [], "纯出场": []}, {}) snapshot = deepcopy(old_state) def fake_restore(conn, work_id, redo_snapshot): state.clear() state.update(deepcopy(redo_snapshot)) with patch.object(pu.psycopg, "connect", side_effect=lambda *_: _RunConn(state)), \ patch.object(pu, "load_entity_contracts", return_value=_fake_contracts()), \ patch.object(pu, "load_window_material", return_value=("正文", {70: "正文", 71: "正文"})), \ patch.object(pu, "load_known", return_value=({}, {}, {})), \ patch.object(pu, "undo_window", side_effect=fake_undo), \ patch.object(pu, "m3_json", side_effect=fake_m3_json), \ patch.object(pu, "_snapshot_redo_window", return_value=snapshot), \ patch.object(pu, "_restore_redo_window", side_effect=fake_restore): pu.run.callback( work_id=8, max_windows=0, max_calls=max_calls, model="MiniMax-M3", redo_window=2, semantic_on=False, ) return old_state, state, calls["n"], undo_calls def test_run_redo_max_calls_finishes_active_retry(): """调用闸不能在首次失败后的已清理中间态退出。""" _, state, call_count, undo_calls = _run_with_failures(max_calls=1, failures=1) check("run-max-calls-已清理窗仍完成即时重试", call_count == 2) check("run-max-calls-重试成功落done", state["window_status"] == "done") check("run-max-calls-首次失败前后均执行章域清理", undo_calls == [(70, 71), (70, 71)]) def test_run_redo_final_failure_restores_full_snapshot(): """显式 redo 两次都失败时，完整恢复字段、别名、presence 与初建卡。""" old_state, state, call_count, _ = _run_with_failures(max_calls=0, failures=2) check("run-redo-final-failure-确实尝试两次", call_count == 2) check("run-redo-final-failure-完整恢复旧窗", state == old_state, detail=f"state={state!r}") class _SnapshotConn: """恢复点 SQL 的离线记录器：提供旧窗快照行，并记录所有恢复写入。""" def __init__(self): self.writes = [] def execute(self, query, params=()): normalized = " ".join(query.split()) if normalized.startswith("SELECT id, draft_payload, revision, updater, deleted"): return _CardResult(rows=[(701, {"字段": {"阵营": "旧阵营"}}, 9, "old", False)]) if normalized.startswith("SELECT id, canonical_name, alias, evidence_window"): return _CardResult(rows=[(801, "旧主角", "旧别名", 7, "ai", "", "t1", "", "t2", False)]) if normalized.startswith("SELECT id, window_no, chapter_no, entity_type"): return _CardResult(rows=[(901, 7, 70, "character", "旧龙套", "旧观察", "", "t3", False)]) if normalized.startswith("SELECT draft_id, watermark_window, update_time"): return _CardResult(rows=[(701, 7, "t4")]) if normalized.startswith("SELECT a.id, a.draft_id, a.window_no"): return _CardResult(rows=[(1001, 701, 7, "阵营", '"更旧阵营"', '"旧阵营"', "t5")]) if normalized.startswith("SELECT status, error_message, updater FROM example_upgrade_window"): return _CardResult(("done", None, "old")) if normalized.startswith("SELECT id FROM muse_knowledge_draft"): return _CardResult(rows=[(701,), (702,)]) self.writes.append((normalized, params)) return _CardResult() def test_redo_snapshot_restore_sql_boundaries(): """恢复点覆盖旧字段、别名、presence、初建卡、水位、审计与窗状态。""" conn = _SnapshotConn() snapshot = pu._snapshot_redo_window(conn, 8, 7) check("redo-snapshot-捕获完整旧卡", snapshot["drafts"][0][1]["字段"]["阵营"] == "旧阵营") check("redo-snapshot-捕获别名presence", snapshot["aliases"] and snapshot["presence"]) check("redo-snapshot-捕获初建卡水位", snapshot["card_states"] == [(701, 7, "t4")]) check("redo-snapshot-捕获旧窗状态", snapshot["window"] == ("done", None, "old")) pu._restore_redo_window(conn, 8, snapshot) writes = conn.writes check( "redo-restore-恢复旧卡字段与revision", any(sql.startswith("UPDATE muse_knowledge_draft SET draft_payload=%s, revision=%s") and params[:3] == (json.dumps({"字段": {"阵营": "旧阵营"}}, ensure_ascii=False), 9, "old") for sql, params in writes), ) check( "redo-restore-软删重试新建卡", any(sql.startswith("UPDATE muse_knowledge_draft SET deleted=TRUE") and params == (702,) for sql, params in writes), ) check( "redo-restore-恢复别名与presence", any("INSERT INTO example_upgrade_alias" in sql and "旧别名" in params for sql, params in writes) and any("INSERT INTO example_upgrade_presence" in sql and "旧龙套" in params for sql, params in writes), ) check( "redo-restore-恢复初建卡水位与旧窗状态", any("INSERT INTO example_upgrade_card_state" in sql and params[0] == 701 for sql, params in writes) and any(sql.startswith("UPDATE example_upgrade_window SET status=%s") and params[:3] == ("done", None, "old") for sql, params in writes), ) def test_presence_chapter_normalization_boundaries(): """归并卡与纯出场章号都必须接受数字字符串，并丢弃非数字值。""" check("presence-归并卡数字字符串", pu._int_chaps(["489", 490, "bad"]) == {489, 490}) check("presence-纯出场数字字符串", pu._int_chaps(["491", " 492 ", None]) == {491, 492}) def test_run_presence_paths_accept_numeric_strings(): """run 的已知更新归并卡与纯出场都把数字字符串章号落成整数。""" state = { "window_status": "pending", "payload": { "type": "character", "名称": "安若雪", "别名": [], "字段": {}, "出场章": [], "_work_id": 8, }, } def fake_m3_json(prompt, model, need_keys, system=None): if need_keys == ("新名字", "已知实体新信息", "纯出场"): return ({ "新名字": [], "已知实体新信息": [ {"名称": "安若雪", "观察点": "出现新变化", "出场章": ["70"]} ], "纯出场": [{"名称": "安若雪", "出场章": ["71"]}], }, {}) return ({"更新": []}, {}) with patch.object(pu.psycopg, "connect", side_effect=lambda *_: _RunConn(state)), \ patch.object(pu, "load_entity_contracts", return_value=_fake_contracts()), \ patch.object( pu, "load_window_material", return_value=("安若雪连续出现", {70: "安若雪出现新变化。", 71: "安若雪继续前进。"}), ), \ patch.object( pu, "load_known", return_value=({"安若雪": (701, "character", "旧摘要")}, {}, {}), ), \ patch.object(pu, "m3_json", side_effect=fake_m3_json): pu.run.callback( work_id=8, max_windows=0, max_calls=0, model="MiniMax-M3", redo_window=0, semantic_on=False, ) check("run-presence-归并与纯出场数字字符串均落库", state["payload"]["出场章"] == [70, 71]) def test_load_known_projection(): """判重底册只投影索引字段，禁止跨网搬运每张卡的完整 payload。""" class Result: def __init__(self, rows): self.rows = rows def fetchall(self): return self.rows class Conn: def __init__(self): self.queries = [] def execute(self, query, params): self.queries.append(query) if "muse_knowledge_draft" in query: return Result([(1, "character", "安若雪", "联邦战士", [])]) if "example_upgrade_alias" in query: return Result([("安若雪", "雪姐")]) return Result([("character", "李四", 7)]) conn = Conn() name_map, presence, aliases_by_draft = pu.load_known(conn, 8) check("known-正名投影", name_map["安若雪"] == (1, "character", "联邦战士")) check("known-别名表投影", name_map["雪姐"] == name_map["安若雪"]) check("known-按卡携带alias表独有别名", aliases_by_draft[1] == {"雪姐"}) check("known-presence保留", presence[("character", "李四")] == {7}) first_query = conn.queries[0] check( "known-禁搬完整payload", "SELECT id, draft_payload FROM" not in first_query and "draft_payload->>'名称'" in first_query, ) merge_conn = _CardConn( { "type": "character", "名称": "安若雪", "别名": [], "一句话摘要": "联邦战士", "字段": {}, "出场章": [488], "_work_id": 8, } ) pu.merge_card( merge_conn, 1, 84, {}, [], chapter_texts={489: "雪姐已经抵达核心区。"}, appearance_chapters=[489], known_aliases=aliases_by_draft[1], ) check("known-alias表独有别名可供merge过滤", merge_conn.payload["出场章"] == [488, 489]) pure_payload = { "名称": "安若雪", "别名": [], "出场章": [488], } pure_conn = _CardConn(pure_payload) changed = pu._append_verified_appearance_chapters( pure_conn, 1, 84, pure_payload, [489], {489: "雪姐已经抵达核心区。"}, known_aliases=aliases_by_draft[1], ) check( "known-alias表独有别名可供纯出场过滤", changed and pure_payload["出场章"] == [488, 489], ) def test_repair_milestone_evidence(): """只修本窗缺证据里程碑；未来重抄项不能借修复调用混回卡体。""" output = { "更新": [ { "draft_id": 1, "变更字段": { "演变历程": [ {"章": 489, "台阶": "同步率飙至100%", "周期": "高光"}, {"章": 493, "台阶": "未来治疗事件", "周期": "高光"}, ] }, } ] } chapter_texts = {488: "防线开启。", 489: "安若雪同步率飙至100%，击碎能量屏障。"} seen = {} def fake_call(prompt, required_keys): seen["prompt"] = prompt seen["required_keys"] = required_keys return ( { "证据修复": [ { "ref": "$.更新[0].变更字段.演变历程[0]", "章": 489, "证据": "同步率飙至100%，击碎能量屏障", } ] }, {}, ) repaired = pu.repair_missing_milestone_evidence( output, title="深空之影", a=488, b=489, text="## 第488章\n防线开启。\n## 第489章\n安若雪同步率飙至100%，击碎能量屏障。", chapter_texts=chapter_texts, call=fake_call, ) milestones = output["更新"][0]["变更字段"]["演变历程"] check("repair-本窗证据补齐", repaired == 1 and "证据" in milestones[0]) check("repair-未来重抄不送修复", "证据" not in milestones[1] and "未来治疗事件" not in seen["prompt"]) check("repair-固定输出键", seen["required_keys"] == ("证据修复",)) # ── ⑥ 三个提示词含关键纪律语句（洞②登场必须 / 洞③≤40字 + 体系级 / 洞②归并去重）── def test_prompts_disciplines(): c = _fake_contracts() obs = pu.observe_prompt(c, "测试书", 1, 12, "正文占位", {"张三": (10, "character", "主角")}) check("observe-登场必须硬约束", "必须给「演变历程」的首条登场里程碑" in obs) check("observe-台阶≤40字", "≤40 字" in obs) check("observe-体系级纪律", "体系级纪律" in obs) check("observe-里程碑要求原文证据", '"证据"' in obs and "正文原样连续短句" in obs) upd = pu.update_prompt(c, "测试书", 1, 12, "正文占位", [(10, {"type": "power_system", "名称": "生物机甲"}, ["观察点"])]) check("update-台阶≤40字", "≤40 字" in upd) check("update-体系级纪律", "体系级纪律" in upd) check("update-登场归并去重", "若本卡演变历程已有登场里程碑则不再重复追加" in upd) check("update-里程碑要求原文证据", '"证据"' in upd and "正文原样连续短句" in upd) rel = pu.relation_prompt(c, "测试书", 1, 12, "正文占位", [(10, {"名称": "张三"}), (11, {"名称": "李四"})], []) check("relation-构造器可渲染", "人物关系增量" in rel) # 判重终判 prompt（小样校准后）：近邻六元组带字段摘选进证据 + 名称互含判据 nbs = [(6223, "power_system", "生物机甲", "机甲体系", '{"境界阶梯":"IV代→V代"}', 0.66)] jp = pu.dedup_judge_prompt({"型": "power_system", "名称": "联邦生物机甲技术", "一句话摘要": "s", "字段": {}}, nbs) check("judge-近邻字段摘选入证据", "字段摘选：" in jp and "境界阶梯" in jp) check("judge-名称互含判据", "同型且名称互含" in jp) check("judge-跨型禁判同一实体仍在", "绝不判同一实体" in jp) if __name__ == "__main__": for fn in (test_classify_new_name, test_debut_milestone, test_clean_milestone_guard, test_merge_material, test_build_embed_text_type_fix, test_int_chaps, test_entity_chapter_evidence_filter, test_new_card_chapter_evidence_chain, test_merge_card_chapter_evidence, test_redo_cleans_legacy_appearance_chapters, test_run_redo_preflight_rejects_unsafe_targets, test_run_rejects_negative_limits_before_redo_side_effects, test_run_redo_max_calls_finishes_active_retry, test_run_redo_final_failure_restores_full_snapshot, test_redo_snapshot_restore_sql_boundaries, test_presence_chapter_normalization_boundaries, test_run_presence_paths_accept_numeric_strings, test_load_known_projection, test_repair_milestone_evidence, test_prompts_disciplines): fn() print(f"\n全部离线自测通过：{_passed} 项（未连库、未发任何网络/嵌入/LLM 调用）")