muse-agent-example/.claude/skills/parse-book/scripts/test_parse_upgrade_offline.py

#!/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 hashlib
import pathlib
import sys
from contextlib import nullcontext
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))


def test_resolve_canonical_name():
    """name_map 的 alias 键命中时，落库正名仍必须取同 draft 的真实活跃卡名称。"""

    target = (701, "item", "一代机械外骨骼")
    name_map = {
        "一代机械外骨骼·铁壳": target,
        "铁壳": target,
        "机械铁壳": target,
    }
    aliases_by_draft = {701: {"铁壳", "机械铁壳"}}
    check(
        "canonical-alias-token解析为真实卡名",
        pu._resolve_canonical_name(name_map, aliases_by_draft, "铁壳") == "一代机械外骨骼·铁壳",
    )
    check(
        "canonical-substring命中alias仍解析为真实卡名",
        pu._resolve_canonical_name(name_map, aliases_by_draft, "机械铁壳") == "一代机械外骨骼·铁壳",
    )
    try:
        pu._resolve_canonical_name({"铁壳": target}, {701: {"铁壳"}}, "铁壳")
    except RuntimeError as exc:
        check("canonical-找不到真实卡名失败关闭", "真实规范名" in str(exc))
    else:
        check("canonical-找不到真实卡名失败关闭", False, detail="未抛 RuntimeError")


# ── ② 登场兜底 _debut_milestone（洞②机械那一保险）──
def test_debut_milestone():
    # 已有登场 → 原样返回，不重复补
    has = [{"章": 10, "台阶": "训练机登场", "周期": "登场"}, {"章": 50, "台阶": "进化", "周期": "成长"}]
    out = pu._debut_milestone(has, "训练机", "item", [], [8], 3, {8: "训练机进入机库。"})
    check("debut-已有登场不补", len(out) == 2 and out[0]["台阶"] == "训练机登场")

    # 无登场 + 有正文实证出场章 → 只补中性名称/型描述，章=min(出场章)，_win 盖窗号
    no_debut = [{"章": 50, "台阶": "V代编队", "周期": "成长"}]
    out = pu._debut_milestone(
        no_debut,
        "4级训练机",
        "item",
        [],
        [15, 3, 8],
        7,
        {3: "4级训练机被推入测试场。", 8: "4级训练机完成返航。", 15: "4级训练机封存。"},
    )
    check("debut-无登场则补一条",
          len(out) == 2 and out[0]["周期"] == "登场" and out[0]["章"] == 3 and out[0]["_win"] == 7)
    check("debut-台阶仅用中性名称与型", out[0]["台阶"] == "物件「4级训练机」登场")
    check("debut-进化台阶仍在尾部", out[1]["台阶"] == "V代编队")

    # 无正文实证出场章 → 不补登场里程碑（真实性优先，不能生成无证据台阶）
    out = pu._debut_milestone([], "无章实体", "item", [], [], 5, {})
    check("debut-无真实章不补", out == [])

    # milestones=None 也当空处理
    out = pu._debut_milestone(None, "空列表实体", "item", [], [2], 5, {2: "空列表实体出现。"})
    check("debut-None当空补登场", len(out) == 1 and out[0]["周期"] == "登场" and out[0]["章"] == 2)

    # 反例：实体第57章首现，但当前窗后续章材料含第59/60章事实；兜底绝不能把未来事实倒灌到57章。
    future_fact = "第59章升级二代装甲，第60章击毁母舰"
    out = pu._debut_milestone(
        [],
        "一代机械外骨骼·铁壳",
        "item",
        ["铁壳"],
        [57, 59, 60],
        8,
        {
            57: "一代机械外骨骼·铁壳第一次走出维修架。",
            59: "铁壳升级成二代装甲。",
            60: "铁壳击毁母舰。",
        },
    )
    check("debut-57章台阶不含59与60章事实",
          out[0]["章"] == 57 and future_fact not in out[0]["台阶"]
          and "升级" not in out[0]["台阶"] and "母舰" not in out[0]["台阶"])
    alias_only = pu._debut_milestone(
        [],
        "二代机械外骨骼·铁壳",
        "item",
        ["铁壳"],
        [57],
        8,
        {57: "铁壳第一次走出维修架。"},
    )
    check("debut-规范名未出现时只用首章真实别名", alias_only[0]["台阶"] == "物件「铁壳」登场")
    ambiguous = pu._debut_milestone(
        [],
        "二代机械外骨骼·铁壳",
        "item",
        ["铁壳", "老铁"],
        [57],
        8,
        {57: "铁壳又被队员叫作老铁。"},
    )
    check("debut-多名称命中退化为仅类型", ambiguous[0]["台阶"] == "物件登场")
    generic_only = pu._debut_milestone(
        [], "未来规范名", "character", ["队长"], [57], 8, {57: "队长首次出现。"}
    )
    check("debut-通用称谓不得进入台阶", generic_only[0]["台阶"] == "人物登场")
    check("debut-缺少首章正文宁可不补",
          pu._debut_milestone([], "铁壳", "item", [], [57], 8, {59: "铁壳升级。"}) == [])


# ── ③ 里程碑台阶守卫 _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 = []
        self.revision = 0

    def execute(self, query, params):
        """按 SQL 用途返回最小结果，或捕获立卡、更新后的 payload。"""

        normalized = " ".join(query.split())
        if normalized.startswith(
                "SELECT draft_payload, status, revision, source_type FROM muse_knowledge_draft"):
            return _CardResult((self.payload, "pending", self.revision, pu.SOURCE_TYPE))
        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 normalized.startswith("INSERT INTO example_upgrade_alias"):
            return _CardResult((params[1],))
        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])
            if "revision=revision+1" in normalized:
                self.revision += 1
                return _CardResult((self.revision,))
        return _CardResult()

    def commit(self):
        """模拟事务提交；离线测试中的状态已在内存立即生效。"""


class _RelationConn:
    """关系卡持久化与 undo 的内存数据库，覆盖 payload、审计、水位和软删。"""

    def __init__(self, drafts=None, states=None):
        self.drafts = {
            did: {
                "payload": deepcopy(payload),
                "deleted": False,
                "status": "pending",
                "source_type": pu.SOURCE_TYPE,
                "revision": 0,
                "updater": "before",
            }
            for did, payload in (drafts or {}).items()
        }
        self.states = dict(states or {})
        self.audits = []
        self.next_draft_id = max(self.drafts, default=700) + 1
        self.before_lock = None

    def execute(self, query, params=()):
        """按生产 SQL 语义更新内存状态，测试不建立任何真实连接。"""

        normalized = " ".join(query.split())
        if normalized.startswith(
                "SELECT draft_payload, status, revision, source_type FROM muse_knowledge_draft"):
            if "FOR UPDATE" in normalized and self.before_lock:
                self.before_lock(params[0])
            row = self.drafts.get(params[0])
            if not row or row["deleted"]:
                return _CardResult(None)
            return _CardResult(
                (row["payload"], row["status"], row["revision"], row["source_type"])
            )
        if normalized.startswith(
                "SELECT work_id, draft_payload, status, revision, source_type, deleted"):
            row = self.drafts.get(params[0])
            if not row:
                return _CardResult(None)
            return _CardResult((
                8, row["payload"], row["status"], row["revision"],
                row["source_type"], row["deleted"],
            ))
        if normalized.startswith("SELECT watermark_window FROM example_upgrade_card_state"):
            watermark = self.states.get(params[0])
            return _CardResult((watermark,) if watermark is not None else None)
        if normalized.startswith("SELECT a.draft_id, a.field_name, a.old_value"):
            window_no = params[1]
            rows = [
                (item["draft_id"], item["field_name"], item["old_value"])
                for item in reversed(self.audits)
                if item["window_no"] == window_no
            ]
            return _CardResult(rows=rows)
        if normalized.startswith("SELECT draft_payload FROM muse_knowledge_draft WHERE id=%s"):
            return _CardResult((self.drafts[params[0]]["payload"],))
        if normalized.startswith("SELECT id, draft_payload FROM muse_knowledge_draft"):
            return _CardResult(rows=[
                (did, row["payload"])
                for did, row in self.drafts.items()
                if not row["deleted"]
            ])
        if normalized.startswith("INSERT INTO example_upgrade_audit"):
            self.audits.append({
                "draft_id": params[0],
                "window_no": params[1],
                "field_name": params[2],
                "old_value": params[3],
                "new_value": params[4],
            })
            return _CardResult()
        if normalized.startswith("INSERT INTO muse_knowledge_draft"):
            did = self.next_draft_id
            self.next_draft_id += 1
            self.drafts[did] = {
                "payload": json.loads(params[1]),
                "deleted": False,
                "status": "pending",
                "source_type": pu.SOURCE_TYPE,
                "revision": 0,
                "updater": "upgrade",
            }
            return _CardResult((did,))
        if normalized.startswith("UPDATE muse_knowledge_draft SET draft_payload=%s"):
            row = self.drafts[params[1]]
            row["payload"] = json.loads(params[0])
            if "revision=revision+1" in normalized:
                row["revision"] += 1
                if "updater='upgrade-undo'" in normalized:
                    row["updater"] = "upgrade-undo"
                return _CardResult((row["revision"],))
            return _CardResult()
        if normalized.startswith("UPDATE muse_knowledge_draft SET deleted=TRUE"):
            row = self.drafts[params[0]]
            row["deleted"] = True
            if "revision=revision+1" in normalized:
                row["revision"] += 1
            if "updater='upgrade-undo'" in normalized:
                row["updater"] = "upgrade-undo"
            return _CardResult()
        if normalized.startswith("INSERT INTO example_upgrade_card_state"):
            did, _, watermark = params[:3]
            if "SET watermark_window=EXCLUDED.watermark_window" in normalized:
                self.states[did] = watermark
            else:
                self.states[did] = max(self.states.get(did, watermark), watermark)
            return _CardResult()
        if normalized.startswith("UPDATE example_upgrade_card_state SET watermark_window=%s"):
            self.states[params[1]] = params[0]
            return _CardResult()
        if normalized.startswith("DELETE FROM example_upgrade_card_state"):
            self.states.pop(params[0], None)
            return _CardResult()
        if normalized.startswith("DELETE FROM example_upgrade_audit"):
            window_no = params[1]
            self.audits = [item for item in self.audits if item["window_no"] != window_no]
            return _CardResult()
        return _CardResult(rows=[])


def test_relation_update_and_new_card_undo_boundaries():
    """关系更新按真实落点审计并可完整 undo；本窗新关系软删且清除 state。"""

    old_payload = {
        "type": pu.RELATION_TYPE,
        "名称": "甲×乙",
        "甲方draft": 11,
        "乙方draft": 12,
        "关系类型": "盟友",
        "字段": {"当前状态": "合作", "演变轨迹": ["[窗3] 初次合作"]},
        "来源": "升格@窗3",
        "_work_id": 8,
    }
    conn = _RelationConn({701: old_payload}, {701: 3})
    pu._update_relation_card(
        conn,
        8,
        8,
        701,
        0,
        {
            "关系类型": "对手",
            "本窗演变": "公开决裂",
            "其他字段": {"当前状态": "敌对", "信任基础": "破裂"},
        },
    )
    updated = conn.drafts[701]["payload"]
    check("relation-update-顶层关系类型落正确位置", updated["关系类型"] == "对手")
    check("relation-update-其他字段逐字段落库",
          updated["字段"]["当前状态"] == "敌对" and updated["字段"]["信任基础"] == "破裂")
    check("relation-update-演变轨迹带窗号", updated["字段"]["演变轨迹"][-1] == "[窗8] 公开决裂")
    audit_by_name = {item["field_name"]: item for item in conn.audits}
    check("relation-update-关系类型使用顶层sentinel",
          pu.TOP_LEVEL_RELATION_TYPE_AUDIT_FIELD in audit_by_name)
    check("relation-update-其他字段完整旧新审计",
          json.loads(audit_by_name["当前状态"]["old_value"]) == "合作"
          and json.loads(audit_by_name["当前状态"]["new_value"]) == "敌对"
          and audit_by_name["信任基础"]["old_value"] is None)
    check("relation-update-演变轨迹完整旧新审计",
          json.loads(audit_by_name["演变轨迹"]["old_value"]) == ["[窗3] 初次合作"]
          and json.loads(audit_by_name["演变轨迹"]["new_value"])[-1] == "[窗8] 公开决裂")
    check("relation-update-state升到本窗且留旧水位审计",
          conn.states[701] == 8
          and json.loads(audit_by_name[pu.CARD_STATE_WATERMARK_AUDIT_FIELD]["old_value"]) == 3)

    pu.undo_window(conn, 8, 8)
    check("relation-update-undo完整恢复payload", conn.drafts[701]["payload"] == old_payload)
    check("relation-update-undo恢复旧水位", conn.states[701] == 3)

    new_payload = {
        "type": pu.RELATION_TYPE,
        "名称": "甲×丙",
        "甲方draft": 11,
        "乙方draft": 13,
        "关系类型": "师徒",
        "字段": {"当前状态": "建立", "演变轨迹": ["[窗8] 首次指点"]},
        "来源": "升格@窗8",
        "_work_id": 8,
    }
    new_id = pu._insert_relation_card(conn, 8, 8, new_payload)
    new_audits = [item for item in conn.audits if item["draft_id"] == new_id]
    check("relation-new-取得RETURNING-id并建state", new_id in conn.drafts and conn.states[new_id] == 8)
    check("relation-new-不是单条占位审计",
          {item["field_name"] for item in new_audits}
          >= {pu.TOP_LEVEL_RELATION_TYPE_AUDIT_FIELD, "当前状态", "演变轨迹",
              pu.CARD_STATE_WATERMARK_AUDIT_FIELD})
    pu.undo_window(conn, 8, 8)
    check("relation-new-undo软删", conn.drafts[new_id]["deleted"] is True)
    check("relation-new-undo清除state", new_id not in conn.states)


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
        self.connection_id = self.state.setdefault("connection_seq", 0) + 1
        self.state["connection_seq"] = self.connection_id

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, tb):
        return False

    def execute(self, query, params=()):
        normalized = " ".join(query.split())
        if self.state.get("track_sql"):
            self.state.setdefault("sql_order", []).append(normalized)
        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, error_message FROM example_upgrade_window"):
            return _CardResult(rows=self.state.get("failed_windows", []))
        if normalized.startswith(
                "SELECT draft_payload, status, revision, source_type FROM muse_knowledge_draft"):
            draft_id = params[0]
            payload = self.state.get("payloads", {}).get(draft_id, self.state.get("payload"))
            revision = self.state.setdefault("revisions", {}).get(draft_id, 0)
            status = self.state.get("draft_status", "pending")
            source_type = self.state.get("draft_source_type", pu.SOURCE_TYPE)
            if self.state.get("bump_revision_after_lock"):
                self.state["revisions"][draft_id] = revision + 1
            return _CardResult((payload, status, revision, source_type))
        if normalized.startswith(
                "SELECT work_id, draft_payload, status, revision, source_type, deleted"):
            draft_id = params[0]
            payload = self.state.get("payloads", {}).get(draft_id, self.state.get("payload"))
            return _CardResult((
                8,
                payload,
                self.state.get("draft_status", "pending"),
                self.state.setdefault("revisions", {}).get(draft_id, 0),
                self.state.get("draft_source_type", pu.SOURCE_TYPE),
                False,
            ))
        if normalized.startswith("SELECT draft_payload FROM muse_knowledge_draft"):
            payload = self.state.get("payloads", {}).get(params[0], self.state.get("payload"))
            return _CardResult((payload,))
        if normalized.startswith("INSERT INTO muse_knowledge_draft"):
            draft_id = self.state.setdefault("next_draft_id", 101)
            self.state["next_draft_id"] = draft_id + 1
            self.state.setdefault("payloads", {})[draft_id] = json.loads(params[1])
            return _CardResult((draft_id,))
        if normalized.startswith("UPDATE muse_knowledge_draft SET draft_payload=%s"):
            self.state.setdefault("draft_update_sql", []).append(normalized)
            if "RETURNING revision" in normalized:
                draft_id = params[1]
                revisions = self.state.setdefault("revisions", {})
                expected_revision = params[-1]
                if self.state.get("draft_status", "pending") != "pending" \
                        or self.state.get("draft_source_type", pu.SOURCE_TYPE) != pu.SOURCE_TYPE \
                        or revisions.get(draft_id, 0) != expected_revision:
                    return _CardResult(None)
            payload = json.loads(params[0])
            if "payloads" in self.state:
                self.state["payloads"][params[1]] = payload
            else:
                self.state["payload"] = payload
            if "revision=revision+1" in normalized:
                revisions = self.state.setdefault("revisions", {})
                revisions[params[1]] = revisions.get(params[1], 0) + 1
                return _CardResult((revisions[params[1]],))
        if normalized.startswith("INSERT INTO example_upgrade_alias"):
            self.state.setdefault("alias_rows", []).append(params)
            return _CardResult((params[1],))
        if normalized.startswith("UPDATE example_upgrade_window SET status='failed'"):
            if "RETURNING status" in normalized and (
                    self.state.get("window_status") != "done"
                    or self.state.get("window_error") is not None
                    or self.state.get("window_updater", "upgrade") != "upgrade"):
                return _CardResult(None)
            self.state["window_status"] = "failed"
            self.state["window_error"] = params[0]
            self.state.setdefault("window_errors", []).append(params[0])
            self.state["window_updater"] = "upgrade"
            if self.state.get("track_events"):
                self.state.setdefault("events", []).append(("failed", self.connection_id))
            if "RETURNING status" in normalized:
                return _CardResult(("failed",))
        if normalized.startswith("UPDATE example_upgrade_window SET status='done'"):
            self.state["window_status"] = "done"
            self.state["window_error"] = None
            self.state["window_updater"] = "upgrade"
        if not normalized.startswith("SELECT"):
            self.writes.append((normalized, params))
        return _CardResult()

    def commit(self):
        """内存状态即时生效。"""
        self.commits += 1
        if self.state.get("track_events"):
            self.state.setdefault("events", []).append(("commit", self.connection_id))


def test_run_alias_paths_store_real_canonical_name():
    """疑似别名与 substring 命中 alias 键时，实际 INSERT 均写活跃卡真实名称。"""

    canonical = "一代机械外骨骼·铁壳"
    target = (701, "item", "一代机械外骨骼")
    state = {
        "payload": {
            "type": "item",
            "名称": canonical,
            "别名": ["铁壳"],
            "字段": {},
            "_work_id": 8,
        },
    }

    def fake_m3_json(prompt, model, need_keys, system=None):
        if need_keys == ("新名字", "已知实体新信息", "纯出场"):
            return ({
                "新名字": [
                    {"型": "item", "名称": "新铁壳", "疑似别名指向": "铁壳",
                     "出场章": [57], "字段": {}},
                    {"型": "item", "名称": "机械铁壳", "出场章": [58], "字段": {}},
                ],
                "已知实体新信息": [],
                "纯出场": [],
            }, {})
        if need_keys == ("更新",):
            return ({"更新": []}, {})
        raise AssertionError(f"不应出现的离线模型调用：{need_keys}")

    name_map = {canonical: target, "铁壳": target}
    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=(
                "新铁壳出现，机械铁壳随后启动。",
                {57: "新铁壳出现。", 58: "机械铁壳随后启动。"},
            )), \
            patch.object(pu, "load_known", return_value=(name_map, {}, {701: {"铁壳"}}, {701: 0})), \
            patch.object(pu, "m3_json", side_effect=fake_m3_json):
        pu._run(
            work_id=8,
            max_windows=0,
            max_calls=0,
            model="MiniMax-M3",
            redo_window=0,
            semantic_on=False,
        )
    check("alias-run-两条alias路径均执行", len(state.get("alias_rows", [])) == 2)
    check("alias-run-canonical_name始终是真实卡名",
          {params[1] for params in state["alias_rows"]} == {canonical}, detail=str(state["alias_rows"]))


def test_run_new_card_registers_normalized_name_and_aliases_same_window():
    """括号名立卡后，同窗后续 alias/substr 必须命中新 canonical 并写真实规范名。"""

    state = {"payloads": {}}

    def fake_m3_json(prompt, model, need_keys, system=None):
        if need_keys == ("新名字", "已知实体新信息", "纯出场"):
            return ({
                "新名字": [
                    {"型": "item", "名称": "白色游魂（无名侦察兵）", "出场章": [70, 71],
                     "字段": {}, "别名": []},
                    {"型": "item", "名称": "侦察兵甲", "疑似别名指向": "无名侦察兵",
                     "出场章": [70], "字段": {}},
                    {"型": "item", "名称": "无名侦察兵甲", "出场章": [71], "字段": {}},
                ],
                "已知实体新信息": [],
                "纯出场": [],
            }, {})
        if need_keys == ("更新",):
            return ({"更新": []}, {})
        raise AssertionError(f"不应出现的离线模型调用：{need_keys}")

    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, "m3_json", side_effect=fake_m3_json):
        pu._run(
            work_id=8,
            max_windows=0,
            max_calls=0,
            model="MiniMax-M3",
            redo_window=0,
            semantic_on=False,
        )
    candidate_rows = [
        params for params in state.get("alias_rows", [])
        if params[2] in {"侦察兵甲", "无名侦察兵甲"}
    ]
    check("new-card-index-同窗alias与substr均命中", len(candidate_rows) == 2, detail=str(state))
    check("new-card-index-canonical_name均为规范化名称",
          {params[1] for params in candidate_rows} == {"白色游魂"}, detail=str(candidate_rows))


def _run_duplicate_entity_update_case(updates):
    """执行同一实体的重复更新输出，并返回写入口调用和窗失败证据。"""

    draft_id = 24473
    state = {
        "window_status": "pending",
        "revisions": {draft_id: 3},
        "payload": {
            "type": "character", "名称": "安捷", "别名": [], "字段": {}, "_work_id": 8,
        },
    }
    contracts = _fake_contracts()
    contracts["character"]["字段"].extend([
        {"key": "阵营", "说明": "当前所属阵营"},
        {"key": "当前状态", "说明": "人物当前状态"},
    ])

    def fake_m3_json(prompt, model, need_keys, system=None):
        if need_keys == ("新名字", "已知实体新信息", "纯出场"):
            return ({
                "新名字": [],
                "已知实体新信息": [
                    {"名称": "安捷", "观察点": "安捷状态发生变化", "出场章": []},
                ],
                "纯出场": [],
            }, {})
        if need_keys == ("更新",):
            return ({"更新": deepcopy(updates)}, {})
        raise AssertionError(f"不应出现的离线模型调用：{need_keys}")

    with patch.object(pu.psycopg, "connect", side_effect=lambda *_: _RunConn(state)), \
            patch.object(pu, "load_entity_contracts", return_value=contracts), \
            patch.object(pu, "load_window_material", return_value=("安捷保持警戒。", {70: "安捷保持警戒。"})), \
            patch.object(pu, "load_known", return_value=(
                {"安捷": (draft_id, "character", "")}, {}, {}, {draft_id: 3},
            )), \
            patch.object(pu, "m3_json", side_effect=fake_m3_json), \
            patch.object(pu, "undo_window"), \
            patch.object(pu, "merge_card", wraps=pu.merge_card) as merge_mock:
        pu._run(8, 0, 0, "MiniMax-M3", 0, False)
    return state, merge_mock


def test_run_exact_and_disjoint_duplicate_entity_updates_merge_once():
    """相同项折叠、互补字段合并、别名稳定去重后，每个实体只写一次。"""

    exact = {
        "draft_id": 24473,
        "变更字段": {"阵营": "探索队"},
        "别名新增": ["小安", 7, "捷哥"],
    }
    disjoint = {
        "draft_id": 24473,
        "变更字段": {"当前状态": "保持警戒"},
        "别名新增": ["捷哥", None, "安队"],
    }
    state, merge_mock = _run_duplicate_entity_update_case([exact, deepcopy(exact), disjoint])
    check("entity-update-相同与互补重复只写一次", merge_mock.call_count == 1,
          detail=str(merge_mock.call_args_list))
    args, kwargs = merge_mock.call_args
    check("entity-update-互补字段按key合并",
          args[3] == {"阵营": "探索队", "当前状态": "保持警戒"}, detail=str(args[3]))
    check("entity-update-别名仅字符串稳定去重并集",
          args[4] == ["小安", "捷哥", "安队"], detail=str(args[4]))
    check("entity-update-仍使用模型读取时revision",
          kwargs.get("expected_revision") == 3 and state["revisions"][24473] == 4,
          detail=f"kwargs={kwargs}, revisions={state['revisions']}")
    check("entity-update-合并后整窗成功", state["window_status"] == "done")


def test_run_conflicting_duplicate_entity_updates_fail_before_any_write():
    """同一实体同一字段值冲突时，整批预检必须在 merge_card 与 alias 写入前稳定失败。"""

    state, merge_mock = _run_duplicate_entity_update_case([
        {"draft_id": 24473, "变更字段": {"阵营": "探索队"}, "别名新增": ["小安"]},
        {"draft_id": 24473, "变更字段": {"阵营": "敌方"}, "别名新增": ["安队"]},
    ])
    check("entity-update-冲突重复零merge写入", merge_mock.call_count == 0,
          detail=str(merge_mock.call_args_list))
    check("entity-update-冲突重复零alias写入", not state.get("alias_rows"),
          detail=str(state.get("alias_rows")))
    check("entity-update-冲突重复两次尝试错误稳定",
          len(state.get("window_errors", [])) == 2
          and len(set(state["window_errors"])) == 1,
          detail=str(state.get("window_errors")))
    check("entity-update-冲突重复整窗失败", state["window_status"] == "failed")


class _RelationRunConn:
    """复用关系内存库执行一个完整窗，验证关系 draft id 真进入提交后 touched。"""

    def __init__(self, db):
        self.db = db

    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, status"):
            return _CardResult(rows=[(8, 57, 60, "pending")])
        if normalized.startswith("SELECT window_no, error_message FROM example_upgrade_window"):
            return _CardResult(rows=[])
        if normalized.startswith("UPDATE example_upgrade_window SET status='failed'"):
            self.db.window_status = "failed"
            if not hasattr(self.db, "window_errors"):
                self.db.window_errors = []
            self.db.window_errors.append(params[0])
            return _CardResult()
        if normalized.startswith("UPDATE example_upgrade_window SET status='done'"):
            self.db.window_status = "done"
            return _CardResult()
        if normalized.startswith("SELECT id, draft_payload, revision FROM muse_knowledge_draft") \
                and "draft_payload->>'type'=%s" in normalized:
            return _CardResult(rows=[
                (did, row["payload"], row["revision"])
                for did, row in self.db.drafts.items()
                if not row["deleted"] and row["payload"].get("type") == pu.RELATION_TYPE
            ])
        return self.db.execute(query, params)

    def commit(self):
        """内存写即时生效，无需额外动作。"""


def test_run_relation_cards_enter_touched_embedding():
    """关系更新与新建都必须进入窗提交后的 touched，复用唯一活向量刷新链路。"""

    character_payloads = {
        11: {"type": "character", "名称": "甲", "字段": {}, "_work_id": 8},
        12: {"type": "character", "名称": "乙", "字段": {}, "_work_id": 8},
        13: {"type": "character", "名称": "丙", "字段": {}, "_work_id": 8},
    }
    relation_payload = {
        "type": pu.RELATION_TYPE,
        "名称": "甲×乙",
        "甲方draft": 11,
        "乙方draft": 12,
        "关系类型": "盟友",
        "字段": {"演变轨迹": ["[窗3] 初次合作"]},
        "来源": "升格@窗3",
        "_work_id": 8,
    }
    db = _RelationConn({**character_payloads, 701: relation_payload}, {701: 3})
    embedded = []

    def fake_m3_json(prompt, model, need_keys, system=None):
        if need_keys == ("新名字", "已知实体新信息", "纯出场"):
            return ({"新名字": [], "已知实体新信息": [], "纯出场": []}, {})
        if need_keys == ("关系",):
            return ({
                "关系": [
                    {"甲方": 11, "乙方": 12, "关系类型": "对手", "本窗演变": "公开决裂",
                     "其他字段": {"当前状态": "敌对"}},
                    {"甲方": 11, "乙方": 13, "关系类型": "师徒", "本窗演变": "首次指点",
                     "其他字段": {"当前状态": "建立"}},
                ]
            }, {})
        raise AssertionError(f"不应出现的离线模型调用：{need_keys}")

    name_map = {
        "甲": (11, "character", ""),
        "乙": (12, "character", ""),
        "丙": (13, "character", ""),
    }
    with patch.object(pu.psycopg, "connect", side_effect=lambda *_: _RelationRunConn(db)), \
            patch.object(pu, "load_entity_contracts", return_value=_fake_contracts()), \
            patch.object(pu, "load_window_material", return_value=(
                "甲与乙决裂，随后甲开始指点丙。",
                {57: "甲与乙共同出现。", 60: "甲开始指点丙。"},
            )), \
            patch.object(pu, "load_known", return_value=(name_map, {}, {}, {11: 0, 12: 0, 13: 0})), \
            patch.object(pu, "m3_json", side_effect=fake_m3_json), \
            patch.object(pu, "_embed_session", return_value=object()), \
            patch.object(pu, "prejudge_semantic", return_value=({}, 0, 0)), \
            patch.object(
                pu,
                "embed_touched_cards",
                side_effect=lambda sess, work_id, touched: embedded.append(set(touched)) or len(touched),
            ):
        pu._run(
            work_id=8,
            max_windows=0,
            max_calls=0,
            model="MiniMax-M3",
            redo_window=0,
            semantic_on=True,
        )
    check("relation-run-新关系取得id", 702 in db.drafts)
    check("relation-run-更新与新建关系均进入touched", embedded == [{701, 702}], detail=str(embedded))


def _run_duplicate_relation_case(relations):
    """执行同一无序人物对的重复关系输出，并返回关系写调用与稳定失败证据。"""

    characters = {
        11: {"type": "character", "名称": "甲", "字段": {}, "_work_id": 8},
        12: {"type": "character", "名称": "乙", "字段": {}, "_work_id": 8},
    }
    relation_payload = {
        "type": pu.RELATION_TYPE,
        "名称": "甲×乙",
        "甲方draft": 11,
        "乙方draft": 12,
        "关系类型": "盟友",
        "字段": {"演变轨迹": ["[窗3] 初次合作"]},
        "来源": "升格@窗3",
        "_work_id": 8,
    }
    db = _RelationConn({**characters, 701: relation_payload}, {701: 3})
    db.window_status = "pending"
    db.window_errors = []

    def fake_m3_json(prompt, model, need_keys, system=None):
        if need_keys == ("新名字", "已知实体新信息", "纯出场"):
            return ({"新名字": [], "已知实体新信息": [], "纯出场": []}, {})
        if need_keys == ("关系",):
            return ({"关系": deepcopy(relations)}, {})
        raise AssertionError(f"不应出现的离线模型调用：{need_keys}")

    name_map = {
        "甲": (11, "character", ""),
        "乙": (12, "character", ""),
    }
    with patch.object(pu.psycopg, "connect", side_effect=lambda *_: _RelationRunConn(db)), \
            patch.object(pu, "load_entity_contracts", return_value=_fake_contracts()), \
            patch.object(pu, "load_window_material", return_value=(
                "甲与乙的关系发生变化。", {57: "甲与乙的关系发生变化。"},
            )), \
            patch.object(pu, "load_known", return_value=(name_map, {}, {}, {11: 0, 12: 0})), \
            patch.object(pu, "m3_json", side_effect=fake_m3_json), \
            patch.object(pu, "undo_window"), \
            patch.object(pu, "_update_relation_card", wraps=pu._update_relation_card) as update_mock, \
            patch.object(pu, "_insert_relation_card", wraps=pu._insert_relation_card) as insert_mock:
        pu._run(8, 0, 0, "MiniMax-M3", 0, False)
    return db, update_mock.call_count, insert_mock.call_count


def test_run_conflicting_duplicate_relations_fail_before_any_relation_write():
    """同一无序实体对内容冲突时，整批预检必须在任何关系卡写入前稳定失败。"""

    db, update_calls, insert_calls = _run_duplicate_relation_case([
        {"甲方": 11, "乙方": 12, "关系类型": "盟友", "本窗演变": "继续合作",
         "其他字段": {"当前状态": "互相信任"}},
        {"甲方": 12, "乙方": 11, "关系类型": "对手", "本窗演变": "公开决裂",
         "其他字段": {"当前状态": "彼此敌对"}},
    ])
    check("relation-冲突重复零关系写入", update_calls == 0 and insert_calls == 0,
          detail=f"update={update_calls}, insert={insert_calls}")
    check("relation-冲突重复两次尝试错误稳定",
          len(db.window_errors) == 2 and len(set(db.window_errors)) == 1,
          detail=str(db.window_errors))
    check("relation-冲突重复整窗失败", db.window_status == "failed")


def test_run_identical_duplicate_relations_write_existing_card_once():
    """完全相同的反向重复项规范成同一无序对，既有关系卡最多更新一次。"""

    first = {"甲方": 11, "乙方": 12, "关系类型": "盟友", "本窗演变": "继续合作",
             "其他字段": {"当前状态": "互相信任"}}
    reversed_duplicate = {**deepcopy(first), "甲方": 12, "乙方": 11}
    db, update_calls, insert_calls = _run_duplicate_relation_case([
        {"甲方": 11, "乙方": 11, "关系类型": "自关系"},
        {"甲方": 11, "乙方": 999, "关系类型": "窗外角色"},
        first,
        reversed_duplicate,
    ])
    check("relation-相同重复既有卡只更新一次", update_calls == 1 and insert_calls == 0,
          detail=f"update={update_calls}, insert={insert_calls}")
    check("relation-相同重复整窗成功", db.window_status == "done",
          detail=str(db.window_errors))
    check("relation-相同重复只推进一次revision", db.drafts[701]["revision"] == 1,
          detail=str(db.drafts[701]))


def test_run_prompt_revision_rejects_entity_and_relation_stale_results():
    """实体与关系模型调用期间 revision 变化时，旧模型结果都不得覆盖外部新值。"""

    entity_state = {
        "window_status": "pending",
        "revisions": {701: 0},
        "payload": {
            "type": "character", "名称": "安若雪", "字段": {}, "_work_id": 8,
        },
    }

    def entity_m3(prompt, model, need_keys, system=None):
        if need_keys == ("新名字", "已知实体新信息", "纯出场"):
            return ({"新名字": [], "已知实体新信息": [
                {"名称": "安若雪", "观察点": "模型观察", "出场章": []}
            ], "纯出场": []}, {})
        if need_keys == ("更新",):
            entity_state["payload"]["字段"]["外部改版"] = "保留"
            entity_state["revisions"][701] += 1
            return ({"更新": [{
                "draft_id": 701, "变更字段": {"阵营": "模型旧结果"}, "别名新增": []
            }]}, {})
        raise AssertionError(f"不应出现的离线模型调用：{need_keys}")

    with patch.object(pu.psycopg, "connect", side_effect=lambda *_: _RunConn(entity_state)), \
            patch.object(pu, "load_entity_contracts", return_value=_fake_contracts()), \
            patch.object(pu, "load_window_material", return_value=("安若雪出现。", {70: "正文"})), \
            patch.object(pu, "load_known", return_value=(
                {"安若雪": (701, "character", "")}, {}, {}, {701: 0},
            )), \
            patch.object(pu, "m3_json", side_effect=entity_m3), \
            patch.object(pu, "undo_window"):
        pu._run(8, 0, 0, "MiniMax-M3", 0, False)
    check("prompt-revision-实体外部改版保留",
          entity_state["payload"]["字段"] == {"外部改版": "保留"})
    check("prompt-revision-实体旧模型结果令窗失败", entity_state["window_status"] == "failed")

    chars = {
        11: {"type": "character", "名称": "甲", "字段": {}, "_work_id": 8},
        12: {"type": "character", "名称": "乙", "字段": {}, "_work_id": 8},
    }
    relation_payload = {
        "type": pu.RELATION_TYPE, "名称": "甲×乙", "甲方draft": 11, "乙方draft": 12,
        "关系类型": "盟友", "字段": {"当前状态": "旧值"}, "_work_id": 8,
    }
    relation_db = _RelationConn({**chars, 701: relation_payload}, {701: 3})

    def relation_m3(prompt, model, need_keys, system=None):
        if need_keys == ("新名字", "已知实体新信息", "纯出场"):
            return ({"新名字": [], "已知实体新信息": [], "纯出场": []}, {})
        if need_keys == ("关系",):
            row = relation_db.drafts[701]
            row["payload"]["字段"]["外部改版"] = "保留"
            row["revision"] += 1
            return ({"关系": [{
                "甲方": 11, "乙方": 12, "关系类型": "对手",
                "本窗演变": "模型旧结果", "其他字段": {"当前状态": "模型覆盖"},
            }]}, {})
        raise AssertionError(f"不应出现的离线模型调用：{need_keys}")

    with patch.object(pu.psycopg, "connect", side_effect=lambda *_: _RelationRunConn(relation_db)), \
            patch.object(pu, "load_entity_contracts", return_value=_fake_contracts()), \
            patch.object(pu, "load_window_material", return_value=("甲与乙出现。", {57: "甲与乙出现。"})), \
            patch.object(pu, "load_known", return_value=(
                {"甲": (11, "character", ""), "乙": (12, "character", "")},
                {}, {}, {11: 0, 12: 0},
            )), \
            patch.object(pu, "m3_json", side_effect=relation_m3), \
            patch.object(pu, "undo_window"):
        pu._run(8, 0, 0, "MiniMax-M3", 0, False)
    relation_fields = relation_db.drafts[701]["payload"]["字段"]
    check("prompt-revision-关系外部改版保留",
          relation_fields == {"当前状态": "旧值", "外部改版": "保留"})
    check("prompt-revision-关系零审计副作用", not relation_db.audits)


def test_relation_participants_lock_before_prompt_and_reject_external_drift():
    """关系参与角色在 prompt 前逐卡锁定；确认、删除或改版后不得调用关系模型。"""

    characters = {
        11: {"type": "character", "名称": "甲", "字段": {}, "_work_id": 8},
        12: {"type": "character", "名称": "乙", "字段": {}, "_work_id": 8},
    }
    for name in ("revision改版", "外部确认", "外部删除"):
        db = _RelationConn(characters)
        relation_calls = []

        def drift_before_lock(draft_id):
            if draft_id != 12:
                return
            row = db.drafts[12]
            if name == "revision改版":
                row["payload"]["字段"]["外部改版"] = "保留"
                row["revision"] += 1
            elif name == "外部确认":
                row["status"] = "confirmed"
            else:
                row["deleted"] = True

        db.before_lock = drift_before_lock

        def fake_m3_json(prompt, model, need_keys, system=None):
            if need_keys == ("新名字", "已知实体新信息", "纯出场"):
                return ({"新名字": [], "已知实体新信息": [], "纯出场": []}, {})
            if need_keys == ("关系",):
                relation_calls.append(prompt)
                return ({"关系": [{"甲方": 11, "乙方": 12, "关系类型": "盟友"}]}, {})
            raise AssertionError(f"不应出现的离线模型调用：{need_keys}")

        with patch.object(pu.psycopg, "connect", side_effect=lambda *_: _RelationRunConn(db)), \
                patch.object(pu, "load_entity_contracts", return_value=_fake_contracts()), \
                patch.object(pu, "load_window_material", return_value=("甲与乙出现。", {57: "甲与乙出现。"})), \
                patch.object(pu, "load_known", return_value=(
                    {"甲": (11, "character", ""), "乙": (12, "character", "")},
                    {}, {}, {11: 0, 12: 0},
                )), \
                patch.object(pu, "m3_json", side_effect=fake_m3_json), \
                patch.object(pu, "undo_window"):
            pu._run(8, 0, 0, "MiniMax-M3", 0, False)
        check(f"relation-participant-{name}-prompt前失败", not relation_calls)
        check(f"relation-participant-{name}-不新建关系",
              set(db.drafts) == {11, 12} and not db.audits)


def test_run_alias_gate_precedes_empty_update_and_deleted_alias_fails_closed():
    """alias 分类先校验目标版本；唯一 tombstone 即使同 canonical 也不可视为成功。"""

    state = {
        "window_status": "pending",
        "revisions": {701: 0},
        "payload": {"type": "item", "名称": "铁壳", "字段": {}, "_work_id": 8},
    }

    def fake_m3(prompt, model, need_keys, system=None):
        if need_keys == ("新名字", "已知实体新信息", "纯出场"):
            state["revisions"][701] += 1
            return ({"新名字": [{
                "型": "item", "名称": "新铁壳", "疑似别名指向": "铁壳",
                "出场章": [70], "字段": {},
            }], "已知实体新信息": [], "纯出场": []}, {})
        if need_keys == ("更新",):
            return ({"更新": []}, {})
        raise AssertionError(f"不应出现的离线模型调用：{need_keys}")

    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: "新铁壳出现。"})), \
            patch.object(pu, "load_known", return_value=(
                {"铁壳": (701, "item", "")}, {}, {}, {701: 0},
            )), \
            patch.object(pu, "m3_json", side_effect=fake_m3), \
            patch.object(pu, "undo_window"):
        pu._run(8, 0, 0, "MiniMax-M3", 0, False)
    check("alias-gate-空更新前版本冲突不写alias", not state.get("alias_rows"))
    check("alias-gate-版本冲突窗失败", state["window_status"] == "failed")

    tombstone = _IntegrityConn(
        {"type": "character", "名称": "安若雪", "字段": {}, "_work_id": 8},
        aliases={"雪姐": ("安若雪", 2)},
    )
    tombstone.aliases["雪姐"]["deleted"] = True
    try:
        pu.merge_card(tombstone, 701, 8, {}, ["雪姐"], expected_revision=4)
    except pu.AliasOwnershipConflict as exc:
        check("alias-tombstone-同canonical仍失败关闭", "deleted" in str(exc))
    else:
        check("alias-tombstone-同canonical仍失败关闭", False, detail="tombstone 被当作幂等成功")
    check("alias-tombstone-不复活不写payload",
          tombstone.aliases["雪姐"]["deleted"] is True
          and "别名" not in tombstone.drafts[701]["payload"])


def test_run_embedding_owner_conflict_marks_window_failed():
    """当前新 hash 无 owner 时，同 draft 旧活向量的 entity owner 仍必须阻断且保持活跃。"""

    state = {
        "window_status": "pending",
        "track_events": True,
        "payload": {
            "type": "character",
            "名称": "安若雪",
            "别名": [],
            "字段": {},
            "_work_id": 8,
        },
        "artifacts": {
            "drafts": [701],
            "states": [701],
            "audits": [701],
            "aliases": ["雪姐"],
            "presence": ["窗7出场"],
        },
    }

    def fake_m3_json(prompt, model, need_keys, system=None):
        if need_keys == ("新名字", "已知实体新信息", "纯出场"):
            return ({
                "新名字": [],
                "已知实体新信息": [{"名称": "安若雪", "观察点": "状态变化", "出场章": [70]}],
                "纯出场": [],
            }, {})
        if need_keys == ("更新",):
            return ({"更新": []}, {})
        raise AssertionError(f"不应出现的离线模型调用：{need_keys}")

    embed_text = "当前新向量文本"
    old_embed_text = "同 draft 旧向量文本"
    old_content_hash = hashlib.sha256(f"{old_embed_text}|{pu.EMBED_MODEL}".encode()).hexdigest()
    embedding_state = {
        "drafts": {
            701: {"payload": deepcopy(state["payload"]), "deleted": False},
        },
        "embeddings": [{
            "draft_id": 701,
            "entity_id": 9001,
            "content_hash": old_content_hash,
            "embed_text": old_embed_text,
            "model": pu.EMBED_MODEL,
            "dimensions": pu.EMBED_DIM,
            "embedding": "[0.3]",
            "deleted": False,
        }],
    }
    embed_calls = []
    real_embed_touched_cards = pu.embed_touched_cards

    def compensate_undo(conn, work_id, win_no):
        """模拟真实 undo 的五域清理，并记录必须发生在 failed 更新之前的连接身份。"""
        state.setdefault("events", []).append(("undo", conn.connection_id))
        for values in state["artifacts"].values():
            values.clear()

    def run_real_embed(sess, work_id, touched):
        with patch.object(pu.psycopg, "connect", side_effect=lambda *_: _EmbeddingConn(embedding_state)), \
                patch.object(pu, "build_embed_text", return_value=embed_text), \
                patch.object(
                    pu,
                    "embed_texts",
                    side_effect=lambda embed_sess, texts: embed_calls.append(list(texts)) or ([[0.9]], set()),
                ):
            return real_embed_touched_cards(sess, work_id, touched)

    with patch.object(pu, "upgrade_work_lock", return_value=nullcontext()), \
            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", "旧摘要")}, {}, {}, {701: 0}),
            ), \
            patch.object(pu, "m3_json", side_effect=fake_m3_json), \
            patch.object(pu, "_embed_session", return_value=object()), \
            patch.object(pu, "prejudge_semantic", return_value=({}, 0, 0)), \
            patch.object(pu, "undo_window", side_effect=compensate_undo), \
            patch.object(
                pu,
                "embed_touched_cards",
                side_effect=run_real_embed,
            ):
        result = CliRunner().invoke(pu.cli, ["run", "--work-id", "8", "--semantic-dedup"])
    check("embed-conflict-run-命令非零", result.exit_code != 0, detail=result.output)
    check("embed-conflict-run-status补偿为failed", state["window_status"] == "failed")
    compensation = state.get("events", [])[-3:]
    check(
        "embed-conflict-run-同事务先undo再failed后commit",
        [event for event, _ in compensation] == ["undo", "failed", "commit"]
        and len({conn_id for _, conn_id in compensation}) == 1,
        detail=str(state.get("events")),
    )
    check(
        "embed-conflict-run-五域无已提交窗残留",
        all(not values for values in state["artifacts"].values()),
        detail=str(state["artifacts"]),
    )
    check("embed-conflict-run-不打印窗完成", "窗7✓" not in result.output, detail=result.output)
    check("embed-old-entity-owner-新向量计算已发生", embed_calls == [[embed_text]])
    check("embed-old-entity-owner-保持活跃与绑定",
          embedding_state["embeddings"][0]["draft_id"] == 701
          and embedding_state["embeddings"][0]["entity_id"] == 9001
          and embedding_state["embeddings"][0]["deleted"] is False)


def _run_owner_conflict_case(state, *, redo_window=0, snapshot=None,
                             postcommit_snapshot=None, current_snapshot=None,
                             undo_effect=None, restore_effect=None):
    """执行一个完整窗直到嵌入 owner 冲突，供补偿边界反例复用。"""

    def fake_m3_json(prompt, model, need_keys, system=None):
        if need_keys == ("新名字", "已知实体新信息", "纯出场"):
            return ({
                "新名字": [],
                "已知实体新信息": [
                    {"名称": "安若雪", "观察点": "状态变化", "出场章": []}
                ],
                "纯出场": [],
            }, {})
        if need_keys == ("更新",):
            return ({"更新": []}, {})
        raise AssertionError(f"不应出现的离线模型调用：{need_keys}")

    def owner_conflict(sess, work_id, touched):
        callback = state.get("before_owner_conflict")
        if callback:
            callback()
        raise pu.EmbeddingOwnershipConflict("离线 owner 冲突")

    cli_args = ["run", "--work-id", "8", "--semantic-dedup"]
    if redo_window:
        cli_args.extend(["--redo-window", str(redo_window)])
    snapshot_calls = {"n": 0}

    def fake_snapshot(conn, work_id, win_no):
        snapshot_calls["n"] += 1
        if snapshot_calls["n"] == 1:
            return deepcopy(snapshot)
        if snapshot_calls["n"] == 2:
            return deepcopy(postcommit_snapshot if postcommit_snapshot is not None else snapshot)
        current = deepcopy(
            current_snapshot if current_snapshot is not None
            else postcommit_snapshot if postcommit_snapshot is not None
            else snapshot
        )
        # marker-first 会在补偿围栏重读前先改变窗口；fake 同步反映该受控变化。
        if isinstance(current, dict) and "window" in current:
            current["window"] = (
                state.get("window_status"), state.get("window_error"), "upgrade"
            )
        return current

    with patch.object(pu, "upgrade_work_lock", return_value=nullcontext()), \
            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", "旧摘要")}, {}, {}, {701: 0},
            )), \
            patch.object(pu, "m3_json", side_effect=fake_m3_json), \
            patch.object(pu, "_embed_session", return_value=object()), \
            patch.object(pu, "prejudge_semantic", return_value=({}, 0, 0)), \
            patch.object(pu, "embed_touched_cards", side_effect=owner_conflict), \
            patch.object(pu, "_snapshot_redo_window", side_effect=fake_snapshot), \
            patch.object(pu, "undo_window", side_effect=undo_effect) as undo, \
            patch.object(pu, "_restore_redo_window", side_effect=restore_effect) as restore:
        result = CliRunner().invoke(pu.cli, cli_args)
    return result, undo, restore


def test_owner_conflict_external_confirmation_preserves_draft_and_persists_breakpoint():
    """嵌入冲突前被外部确认/改版的 touched 卡绝不 undo，并持久化补偿失败断点。"""

    original_payload = {
        "type": "character", "名称": "安若雪", "字段": {}, "_work_id": 8,
    }
    state = {
        "window_status": "pending",
        "payload": deepcopy(original_payload),
        "revisions": {701: 0},
    }

    def confirm_externally():
        state["draft_status"] = "confirmed"
        state["revisions"][701] = 1
        state["payload"]["字段"]["人工确认"] = "必须保留"

    state["before_owner_conflict"] = confirm_externally
    result, undo, _ = _run_owner_conflict_case(state)
    check("compensation-fence-外部确认后命令非零", result.exit_code != 0, detail=result.output)
    check("compensation-fence-绝不undo或软删外部卡", not undo.called
          and state["draft_status"] == "confirmed"
          and state["payload"]["字段"] == {"人工确认": "必须保留"})
    check("compensation-fence-窗口持久failed断点",
          state["window_status"] == "failed"
          and str(state.get("window_error", "")).startswith(pu.COMPENSATION_FAILED_PREFIX),
          detail=str(state))


def test_redo_owner_conflict_restores_snapshot_without_normal_undo():
    """显式 redo 的嵌入冲突恢复 redo 前五域和窗状态，不进入普通 undo+failed。"""

    old_domains = {
        "fields": {"阵营": "旧阵营"},
        "aliases": ["旧别名"],
        "presence": ["旧出场"],
        "states": {701: 6},
        "audits": ["旧审计"],
    }
    snapshot = {
        "window_no": 1,
        "domains": deepcopy(old_domains),
        "window": ("done", None, "old"),
    }
    state = {
        "windows": [(1, 70, 71)],
        "window_status": "done",
        "payload": {"type": "character", "名称": "安若雪", "字段": {}, "_work_id": 8},
        "revisions": {701: 0},
        **deepcopy(old_domains),
    }
    undo_calls = []
    restore_calls = []

    def initial_redo_undo(conn, work_id, win_no, *, from_chapter=None, to_chapter=None):
        undo_calls.append((win_no, from_chapter, to_chapter))
        for key in old_domains:
            state[key] = {} if isinstance(old_domains[key], dict) else []

    def restore_snapshot(conn, work_id, redo_snapshot):
        restore_calls.append(redo_snapshot)
        state.update(deepcopy(redo_snapshot["domains"]))
        state["window_status"] = redo_snapshot["window"][0]
        state["window_error"] = redo_snapshot["window"][1]

    result, _, _ = _run_owner_conflict_case(
        state,
        redo_window=1,
        snapshot=snapshot,
        undo_effect=initial_redo_undo,
        restore_effect=restore_snapshot,
    )
    check("redo-owner-conflict-命令非零且调用restore", result.exit_code != 0
          and restore_calls == [snapshot], detail=result.output)
    check("redo-owner-conflict-仅初始章域undo不走普通补偿undo",
          undo_calls == [(1, 70, 71)], detail=str(undo_calls))
    check("redo-owner-conflict-恢复原五域和窗口状态",
          all(state[key] == value for key, value in old_domains.items())
          and state["window_status"] == "done" and state["window_error"] is None,
          detail=str(state))


def test_redo_full_fence_rejects_untouched_and_auxiliary_domain_drift():
    """redo 补偿覆盖的任一整书域漂移都必须阻断 restore，并保留 durable failed。"""

    postcommit = {
        "window_no": 1,
        "drafts": [
            (701, {"名称": "安若雪", "字段": {}}, "pending", 0, "upgrade", False),
            (702, {"名称": "未参与角色", "字段": {"状态": "原值"}},
             "pending", 4, "old", False),
        ],
        "aliases": [(801, "安若雪", "雪姐", 1, "ai", "", "t1", "", "t2", False)],
        "presence": [(901, 1, 70, "character", "路人", "出现", "", "t3", False)],
        "card_states": [(701, 1, "t4"), (702, 0, "t4")],
        "audits": [(1001, 701, 1, "阵营", None, '"联邦"', "t5")],
        "window": ("done", None, "upgrade"),
    }
    cases = {}
    for domain in ("drafts", "aliases", "presence", "card_states", "audits"):
        drifted = deepcopy(postcommit)
        if domain == "drafts":
            # 702 未进入 touched；payload 与 revision 同时模拟外部 CAS 改版。
            drifted[domain][1] = (
                702, {"名称": "未参与角色", "字段": {"状态": "外部改版"}},
                "pending", 5, "external", False,
            )
        else:
            drifted[domain].append(tuple([999] + ["external"] * (len(drifted[domain][0]) - 1)))
        cases[domain] = drifted
    status_drift = deepcopy(postcommit)
    untouched = status_drift["drafts"][1]
    status_drift["drafts"][1] = (
        untouched[0], untouched[1], "confirmed", *untouched[3:]
    )
    cases["draft_status"] = status_drift

    for domain, current in cases.items():
        state = {
            "windows": [(1, 70, 71)],
            "window_status": "done",
            "track_sql": True,
            "payload": {"type": "character", "名称": "安若雪", "字段": {}, "_work_id": 8},
            "revisions": {701: 0},
        }
        result, undo, restore = _run_owner_conflict_case(
            state,
            redo_window=1,
            snapshot=deepcopy(postcommit),
            postcommit_snapshot=deepcopy(postcommit),
            current_snapshot=current,
            undo_effect=lambda *args, **kwargs: None,
        )
        check(f"redo-full-fence-{domain}-阻断restore",
              result.exit_code != 0 and not restore.called and undo.call_count == 1,
              detail=result.output)
        check(f"redo-full-fence-{domain}-durable-failed",
              state["window_status"] == "failed"
              and str(state.get("window_error", "")).startswith(pu.COMPENSATION_FAILED_PREFIX),
              detail=str(state))
        lock_sql = next((sql for sql in state.get("sql_order", [])
                         if sql.startswith("LOCK TABLE muse_knowledge_draft")), "")
        lock_tables = (
            "muse_knowledge_draft", "example_upgrade_alias", "example_upgrade_presence",
            "example_upgrade_card_state", "example_upgrade_audit", "example_upgrade_window",
        )
        check(f"redo-full-fence-{domain}-固定六域阻写顺序",
              all(name in lock_sql for name in lock_tables)
              and [lock_sql.index(name) for name in lock_tables]
              == sorted(lock_sql.index(name) for name in lock_tables),
              detail=lock_sql)


def test_compensation_exception_persists_durable_failed_and_next_run_stops():
    """undo/restore 异常后另事务落 failed 断点，下一次 run 必须显式阻断而非跳过。"""

    for name, redo_window in (("undo异常", 0), ("restore异常", 1)):
        state = {
            "window_status": "pending",
            "track_events": True,
            "payload": {"type": "character", "名称": "安若雪", "字段": {}, "_work_id": 8},
            "revisions": {701: 0},
        }
        if redo_window:
            state["windows"] = [(1, 70, 71)]

        def fail_compensation(*args, **kwargs):
            raise RuntimeError(f"{name}离线故障")

        result, undo, restore = _run_owner_conflict_case(
            state,
            redo_window=redo_window,
            snapshot={"window_no": 1, "window": ("done", None, "old")},
            undo_effect=None if redo_window else fail_compensation,
            restore_effect=fail_compensation if redo_window else None,
        )
        error_message = str(state.get("window_error", ""))
        check(f"compensation-durable-{name}-另事务持久failed",
              result.exit_code != 0 and state["window_status"] == "failed"
              and error_message.startswith(pu.COMPENSATION_FAILED_PREFIX),
              detail=f"output={result.output!r}, state={state!r}")
        check(f"compensation-durable-{name}-确实经过目标失败点",
              restore.called if redo_window else undo.called)
        compensation_conn = (restore if redo_window else undo).call_args.args[0]
        durable_events = state.get("events", [])[-2:]
        check(f"compensation-durable-{name}-failed另事务提交",
              [event for event, _ in durable_events] == ["failed", "commit"]
              and durable_events[0][1] == durable_events[1][1]
              and durable_events[0][1] != compensation_conn.connection_id,
              detail=str(state.get("events")))

        state["failed_windows"] = [(1 if redo_window else 7, error_message)]
        with patch.object(pu, "upgrade_work_lock", return_value=nullcontext()), \
                patch.object(pu.psycopg, "connect", side_effect=lambda *_: _RunConn(state)), \
                patch.object(pu, "load_entity_contracts", return_value=_fake_contracts()), \
                patch.object(pu, "undo_window") as next_undo:
            next_result = CliRunner().invoke(pu.cli, ["run", "--work-id", "8"])
        check(f"compensation-durable-{name}-下次run不跳过",
              next_result.exit_code != 0 and pu.COMPENSATION_FAILED_PREFIX in next_result.output
              and not next_undo.called,
              detail=next_result.output)


def test_run_embedding_payload_drift_marks_window_failed_without_vector_writes():
    """HTTP 期间 payload 漂移时，写段必须锁行复验并在任何向量 UPDATE/UPSERT 前阻断。"""

    state = {
        "window_status": "pending",
        "payload": {
            "type": "character", "名称": "安若雪", "别名": [], "字段": {}, "_work_id": 8,
        },
    }
    embedding_state = {
        "drafts": {701: {
            "payload": deepcopy(state["payload"]), "deleted": False,
            "tenant_id": pu.TENANT, "status": "pending",
        }},
        "embeddings": [],
    }
    real_embed_touched_cards = pu.embed_touched_cards

    def fake_m3_json(prompt, model, need_keys, system=None):
        if need_keys == ("新名字", "已知实体新信息", "纯出场"):
            return ({"新名字": [], "已知实体新信息": [
                {"名称": "安若雪", "观察点": "状态变化", "出场章": [70]}
            ], "纯出场": []}, {})
        if need_keys == ("更新",):
            return ({"更新": []}, {})
        raise AssertionError(f"不应出现的离线模型调用：{need_keys}")

    def run_real_embed(sess, work_id, touched):
        def drift_during_http(embed_sess, texts):
            embedding_state["drafts"][701]["payload"]["字段"]["HTTP后漂移"] = "新值"
            return ([[0.7]], set())

        with patch.object(pu.psycopg, "connect", side_effect=lambda *_: _EmbeddingConn(embedding_state)), \
                patch.object(pu, "embed_texts", side_effect=drift_during_http):
            return real_embed_touched_cards(sess, work_id, touched)

    with patch.object(pu, "upgrade_work_lock", return_value=nullcontext()), \
            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", "旧摘要")}, {}, {}, {701: 0},
            )), \
            patch.object(pu, "m3_json", side_effect=fake_m3_json), \
            patch.object(pu, "_embed_session", return_value=object()), \
            patch.object(pu, "prejudge_semantic", return_value=({}, 0, 0)), \
            patch.object(pu, "embed_touched_cards", side_effect=run_real_embed):
        result = CliRunner().invoke(pu.cli, ["run", "--work-id", "8", "--semantic-dedup"])

    vector_writes = [sql for sql in embedding_state.get("sql_order", [])
                     if sql.startswith("UPDATE example_knowledge_embedding")
                     or sql.startswith("INSERT INTO example_knowledge_embedding")]
    check("embed-payload-drift-命令非零且窗failed",
          result.exit_code != 0 and state["window_status"] == "failed", detail=result.output)
    check("embed-payload-drift-无窗完成", "窗7✓" not in result.output, detail=result.output)
    check("embed-payload-drift-零向量写入", not vector_writes, detail=str(vector_writes))


class _EmbeddingConn:
    """增量嵌入 fake DB：复现旧软删 draft 占用同 hash 唯一键的冲突行。"""

    def __init__(self, state):
        self.state = state

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, tb):
        return False

    def execute(self, query, params=()):
        normalized = " ".join(query.split())
        self.state.setdefault("sql_order", []).append(normalized)
        if normalized.startswith("SELECT tenant_id, deleted, status, draft_payload"):
            draft = self.state["drafts"].get(params[0])
            if not draft:
                return _CardResult(None)
            return _CardResult((
                draft.get("tenant_id", pu.TENANT),
                draft["deleted"],
                draft.get("status", "pending"),
                draft["payload"],
            ))
        self.state.setdefault("sql_order", []).append(normalized)
        if normalized.startswith("SELECT draft_payload FROM muse_knowledge_draft"):
            draft = self.state["drafts"].get(params[0])
            if not draft or draft["deleted"]:
                return _CardResult(None)
            return _CardResult((draft["payload"],))
        if normalized.startswith("SELECT content_hash FROM example_knowledge_embedding"):
            rows = [
                (row["content_hash"],)
                for row in self.state["embeddings"]
                if row["draft_id"] == params[1] and not row["deleted"]
            ]
            return _CardResult(rows=rows)
        if normalized.startswith("SELECT id, content_hash, entity_id FROM example_knowledge_embedding"):
            rows = [
                (index + 1, row["content_hash"], row.get("entity_id"))
                for index, row in enumerate(self.state["embeddings"])
                if row["draft_id"] == params[1] and not row["deleted"]
            ]
            return _CardResult(rows=rows)
        if normalized.startswith("SELECT e.draft_id"):
            content_hash = params[1]
            row = next(
                (item for item in self.state["embeddings"] if item["content_hash"] == content_hash),
                None,
            )
            if row is None:
                return _CardResult(None)
            owner = self.state["drafts"].get(row["draft_id"])
            owner_values = (
                row["draft_id"],
                row["deleted"],
                owner["deleted"] if owner else True,
                owner["payload"] if owner else None,
            )
            if "e.entity_id" in normalized:
                owner_values = (owner_values[0], row.get("entity_id"), *owner_values[1:])
            return _CardResult(owner_values)
        if normalized.startswith("UPDATE example_knowledge_embedding SET deleted=TRUE"):
            for row in self.state["embeddings"]:
                if row["draft_id"] == params[2] and not row["deleted"] \
                        and row["content_hash"] != params[3] \
                        and ("entity_id IS NULL" not in normalized or row.get("entity_id") is None):
                    row["deleted"] = True
            return _CardResult()
        if normalized.startswith("INSERT INTO example_knowledge_embedding"):
            draft_id, content_hash, embed_text, model, dimensions, embedding = params[:6]
            row = next(
                (item for item in self.state["embeddings"] if item["content_hash"] == content_hash),
                None,
            )
            is_new = row is None
            if is_new:
                row = {"entity_id": None}
                self.state["embeddings"].append(row)
            previous_owner = row.get("draft_id")
            previous_draft = self.state["drafts"].get(previous_owner)
            guarded = "WHERE example_knowledge_embedding.draft_id=EXCLUDED.draft_id" in normalized
            can_update = is_new or previous_owner == draft_id or row.get("deleted") \
                or not previous_draft or previous_draft["deleted"]
            if "example_knowledge_embedding.entity_id IS NULL" in normalized:
                can_update = can_update and row.get("entity_id") is None
            if guarded and not can_update:
                return _CardResult(None)
            # 按 SQL 文本模拟真实 ON CONFLICT：旧实现没有迁移 draft_id，测试应先 RED。
            if is_new or "draft_id=EXCLUDED.draft_id" in normalized:
                row["draft_id"] = draft_id
            row.update({
                "content_hash": content_hash,
                "embed_text": embed_text if "embed_text=EXCLUDED.embed_text" in normalized or is_new else row.get("embed_text"),
                "model": model,
                "dimensions": dimensions if "dimensions=EXCLUDED.dimensions" in normalized or is_new else row.get("dimensions"),
                "embedding": embedding if "embedding=EXCLUDED.embedding" in normalized or is_new else row.get("embedding"),
                "deleted": False,
            })
            return _CardResult((row["draft_id"],))
        return _CardResult()

    def commit(self):
        """fake DB 写入即时生效。"""


def test_embed_touched_migrates_soft_deleted_hash_owner():
    """全重抽新 draft 命中旧软删同 hash 时，唯一活向量必须迁到新 draft 并刷新当前列。"""

    payload = {"type": pu.RELATION_TYPE, "名称": "甲×乙", "字段": {"演变轨迹": ["首次合作"]}}
    embed_text = "当前关系卡嵌入文本"
    content_hash = hashlib.sha256(f"{embed_text}|{pu.EMBED_MODEL}".encode()).hexdigest()
    state = {
        "drafts": {
            701: {"payload": deepcopy(payload), "deleted": True},
            902: {"payload": deepcopy(payload), "deleted": False},
        },
        "embeddings": [{
            "draft_id": 701,
            "content_hash": content_hash,
            "embed_text": "旧文本",
            "model": pu.EMBED_MODEL,
            "dimensions": pu.EMBED_DIM,
            "embedding": "[0.1]",
            "deleted": True,
        }],
    }
    current_vector = [0.9, 0.8]
    with patch.object(pu.psycopg, "connect", side_effect=lambda *_: _EmbeddingConn(state)), \
            patch.object(pu, "build_embed_text", return_value=embed_text), \
            patch.object(pu, "embed_texts", return_value=([current_vector], set())):
        done = pu.embed_touched_cards(object(), 8, {902})
    active = [row for row in state["embeddings"] if not row["deleted"]]
    check("embed-conflict-只保留唯一活向量", done == 1 and len(active) == 1)
    check("embed-conflict-同hash迁到新draft", active[0]["draft_id"] == 902, detail=str(active))
    check("embed-conflict-同步当前嵌入列",
          active[0]["embed_text"] == embed_text
          and active[0]["dimensions"] == pu.EMBED_DIM
          and json.loads(active[0]["embedding"]) == current_vector)
    sql_order = state.get("sql_order", [])
    table_lock = next((i for i, sql in enumerate(sql_order)
                       if sql.startswith("LOCK TABLE muse_knowledge_draft, example_knowledge_embedding")), None)
    row_locks = [i for i, sql in enumerate(sql_order) if "FOR UPDATE" in sql]
    check("embed-write-固定表锁早于draft与hash行锁",
          table_lock is not None and row_locks and table_lock < min(row_locks), detail=str(sql_order))


def test_embed_touched_rejects_two_active_drafts_with_same_hash():
    """两个活跃 touched draft 当前文本同 hash 时必须失败关闭，禁止按循环顺序抢占唯一行。"""

    state = {
        "drafts": {
            801: {"payload": {"名称": "重复卡"}, "deleted": False},
            802: {"payload": {"名称": "重复卡"}, "deleted": False},
        },
        "embeddings": [],
    }
    embed_calls = []
    try:
        with patch.object(pu.psycopg, "connect", side_effect=lambda *_: _EmbeddingConn(state)), \
                patch.object(pu, "build_embed_text", return_value="完全相同文本"), \
                patch.object(
                    pu,
                    "embed_texts",
                    side_effect=lambda sess, texts: embed_calls.append(list(texts)) or ([[0.1], [0.1]], set()),
                ):
            pu.embed_touched_cards(object(), 8, {801, 802})
    except pu.EmbeddingOwnershipConflict:
        check("embed-duplicate-active-失败关闭", True)
    else:
        check("embed-duplicate-active-失败关闭", False, detail="同 hash 活跃 owner 冲突未抛硬错误")
    check("embed-duplicate-active-不调用嵌入不落错误owner", not embed_calls and not state["embeddings"])


def test_embed_touched_rejects_existing_active_hash_owner():
    """单个新候选撞到另一活跃 draft 的当前 hash 时，也不得迁移或复活成错误 owner。"""

    embed_text = "跨窗完全相同文本"
    content_hash = hashlib.sha256(f"{embed_text}|{pu.EMBED_MODEL}".encode()).hexdigest()
    state = {
        "drafts": {
            811: {"payload": {"名称": "重复卡"}, "deleted": False},
            812: {"payload": {"名称": "重复卡"}, "deleted": False},
        },
        "embeddings": [{
            "draft_id": 811,
            "content_hash": content_hash,
            "embed_text": embed_text,
            "model": pu.EMBED_MODEL,
            "dimensions": pu.EMBED_DIM,
            "embedding": "[0.2]",
            "deleted": False,
        }],
    }
    try:
        with patch.object(pu.psycopg, "connect", side_effect=lambda *_: _EmbeddingConn(state)), \
                patch.object(pu, "build_embed_text", return_value=embed_text), \
                patch.object(pu, "embed_texts", return_value=([[0.8]], set())):
            pu.embed_touched_cards(object(), 8, {812})
    except pu.EmbeddingOwnershipConflict:
        check("embed-existing-active-失败关闭", True)
    else:
        check("embed-existing-active-失败关闭", False, detail="已有活跃 owner 未阻断")
    check("embed-existing-active-owner保持不变",
          len(state["embeddings"]) == 1
          and state["embeddings"][0]["draft_id"] == 811
          and state["embeddings"][0]["deleted"] is False)


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, "upgrade_work_lock", return_value=nullcontext()), \
                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(
            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, status, revision, updater, deleted"):
            return _CardResult(rows=[
                (701, {"字段": {"阵营": "旧阵营"}}, "confirmed", 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-捕获旧卡status", snapshot["drafts"][0][2] == "confirmed")
    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-恢复旧卡字段status与revision",
        any(sql.startswith("UPDATE muse_knowledge_draft SET draft_payload=%s, status=%s, revision=%s")
            and params[:4] == (
                json.dumps({"字段": {"阵营": "旧阵营"}}, ensure_ascii=False),
                "confirmed", 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_pure_presence_rejects_external_state_and_revision_conflicts():
    """窗尾纯出场必须锁当前 draft；外部终态或 revision 漂移都不得覆盖 payload。"""

    original_payload = {
        "type": "character",
        "名称": "安若雪",
        "别名": [],
        "字段": {},
        "出场章": [],
        "_work_id": 8,
    }

    def fake_m3_json(prompt, model, need_keys, system=None):
        if need_keys == ("新名字", "已知实体新信息", "纯出场"):
            return ({
                "新名字": [],
                "已知实体新信息": [],
                "纯出场": [{"名称": "安若雪", "出场章": [70]}],
            }, {})
        raise AssertionError(f"不应出现的离线模型调用：{need_keys}")

    for name, extra_state in (
            ("外部状态终态", {"draft_status": "confirmed"}),
            ("外部revision漂移", {"revisions": {701: 0}})):
        state = {
            "window_status": "pending",
            "payload": deepcopy(original_payload),
            **extra_state,
        }
        def drift_during_observe(prompt, model, need_keys, system=None):
            if need_keys == ("新名字", "已知实体新信息", "纯出场"):
                if name == "外部revision漂移":
                    state["revisions"][701] += 1
                return fake_m3_json(prompt, model, need_keys, system)
            return fake_m3_json(prompt, model, need_keys, system)

        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", "旧摘要")}, {}, {}, {701: 0}),
                ), \
                patch.object(pu, "m3_json", side_effect=drift_during_observe), \
                patch.object(pu, "undo_window"):
            pu._run(
                work_id=8,
                max_windows=0,
                max_calls=0,
                model="MiniMax-M3",
                redo_window=0,
                semantic_on=False,
            )
        check(f"pure-presence-{name}-payload零覆盖", state["payload"] == original_payload)
        check(f"pure-presence-{name}-窗失败关闭", state["window_status"] == "failed")
        if name == "外部状态终态":
            check("pure-presence-终态在UPDATE前阻断", not state.get("draft_update_sql"))
        else:
            check(
                "pure-presence-模型期间revision变化在UPDATE前阻断",
                not state.get("draft_update_sql"),
                detail=str(state.get("draft_update_sql")),
            )


def test_run_presence_paths_accept_numeric_strings():
    """同卡横跨新信息与纯出场时，有无模型更新都只走一次实体写入并落全章节。"""

    def run_case(updates):
        state = {
            "window_status": "pending",
            "revisions": {701: 0},
            "payload": {
                "type": "character",
                "名称": "安若雪",
                "别名": [],
                "字段": {},
                "出场章": [],
                "_work_id": 8,
            },
        }

        def fake_m3_json(prompt, model, need_keys, system=None):
            if need_keys == ("新名字", "已知实体新信息", "纯出场"):
                return ({
                    "新名字": [],
                    "已知实体新信息": [
                        {"名称": "安若雪", "观察点": "出现新变化", "出场章": ["70"]}
                    ],
                    "纯出场": [{"名称": "安若雪", "出场章": ["71"]}],
                }, {})
            return ({"更新": deepcopy(updates)}, {})

        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", "旧摘要")}, {}, {}, {701: 0}),
                ), \
                patch.object(pu, "m3_json", side_effect=fake_m3_json), \
                patch.object(pu, "undo_window"), \
                patch.object(pu, "merge_card", wraps=pu.merge_card) as merge_mock, \
                patch.object(
                    pu, "_write_locked_upgrade_draft", wraps=pu._write_locked_upgrade_draft
                ) as write_mock:
            pu._run(8, 0, 0, "MiniMax-M3", 0, False)
        return state, merge_mock.call_count, write_mock.call_count

    updated, merge_calls, write_calls = run_case([
        {"draft_id": 701, "变更字段": {"一句话摘要": "新摘要"}, "别名新增": []},
    ])
    check("run-presence-有更新时只写一次",
          merge_calls == 1 and write_calls == 1 and updated["revisions"][701] == 1,
          detail=f"merge={merge_calls}, write={write_calls}, state={updated}")
    check("run-presence-有更新时一次消费两数组章节",
          updated["window_status"] == "done" and updated["payload"]["出场章"] == [70, 71],
          detail=str(updated))

    unchanged, merge_calls, write_calls = run_case([])
    check("run-presence-无更新时由机械收尾写一次",
          merge_calls == 0 and write_calls == 1 and unchanged["revisions"][701] == 1,
          detail=f"merge={merge_calls}, write={write_calls}, state={unchanged}")
    check("run-presence-无更新时两数组章节仍落库",
          unchanged["window_status"] == "done" and unchanged["payload"]["出场章"] == [70, 71],
          detail=str(unchanged))


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", "安若雪", "联邦战士", [], 6)])
            if "example_upgrade_alias" in query:
                return Result([("安若雪", "雪姐")])
            return Result([("character", "李四", 7)])

    conn = Conn()
    name_map, presence, aliases_by_draft, revisions = 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})
    check("known-revision与模型底册同步捕获", revisions == {1: 6})
    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"] == ("证据修复",))


class _IntegrityConn:
    """升格写入完整性内存库：模拟 draft 行锁、alias 唯一键、审计与水位撤销。"""

    def __init__(self, payload, *, status="pending", source_type=None, revision=4,
                 watermark=3, aliases=None):
        self.drafts = {
            701: {
                "payload": deepcopy(payload),
                "status": status,
                "source_type": source_type or pu.SOURCE_TYPE,
                "revision": revision,
                "deleted": False,
                "updater": "before",
            }
        }
        self.states = {} if watermark is None else {701: watermark}
        self.aliases = {
            alias: {
                "canonical_name": canonical,
                "evidence_window": evidence_window,
                "deleted": False,
            }
            for alias, (canonical, evidence_window) in (aliases or {}).items()
        }
        self.audits = []
        self.sql_order = []

    def execute(self, query, params=()):
        """按生产 SQL 的关键语义更新内存状态，任何未模拟读查询默认返回空集。"""
        normalized = " ".join(query.split())
        self.sql_order.append(normalized)
        if normalized.startswith(
                "SELECT draft_payload, status, revision, source_type FROM muse_knowledge_draft"):
            row = self.drafts.get(params[0])
            if not row or row["deleted"]:
                return _CardResult(None)
            return _CardResult((row["payload"], row["status"], row["revision"], row["source_type"]))
        if normalized.startswith("SELECT draft_payload FROM muse_knowledge_draft WHERE id=%s"):
            return _CardResult((self.drafts[params[0]]["payload"],))
        if normalized.startswith("SELECT watermark_window FROM example_upgrade_card_state"):
            watermark = self.states.get(params[0])
            return _CardResult((watermark,) if watermark is not None else None)
        if normalized.startswith("INSERT INTO example_upgrade_audit"):
            self.audits.append({
                "draft_id": params[0],
                "window_no": params[1],
                "field_name": params[2],
                "old_value": params[3],
                "new_value": params[4],
            })
            return _CardResult()
        if normalized.startswith("INSERT INTO example_upgrade_alias"):
            work_id, canonical_name, alias, evidence_window = params[:4]
            existing = self.aliases.get(alias)
            if existing is None:
                self.aliases[alias] = {
                    "canonical_name": canonical_name,
                    "evidence_window": evidence_window,
                    "deleted": False,
                }
                return _CardResult((canonical_name,))
            if existing["canonical_name"] == canonical_name and not existing.get("deleted", False):
                return _CardResult((canonical_name,))
            return _CardResult(None)
        if normalized.startswith("SELECT canonical_name, deleted FROM example_upgrade_alias"):
            existing = self.aliases.get(params[2])
            return _CardResult(
                (existing["canonical_name"], existing.get("deleted", False)) if existing else None
            )
        if normalized.startswith("UPDATE muse_knowledge_draft SET draft_payload=%s"):
            row = self.drafts[params[1]]
            if "RETURNING revision" in normalized:
                expected_revision = params[-1]
                if row["status"] != "pending" or row["source_type"] != pu.SOURCE_TYPE \
                        or row["revision"] != expected_revision or row["deleted"]:
                    return _CardResult(None)
                row["payload"] = json.loads(params[0])
                row["revision"] += 1
                if "updater='upgrade-undo'" in normalized:
                    row["updater"] = "upgrade-undo"
                return _CardResult((row["revision"],))
            row["payload"] = json.loads(params[0])
            if "revision=revision+1" in normalized:
                row["revision"] += 1
            if "updater='upgrade-undo'" in normalized:
                row["updater"] = "upgrade-undo"
            return _CardResult()
        if normalized.startswith("INSERT INTO example_upgrade_card_state"):
            draft_id, _, watermark = params[:3]
            if "SET watermark_window=EXCLUDED.watermark_window" in normalized:
                self.states[draft_id] = watermark
            else:
                self.states[draft_id] = max(self.states.get(draft_id, watermark), watermark)
            return _CardResult()
        if normalized.startswith("SELECT a.draft_id, a.field_name, a.old_value"):
            window_no = params[1]
            rows = [
                (audit["draft_id"], audit["field_name"], audit["old_value"])
                for audit in reversed(self.audits)
                if audit["window_no"] == window_no
            ]
            return _CardResult(rows=rows)
        if normalized.startswith("DELETE FROM example_upgrade_card_state"):
            self.states.pop(params[0], None)
            return _CardResult()
        if normalized.startswith("DELETE FROM example_upgrade_audit"):
            window_no = params[1]
            self.audits = [audit for audit in self.audits if audit["window_no"] != window_no]
            return _CardResult()
        if normalized.startswith("SELECT id, draft_payload FROM muse_knowledge_draft"):
            return _CardResult(rows=[
                (draft_id, row["payload"])
                for draft_id, row in self.drafts.items()
                if not row["deleted"]
            ])
        if normalized.startswith("DELETE FROM example_upgrade_alias"):
            evidence_window = params[2]
            self.aliases = {
                alias: row for alias, row in self.aliases.items()
                if row["evidence_window"] != evidence_window
            }
            return _CardResult()
        return _CardResult(rows=[])


def test_upgrade_writes_lock_current_draft_and_reject_invalid_state():
    """实体与关系写入必须锁后取当前值，并拒绝非 pending 或非 upgrade_book。"""
    current_payload = {
        "type": pu.RELATION_TYPE,
        "名称": "甲×乙",
        "关系类型": "盟友",
        "字段": {"当前状态": "数据库新值", "并发新增": "必须保留"},
        "_work_id": 8,
    }
    stale_payload = deepcopy(current_payload)
    stale_payload["字段"] = {"当前状态": "模型调用前旧值"}
    conn = _IntegrityConn(current_payload, revision=9)
    pu._update_relation_card(
        conn,
        8,
        8,
        701,
        9,
        {"关系类型": "盟友", "本窗演变": "继续合作", "其他字段": {"信任基础": "共同目标"}},
    )
    updated = conn.drafts[701]
    first_write = next(index for index, sql in enumerate(conn.sql_order)
                       if sql.startswith(("INSERT", "UPDATE", "DELETE")))
    lock_index = next(index for index, sql in enumerate(conn.sql_order)
                      if "FROM muse_knowledge_draft" in sql and "FOR UPDATE" in sql)
    check("write-lock-关系基于锁后当前payload", updated["payload"]["字段"]["并发新增"] == "必须保留")
    check("write-lock-先锁行后写入", lock_index < first_write, detail=str(conn.sql_order))
    check("write-lock-revision受保护只增一次", updated["revision"] == 10)

    guarded = _IntegrityConn(current_payload, revision=5)
    try:
        pu._write_locked_upgrade_draft(guarded, 701, current_payload, 4)
    except pu.UpgradeDraftWriteConflict:
        check("write-lock-过期revision失败关闭", True)
    else:
        check("write-lock-过期revision失败关闭", False, detail="过期 revision 被写入")
    check("write-lock-过期revision零覆盖", guarded.drafts[701]["revision"] == 5)

    for name, status, source_type in (
            ("非pending", "confirmed", pu.SOURCE_TYPE),
            ("非upgrade来源", "pending", "parse_book")):
        invalid = _IntegrityConn(
            {"type": "character", "名称": "安若雪", "字段": {}, "_work_id": 8},
            status=status,
            source_type=source_type,
        )
        try:
            pu.merge_card(invalid, 701, 8, {"阵营": "联邦"}, [])
        except pu.UpgradeDraftWriteConflict:
            check(f"write-lock-{name}失败关闭", True)
        else:
            check(f"write-lock-{name}失败关闭", False, detail="非法 draft 被写入")
        check(f"write-lock-{name}零写入", not invalid.audits and invalid.states == {701: 3})


def test_alias_claim_payload_undo_and_entity_watermark():
    """alias 不同 owner 必须拒绝；同 owner 幂等后 payload 与实体水位都可精确撤销。"""
    old_payload = {
        "type": "character",
        "名称": "安若雪",
        "一句话摘要": "旧摘要",
        "字段": {"阵营": "联邦"},
        "_work_id": 8,
    }
    conflict = _IntegrityConn(old_payload, aliases={"雪姐": ("另一人物", 2)})
    try:
        pu.merge_card(conflict, 701, 8, {}, ["雪姐"])
    except pu.AliasOwnershipConflict as exc:
        check("alias-不同canonical确定性失败", "雪姐" in str(exc))
    else:
        check("alias-不同canonical确定性失败", False, detail="唯一冲突被静默吞掉")
    check("alias-冲突不写payload", conflict.drafts[701]["payload"] == old_payload)

    conn = _IntegrityConn(old_payload, revision=6, watermark=3,
                          aliases={"雪姐": ("安若雪", 2)})
    pu.merge_card(conn, 701, 8, {"一句话摘要": "新摘要"}, ["雪姐"])
    check("alias-同canonical幂等后才写payload", conn.drafts[701]["payload"]["别名"] == ["雪姐"])
    check("alias-同canonical保留原证据窗", conn.aliases["雪姐"]["evidence_window"] == 2)
    watermark_audit = next(
        audit for audit in conn.audits
        if audit["field_name"] == pu.CARD_STATE_WATERMARK_AUDIT_FIELD
    )
    check("entity-watermark-审计完整旧值", json.loads(watermark_audit["old_value"]) == 3)
    check("entity-watermark-推进到本窗", conn.states[701] == 8)
    pu.undo_window(conn, 8, 8)
    check("alias-undo恢复顶层键缺失", conn.drafts[701]["payload"] == old_payload)
    check("entity-watermark-undo恢复旧值", conn.states[701] == 3)
    check("alias-undo不删旧窗同canonical映射", "雪姐" in conn.aliases)

    payload_with_empty_aliases = {**deepcopy(old_payload), "别名": []}
    fresh = _IntegrityConn(payload_with_empty_aliases, watermark=None)
    pu.merge_card(fresh, 701, 8, {}, ["若雪"])
    check("alias-新映射成功后写payload", fresh.drafts[701]["payload"]["别名"] == ["若雪"])
    pu.undo_window(fresh, 8, 8)
    check("alias-undo恢复顶层空列表", fresh.drafts[701]["payload"] == payload_with_empty_aliases)
    check("alias-undo删除本窗新映射", "若雪" not in fresh.aliases)
    check("entity-watermark-undo删除本窗新state", 701 not in fresh.states)


def test_undo_advances_revision_and_invalidates_precompensation_cas():
    """撤销恢复与新卡软删都推进 revision，使基于撤销前版本的外部 CAS 必然失败。"""

    old_payload = {
        "type": "character", "名称": "安若雪", "一句话摘要": "旧摘要",
        "字段": {}, "_work_id": 8,
    }
    conn = _IntegrityConn(old_payload, revision=4)
    pu.merge_card(
        conn, 701, 8, {"一句话摘要": "窗内摘要"}, [], expected_revision=4
    )
    stale_revision = conn.drafts[701]["revision"]
    stale_payload = deepcopy(conn.drafts[701]["payload"])
    pu.undo_window(conn, 8, 8)
    check("undo-revision-既有卡恢复后单调推进且标记updater",
          conn.drafts[701]["revision"] > stale_revision
          and conn.drafts[701]["updater"] == "upgrade-undo")
    try:
        pu._write_locked_upgrade_draft(conn, 701, stale_payload, stale_revision)
    except pu.UpgradeDraftWriteConflict:
        check("undo-revision-撤销前旧CAS失败", True)
    else:
        check("undo-revision-撤销前旧CAS失败", False, detail="旧 revision 仍写入成功")
    check("undo-revision-旧CAS不覆盖恢复payload", conn.drafts[701]["payload"] == old_payload)

    relation_conn = _RelationConn()
    new_payload = {
        "type": pu.RELATION_TYPE, "名称": "甲×乙", "甲方draft": 11, "乙方draft": 12,
        "关系类型": "盟友", "字段": {"演变轨迹": ["[窗8] 建立"]},
        "来源": "升格@窗8", "_work_id": 8,
    }
    new_id = pu._insert_relation_card(relation_conn, 8, 8, new_payload)
    before_delete_revision = relation_conn.drafts[new_id]["revision"]
    pu.undo_window(relation_conn, 8, 8)
    deleted = relation_conn.drafts[new_id]
    check("undo-revision-新卡软删revision单调且标记updater",
          deleted["deleted"] is True and deleted["revision"] > before_delete_revision
          and deleted["updater"] == "upgrade-undo")


# ── ⑥ 三个提示词含关键纪律语句（洞②登场必须 / 洞③≤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_resolve_canonical_name,
               test_debut_milestone, test_clean_milestone_guard,
               test_merge_material, test_build_embed_text_type_fix, test_int_chaps,
               test_relation_update_and_new_card_undo_boundaries,
               test_upgrade_writes_lock_current_draft_and_reject_invalid_state,
               test_alias_claim_payload_undo_and_entity_watermark,
               test_undo_advances_revision_and_invalidates_precompensation_cas,
               test_entity_chapter_evidence_filter, test_new_card_chapter_evidence_chain,
               test_merge_card_chapter_evidence, test_redo_cleans_legacy_appearance_chapters,
               test_run_alias_paths_store_real_canonical_name,
               test_run_new_card_registers_normalized_name_and_aliases_same_window,
               test_run_exact_and_disjoint_duplicate_entity_updates_merge_once,
               test_run_conflicting_duplicate_entity_updates_fail_before_any_write,
               test_run_relation_cards_enter_touched_embedding,
               test_run_conflicting_duplicate_relations_fail_before_any_relation_write,
               test_run_identical_duplicate_relations_write_existing_card_once,
               test_run_prompt_revision_rejects_entity_and_relation_stale_results,
               test_relation_participants_lock_before_prompt_and_reject_external_drift,
               test_run_alias_gate_precedes_empty_update_and_deleted_alias_fails_closed,
               test_run_embedding_owner_conflict_marks_window_failed,
               test_owner_conflict_external_confirmation_preserves_draft_and_persists_breakpoint,
               test_redo_owner_conflict_restores_snapshot_without_normal_undo,
               test_redo_full_fence_rejects_untouched_and_auxiliary_domain_drift,
               test_compensation_exception_persists_durable_failed_and_next_run_stops,
               test_run_embedding_payload_drift_marks_window_failed_without_vector_writes,
               test_embed_touched_migrates_soft_deleted_hash_owner,
               test_embed_touched_rejects_two_active_drafts_with_same_hash,
               test_embed_touched_rejects_existing_active_hash_owner,
               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_pure_presence_rejects_external_state_and_revision_conflicts,
               test_run_presence_paths_accept_numeric_strings,
               test_load_known_projection, test_repair_milestone_evidence,
               test_prompts_disciplines):
        fn()
    print(f"\n全部离线自测通过：{_passed} 项（未连库、未发任何网络/嵌入/LLM 调用）")