修复: 封闭升格续跑并发与补偿边界

.claude/skills/embed/SKILL.md

@@ -10,7 +10,7 @@ description: New-API 嵌入封装——Qwen3-Embedding-8B、dimensions=1024、
 ## 用法
 
 ```bash
-# 批量嵌入所有待嵌知识草稿行（默认：muse_knowledge_draft 中 status='pending' 且尚无嵌入行的）
+# 批量补嵌 pending 草稿（无活向量，或活向量的当前 payload+model hash 已过期）
 .venv/bin/python .claude/skills/embed/scripts/embed_drafts.py
 
 # 指定 work（=参考书拆书批次）或限量
@@ -24,8 +24,8 @@ description: New-API 嵌入封装——Qwen3-Embedding-8B、dimensions=1024、
 
 - **嵌入文本构造**：`【型】名称：一句话摘要\n字段正文摘选`（draft_payload 的 embed_text 字段优先；无则按固定拼接），与检索端 query 语义对齐。
 - **幂等与 owner**：sha256(嵌入文本+模型) 为 `content_hash`（uk: tenant+hash+model）。只有唯一行 `deleted=FALSE`、绑定同一 `draft_id`，且 owner draft 同租户并 `deleted=FALSE` 时才幂等跳过；`entity_id` 非空或其他 active draft owner 明确冲突并失败，绝不迁移 owner。旧 owner draft 已软删时，允许在写前活性重验后把唯一行条件迁到当前 draft。
-- **批量**：每请求 ≤16 条文本；失败整批重试 2 次（指数退避），仍失败逐条降级重试，坏行记错并继续（不断批）。
-- **落库与 reset/confirm/parse 并发**：HTTP 期间不持数据库事务。每个 draft 写入使用独立事务，先 `SELECT ... FOR UPDATE` 锁定 draft 并重验租户、`deleted=FALSE`、`status='pending'`；同时读取当前 `draft_payload`，重构文本与 hash，和 HTTP 前快照任一不一致即跳过。随后 `SELECT ... FOR UPDATE` 锁定该 draft 全部活向量：任一 `entity_id` 非空则冲突失败，同 hash 当前活向量则幂等跳过，其他 hash 的无 entity 旧活向量在 UPSERT 前统一软删，保证每 draft 仅一个活向量。最后锁同 hash 唯一行，执行带 owner 条件的 UPSERT 并用 `RETURNING draft_id` 校验。该 draft 行锁与 reset 的七表 `SHARE ROW EXCLUSIVE` 配合：embed 先锁时 reset 等待且随后能发现快照漂移；reset 先完成时 embed 等待后看到软删并跳过。confirm/parse 先完成时 embed 在锁后看到状态或 payload/hash 漂移并跳过。以上跳过路径均零向量写入。
+- **批量**：读取每个 pending draft 的全部活向量，在 Python 中复用统一文本与 hash 规则筛选“无活向量”或“活向量 hash/model 与当前目标不一致”的候选；对完整候选集完成状态、同批 hash 与目标 owner 只读预检并提交后，`limit` 才限制实际 HTTP/写入行，每个实际 chunk 在 HTTP 前再次预查 owner 以封住竞态。每请求 ≤16 条文本；响应 `index` 必须是范围内唯一整数并完整覆盖请求槽位，缺项、重复或越界进入既有整批重试和逐条降级。失败整批重试 2 次（指数退避），仍失败逐条降级重试，坏行记错并继续（不断批）。网络异常、返回 `bad`、向量缺项或 `None` 均只记失败，不改旧向量，下一轮仍可重试；owner 冲突、多条活向量、同批目标 hash 冲突属于确定性异常，明确报告后令整条命令失败退出，不降级成失败计数。
+- **落库与 reset/confirm/parse 并发**：HTTP 期间不持数据库事务。每个 draft 写入使用独立事务，先 `SELECT ... FOR UPDATE` 锁定 draft 并重验租户、`deleted=FALSE`、`status='pending'`；同时读取当前 `draft_payload`，重构文本与 hash，和 HTTP 前快照任一不一致即跳过。随后 `SELECT ... FOR UPDATE` 锁定该 draft 全部活向量：多条活向量是异常状态并失败关闭，任一 `entity_id` 非空则冲突失败，同 hash 且同 model 的当前活向量才幂等跳过，其他 hash 或 model 的无 entity 旧活向量在 UPSERT 前统一软删，保证每 draft 仅一个活向量。最后锁同 hash 唯一行，执行带 owner 条件的 UPSERT 并用 `RETURNING draft_id` 校验。同批多个 draft 的目标 hash 相同时整组失败，不按执行顺序抢 owner。该 draft 行锁与 reset 的 11 表 `SHARE ROW EXCLUSIVE`（4 个输入源表 + 7 个产出表）配合：embed 先锁时 reset 等待且随后能发现快照漂移；reset 先完成时 embed 等待后看到软删并跳过。confirm/parse 先完成时 embed 在锁后看到状态或 payload/hash 漂移并跳过。以上跳过或失败路径均零向量写入。
 - **落库字段**：`example_knowledge_embedding(draft_id, content_hash, embed_text, model, dimensions=1024, embedding)`；draft 确认落 entity 后由 confirm 流程回填 entity_id（嵌入行不动）。
 - 汇报：新嵌 N、跳过 M、失败 K；幂等、失活和冲突原因均输出可追踪明细。
 

.claude/skills/embed/scripts/embed_drafts.py

@@ -37,7 +37,21 @@ def embed_texts(sess, texts):
             "model": MODEL, "input": batch, "dimensions": DIM}, timeout=120)
         r.raise_for_status()
         data = r.json()["data"]
-        return [d["embedding"] for d in sorted(data, key=lambda d: d["index"])]
+        # 响应 index 是请求槽位，不能排序后压缩；缺项、重复或越界都必须让本次调用失败并进入重试。
+        vectors = [None] * len(batch)
+        seen = set()
+        for item in data:
+            index = item["index"]
+            if type(index) is not int or not 0 <= index < len(batch):
+                raise ValueError(f"embedding 响应 index 越界或非整数：{index!r}")
+            if index in seen:
+                raise ValueError(f"embedding 响应 index 重复：{index}")
+            vectors[index] = item["embedding"]
+            seen.add(index)
+        if len(seen) != len(batch):
+            missing = sorted(set(range(len(batch))) - seen)
+            raise ValueError(f"embedding 响应缺少 index：{missing}")
+        return vectors
 
     for attempt in range(3):
         try:
@@ -72,6 +86,12 @@ def build_embed_text(payload: dict) -> str:
     return f"【{t}】{name}：{brief}\n{body}"[:4000]
 
 
+def _content_hash(text):
+    """统一生成向量幂等键，候选筛选与写前复验必须共用同一规则。"""
+
+    return hashlib.sha256(f"{text}|{MODEL}".encode()).hexdigest()
+
+
 class EmbeddingOwnershipConflict(RuntimeError):
     """同 hash 唯一行已归实体或其他活跃 draft，禁止迁移 owner。"""
 
@@ -152,7 +172,7 @@ def _write_embedding(conn, draft_id, content_hash, text, vector):
     # HTTP 期间 payload 可能被 parse/confirm 更新；锁内必须按当前 payload 重构文本与 hash，
     # 只要与 HTTP 请求所依据的快照不同，就丢弃陈旧向量，绝不覆盖并发产生的新结果。
     current_text = build_embed_text(current_payload or {})
-    current_hash = hashlib.sha256(f"{current_text}|{MODEL}".encode()).hexdigest()
+    current_hash = _content_hash(current_text)
     if current_text != text or current_hash != content_hash:
         click.echo(
             f"  [跳过] draft={draft_id} 写前 payload/hash 漂移，"
@@ -163,21 +183,27 @@ def _write_embedding(conn, draft_id, content_hash, text, vector):
 
     # 锁定该 draft 的全部活向量，保证 entity 归属和“每 draft 唯一活向量”在同一事务内判定。
     live_embeddings = conn.execute(
-        """SELECT id, content_hash, entity_id FROM example_knowledge_embedding
+        """SELECT id, content_hash, model, entity_id FROM example_knowledge_embedding
            WHERE tenant_id=%s AND draft_id=%s AND deleted=FALSE
            FOR UPDATE""",
         (TENANT, draft_id),
     ).fetchall()
+    if len(live_embeddings) > 1:
+        raise EmbeddingOwnershipConflict(
+            f"draft={draft_id} 存在多条活向量，状态异常，禁止自动修复：{live_embeddings}"
+        )
     entity_rows = [
-        (row_id, row_hash, entity_id)
-        for row_id, row_hash, entity_id in live_embeddings
+        (row_id, row_hash, row_model, entity_id)
+        for row_id, row_hash, row_model, entity_id in live_embeddings
         if entity_id is not None
     ]
     if entity_rows:
         raise EmbeddingOwnershipConflict(
             f"draft={draft_id} 存在 entity_id 非空旧活向量，禁止覆盖：{entity_rows}"
         )
-    if any(row_hash == content_hash for _, row_hash, _ in live_embeddings):
+    if any(
+            row_hash == content_hash and row_model == MODEL
+            for _, row_hash, row_model, _ in live_embeddings):
         click.echo(f"  [跳过] draft={draft_id} 同 hash 活向量已由当前 draft 持有")
         return False
 
@@ -192,8 +218,8 @@ def _write_embedding(conn, draft_id, content_hash, text, vector):
         conn.execute(
             """UPDATE example_knowledge_embedding SET deleted=TRUE, updater=%s
                WHERE tenant_id=%s AND draft_id=%s AND deleted=FALSE
-                 AND entity_id IS NULL AND content_hash!=%s""",
-            (ACTOR, TENANT, draft_id, content_hash),
+                 AND entity_id IS NULL AND (content_hash!=%s OR model!=%s)""",
+            (ACTOR, TENANT, draft_id, content_hash, MODEL),
         )
 
     # 条件 UPSERT 是行锁检查后的第二道防线：当预查时唯一行尚不存在、随后被并发插入时，
@@ -228,9 +254,169 @@ def _write_embedding(conn, draft_id, content_hash, text, vector):
     return True
 
 
+def _load_bulk_candidates(conn, work_id, limit):
+    """读取 pending draft 的全部活向量，在 Python 中按当前文本和模型筛选补嵌候选。"""
+
+    sql = """SELECT d.id, d.draft_payload,
+                    e.id, e.content_hash, e.model, e.entity_id
+             FROM muse_knowledge_draft d
+             LEFT JOIN example_knowledge_embedding e
+               ON e.tenant_id=%s AND e.draft_id=d.id AND e.deleted=FALSE
+             WHERE d.tenant_id=%s AND d.deleted=FALSE AND d.status='pending'"""
+    args = [TENANT, TENANT]
+    if work_id is not None:
+        sql += " AND d.source_id=%s"
+        args.append(work_id)
+    # 必须先取得每个 draft 的全部活向量，不能在 SQL 层 LIMIT 后漏掉旧 hash 或异常状态。
+    sql += " ORDER BY d.id, e.id"
+    rows = conn.execute(sql, args).fetchall()
+
+    grouped = {}
+    for draft_id, payload, embedding_id, row_hash, row_model, entity_id in rows:
+        draft = grouped.setdefault(draft_id, {"payload": payload, "embeddings": []})
+        if embedding_id is not None:
+            draft["embeddings"].append((embedding_id, row_hash, row_model, entity_id))
+
+    candidates = []
+    failures_by_draft = {}
+    repair_targets = {}
+    for draft_id in sorted(grouped):
+        draft = grouped[draft_id]
+        text = build_embed_text(draft["payload"] or {})
+        content_hash = _content_hash(text)
+        live_embeddings = draft["embeddings"]
+        if len(live_embeddings) == 1:
+            _, row_hash, row_model, entity_id = live_embeddings[0]
+            if entity_id is None and row_hash == content_hash and row_model == MODEL:
+                continue
+
+        # 所有非健康目标都参与同批 hash 冲突检查，不能因其中一条先被判异常而放行另一条。
+        repair_targets.setdefault(content_hash, []).append(draft_id)
+        if len(live_embeddings) > 1:
+            failures_by_draft.setdefault(draft_id, []).append(
+                f"存在多条活向量，状态异常，禁止自动修复：{live_embeddings}"
+            )
+            continue
+        if live_embeddings:
+            _, row_hash, row_model, entity_id = live_embeddings[0]
+            if entity_id is not None:
+                failures_by_draft.setdefault(draft_id, []).append(
+                    f"活向量已归 entity={entity_id}，禁止 draft 补嵌迁移 owner"
+                )
+                continue
+        candidates.append((draft_id, content_hash, text))
+
+    # 相同目标 hash 的多个 draft 不能靠执行顺序决定 owner；冲突检查必须发生在 limit 之前。
+    conflicted_drafts = set()
+    for content_hash, draft_ids in repair_targets.items():
+        if len(draft_ids) < 2:
+            continue
+        reason = (
+            f"同批目标 hash 冲突：hash={content_hash}，drafts={draft_ids}，"
+            "禁止按执行顺序抢 owner"
+        )
+        for draft_id in draft_ids:
+            failures_by_draft.setdefault(draft_id, []).append(reason)
+            conflicted_drafts.add(draft_id)
+    candidates = [candidate for candidate in candidates if candidate[0] not in conflicted_drafts]
+
+    # limit 只能限制后续 HTTP/写入；先对完整候选集预查目标 hash owner，避免范围外冲突被隐藏。
+    prechecked_candidates = []
+    for draft_id, content_hash, text in candidates:
+        try:
+            action = _embedding_owner_action(conn, draft_id, content_hash)
+        except EmbeddingOwnershipConflict as exc:
+            failures_by_draft.setdefault(draft_id, []).append(str(exc))
+            continue
+        if action != "skip":
+            prechecked_candidates.append((draft_id, content_hash, text))
+
+    # 全量只读预检完成后释放事务，再截取实际处理行；每个 chunk 仍会再次预查以封住其后竞态。
+    conn.commit()
+    if limit and limit > 0:
+        prechecked_candidates = prechecked_candidates[:int(limit)]
+    failures = [
+        (draft_id, "；".join(reasons))
+        for draft_id, reasons in sorted(failures_by_draft.items())
+    ]
+    return prechecked_candidates, failures
+
+
+def _run_bulk(conn, sess, work_id, limit):
+    """执行一次 bulk 补嵌；HTTP 前后均保持既有 owner、锁和 stale-write 边界。"""
+
+    rows, read_failures = _load_bulk_candidates(conn, work_id, limit)
+    for draft_id, reason in read_failures:
+        click.echo(f"  [失败] draft={draft_id}: {reason}", err=True)
+    if read_failures:
+        details = "；".join(
+            f"draft={draft_id}: {reason}" for draft_id, reason in read_failures
+        )
+        raise EmbeddingOwnershipConflict(f"bulk 候选存在确定性冲突，已失败关闭：{details}")
+
+    done = skip = fail = 0
+    click.echo(f"待补嵌草稿: {len(rows)} 条（筛选失败 {len(read_failures)} 条）")
+
+    for i in range(0, len(rows), BATCH):
+        chunk = rows[i:i + BATCH]
+        metas = []
+        for draft_id, content_hash, text in chunk:
+            action = _embedding_owner_action(conn, draft_id, content_hash)
+            if action == "skip":
+                skip += 1
+                click.echo(f"  [跳过] draft={draft_id} 同 hash 活向量已由当前 draft 持有")
+                continue
+            metas.append((draft_id, content_hash, text))
+
+        # owner 预查只用于避免无效 HTTP；HTTP 期间不持数据库事务或表锁。
+        conn.commit()
+        if not metas:
+            continue
+        texts = [meta[2] for meta in metas]
+        try:
+            vecs, bad = embed_texts(sess, texts)
+        except Exception as exc:
+            for draft_id, _, _ in metas:
+                fail += 1
+                click.echo(f"  [失败] draft={draft_id}: HTTP 嵌入失败：{exc}", err=True)
+            continue
+
+        bad = set(bad or ())
+        for j, (draft_id, content_hash, text) in enumerate(metas):
+            if j in bad:
+                fail += 1
+                click.echo(f"  [失败] draft={draft_id}: HTTP 返回 bad，保留旧向量", err=True)
+                continue
+            try:
+                vector = vecs[j]
+            except (IndexError, TypeError):
+                fail += 1
+                click.echo(f"  [失败] draft={draft_id}: HTTP 返回向量缺项，保留旧向量", err=True)
+                continue
+            if vector is None:
+                fail += 1
+                click.echo(f"  [失败] draft={draft_id}: HTTP 返回空向量，保留旧向量", err=True)
+                continue
+
+            # 每个 draft 独立事务：确定性冲突回滚当前事务并向上抛，使命令以非零状态退出。
+            with conn.transaction():
+                written = _write_embedding(conn, draft_id, content_hash, text, vector)
+            if written:
+                done += 1
+            else:
+                skip += 1
+        click.echo(
+            f"  进度 {min(i + BATCH, len(rows))}/{len(rows)}"
+            f"（新嵌{done} 跳过{skip} 失败{fail}）"
+        )
+
+    click.echo(f"完成：新嵌 {done}、跳过 {skip}、失败 {fail}")
+    return {"done": done, "skip": skip, "fail": fail}
+
+
 @click.command()
 @click.option("--work-id", type=int, help="限定拆书批次的 work（draft.work_id=0 为全局行，用 source_id 关联参考书）")
-@click.option("--limit", type=int, default=0, help="最多处理条数（0=不限）")
+@click.option("--limit", type=click.IntRange(min=0), default=0, help="最多处理条数（0=不限）")
 @click.option("--probe", help="自由文本试嵌（打印维度与前 5 维，不落库）")
 def main(work_id, limit, probe):
     sess = _session()
@@ -243,51 +429,7 @@ def main(work_id, limit, probe):
         return
 
     with psycopg.connect(DSN) as conn:
-        # 待嵌=pending 草稿且无嵌入行
-        sql = """SELECT d.id, d.draft_payload FROM muse_knowledge_draft d
-                 WHERE d.tenant_id=%s AND d.deleted=FALSE AND d.status='pending'
-                   AND NOT EXISTS (SELECT 1 FROM example_knowledge_embedding e
-                                   WHERE e.tenant_id=%s AND e.draft_id=d.id AND e.deleted=FALSE)"""
-        args = [TENANT, TENANT]
-        if work_id is not None:
-            sql += " AND d.source_id=%s"
-            args.append(work_id)
-        sql += " ORDER BY d.id"
-        if limit:
-            sql += f" LIMIT {int(limit)}"
-        rows = conn.execute(sql, args).fetchall()
-        click.echo(f"待嵌草稿: {len(rows)} 条")
-
-        done = skip = fail = 0
-        for i in range(0, len(rows), BATCH):
-            chunk = rows[i:i + BATCH]
-            texts, metas = [], []
-            for did, payload in chunk:
-                text = build_embed_text(payload or {})
-                h = hashlib.sha256(f"{text}|{MODEL}".encode()).hexdigest()
-                if _embedding_owner_action(conn, did, h) == "skip":
-                    skip += 1  # 仅当前 active draft 持有 active 唯一行时才幂等跳过。
-                    continue
-                texts.append(text)
-                metas.append((did, h, text))
-            # owner 预查只用于避免无效 HTTP；明确提交读事务，HTTP 期间不持数据库事务或表锁。
-            conn.commit()
-            if not texts:
-                continue
-            vecs, bad = embed_texts(sess, texts)
-            for j, (did, h, text) in enumerate(metas):
-                if j in bad:
-                    fail += 1
-                    continue
-                # 每个 draft 独立事务：draft 行锁、owner 行锁、条件 UPSERT 同生共死。
-                with conn.transaction():
-                    written = _write_embedding(conn, did, h, text, vecs[j])
-                if written:
-                    done += 1
-                else:
-                    skip += 1
-            click.echo(f"  进度 {min(i + BATCH, len(rows))}/{len(rows)}（新嵌{done} 跳过{skip} 失败{fail}）")
-        click.echo(f"完成：新嵌 {done}、跳过 {skip}、失败 {fail}")
+        _run_bulk(conn, sess, work_id, limit)
 
 
 if __name__ == "__main__":

.claude/skills/embed/scripts/test_embed_drafts_offline.py

@@ -1,11 +1,14 @@
 #!/usr/bin/env python3
 """embed_drafts 并发与向量 owner 规则的纯离线测试。"""
 
+import copy
 import hashlib
 import pathlib
 import sys
 import unittest
-from unittest.mock import patch
+from unittest.mock import MagicMock, Mock, patch
+
+from click.testing import CliRunner
 
 
 SCRIPT_DIR = pathlib.Path(__file__).resolve().parent
@@ -49,7 +52,7 @@ class _EmbeddingConnection:
         }
         self.draft_live_embeddings = [
             {"id": row_id, "content_hash": content_hash,
-             "entity_id": entity_id, "deleted": False}
+             "model": embed.MODEL, "entity_id": entity_id, "deleted": False}
             for row_id, content_hash, entity_id in (draft_live_embeddings or [])
         ]
         self.embedding = None
@@ -83,11 +86,12 @@ class _EmbeddingConnection:
                 self.candidate["status"], self.candidate["payload"],
             ))
 
-        if normalized.startswith("select id, content_hash, entity_id from example_knowledge_embedding"):
+        if normalized.startswith(
+                "select id, content_hash, model, entity_id from example_knowledge_embedding"):
             self.events.append("draft_vectors_lock")
             self.assert_draft_vectors_lock_sql(normalized, params)
             rows = [
-                (row["id"], row["content_hash"], row["entity_id"])
+                (row["id"], row["content_hash"], row["model"], row["entity_id"])
                 for row in self.draft_live_embeddings if not row["deleted"]
             ]
             return _Result(rows=rows)
@@ -141,6 +145,7 @@ class _EmbeddingConnection:
             self.draft_live_embeddings.append({
                 "id": 999,
                 "content_hash": params[1],
+                "model": params[3],
                 "entity_id": None,
                 "deleted": False,
             })
@@ -151,7 +156,8 @@ class _EmbeddingConnection:
             self.assert_old_vectors_update_sql(normalized, params)
             for row in self.draft_live_embeddings:
                 if (not row["deleted"] and row["entity_id"] is None
-                        and row["content_hash"] != params[2]):
+                        and (row["content_hash"] != params[3]
+                             or row["model"] != params[4])):
                     row["deleted"] = True
             return _Result()
 
@@ -185,11 +191,11 @@ class _EmbeddingConnection:
 
     @staticmethod
     def assert_old_vectors_update_sql(sql, params):
-        """只软删当前 draft 的无 entity 旧 hash 活向量。"""
+        """只软删当前 draft 的无 entity 旧 hash 或旧 model 活向量。"""
 
         assert "entity_id is null" in sql
-        assert "content_hash!=%s" in sql
-        assert params == (embed.ACTOR, embed.TENANT, 101, CONTENT_HASH)
+        assert "content_hash!=%s or model!=%s" in sql
+        assert params == (embed.ACTOR, embed.TENANT, 101, CONTENT_HASH, embed.MODEL)
 
     @staticmethod
     def assert_owner_sql(sql, params):
@@ -221,6 +227,223 @@ class _EmbeddingConnection:
         assert params[1] == CONTENT_HASH
 
 
+class _BulkConnection:
+    """模拟 bulk 的候选读取、owner 预查和逐 draft 写事务。"""
+
+    def __init__(self, drafts, vectors=None):
+        self.drafts = {
+            draft_id: {
+                "tenant": embed.TENANT,
+                "deleted": False,
+                "status": "pending",
+                "payload": payload,
+                "source_id": source_id,
+            }
+            for draft_id, payload, source_id in drafts
+        }
+        self.vectors = [dict(vector) for vector in (vectors or [])]
+        self.events = []
+        self.sql = []
+        self._next_vector_id = 1000
+
+    def execute(self, sql, params=None):
+        normalized = " ".join(sql.split()).lower()
+        self.sql.append((normalized, params))
+
+        if normalized.startswith("select d.id, d.draft_payload"):
+            self._assert_bulk_select_sql(normalized)
+            work_id = params[2] if len(params) == 3 else None
+            rows = []
+            for draft_id, draft in sorted(self.drafts.items()):
+                if (draft["tenant"] != embed.TENANT or draft["deleted"]
+                        or draft["status"] != "pending"):
+                    continue
+                if work_id is not None and draft["source_id"] != work_id:
+                    continue
+                active = sorted(
+                    (row for row in self.vectors
+                     if row["tenant"] == embed.TENANT
+                     and row["draft_id"] == draft_id and not row["deleted"]),
+                    key=lambda row: row["id"],
+                )
+                if not active:
+                    rows.append((draft_id, draft["payload"], None, None, None, None))
+                    continue
+                rows.extend((
+                    draft_id, draft["payload"], row["id"], row["content_hash"],
+                    row["model"], row["entity_id"],
+                ) for row in active)
+            self.events.append("bulk_select")
+            return _Result(rows=rows)
+
+        if normalized.startswith("lock table"):
+            self.events.append("table_lock")
+            return _Result()
+
+        if normalized.startswith(
+                "select tenant_id, deleted, status, draft_payload from muse_knowledge_draft"):
+            draft_id = params[0]
+            draft = self.drafts.get(draft_id)
+            self.events.append(f"candidate_lock:{draft_id}")
+            if draft is None:
+                return _Result(None)
+            return _Result((
+                draft["tenant"], draft["deleted"], draft["status"], draft["payload"],
+            ))
+
+        if normalized.startswith(
+                "select id, content_hash, model, entity_id from example_knowledge_embedding"):
+            self.assertInSql(normalized, "for update")
+            draft_id = params[1]
+            rows = [
+                (row["id"], row["content_hash"], row["model"], row["entity_id"])
+                for row in self.vectors
+                if row["tenant"] == params[0] and row["draft_id"] == draft_id
+                and not row["deleted"]
+            ]
+            self.events.append(f"draft_vectors_lock:{draft_id}")
+            return _Result(rows=rows)
+
+        if normalized.startswith("select e.draft_id, e.entity_id, e.deleted"):
+            tenant, content_hash, model = params
+            owner = next((
+                row for row in self.vectors
+                if row["tenant"] == tenant and row["content_hash"] == content_hash
+                and row["model"] == model
+            ), None)
+            self.events.append("owner_lock" if "for update of e" in normalized else "owner_read")
+            if owner is None:
+                return _Result(None)
+            owner_draft = self.drafts.get(owner["draft_id"])
+            return _Result((
+                owner["draft_id"], owner["entity_id"], owner["deleted"],
+                True if owner_draft is None else owner_draft["deleted"],
+                None if owner_draft is None else owner_draft["tenant"],
+            ))
+
+        if normalized.startswith("update example_knowledge_embedding set deleted=true"):
+            self.assertInSql(normalized, "content_hash!=%s or model!=%s")
+            _, tenant, draft_id, content_hash, model = params
+            for row in self.vectors:
+                if (row["tenant"] == tenant and row["draft_id"] == draft_id
+                        and not row["deleted"] and row["entity_id"] is None
+                        and (row["content_hash"] != content_hash or row["model"] != model)):
+                    row["deleted"] = True
+            self.events.append(f"old_vectors_soft_delete:{draft_id}")
+            return _Result()
+
+        if normalized.startswith("insert into example_knowledge_embedding"):
+            draft_id, content_hash, _, model = params[:4]
+            owner = next((
+                row for row in self.vectors
+                if row["tenant"] == embed.TENANT and row["content_hash"] == content_hash
+                and row["model"] == model
+            ), None)
+            if owner is not None:
+                owner_draft = self.drafts.get(owner["draft_id"])
+                owner_active = owner_draft is not None and not owner_draft["deleted"]
+                if (owner["entity_id"] is not None
+                        or (owner["draft_id"] != draft_id and owner_active)):
+                    return _Result(None)
+                owner.update({"draft_id": draft_id, "entity_id": None, "deleted": False})
+            else:
+                self.vectors.append({
+                    "id": self._next_vector_id,
+                    "tenant": embed.TENANT,
+                    "draft_id": draft_id,
+                    "content_hash": content_hash,
+                    "model": model,
+                    "entity_id": None,
+                    "deleted": False,
+                })
+                self._next_vector_id += 1
+            self.events.append(f"upsert:{draft_id}")
+            return _Result((draft_id,))
+
+        raise AssertionError(f"未覆盖的 bulk 离线 SQL：{normalized}")
+
+    def commit(self):
+        self.events.append("commit")
+
+    def transaction(self):
+        return self._Transaction(self)
+
+    @staticmethod
+    def assertInSql(sql, fragment):
+        assert fragment in sql, f"SQL 缺少约束：{fragment}\n{sql}"
+
+    @classmethod
+    def _assert_bulk_select_sql(cls, sql):
+        """候选查询必须读取全部活向量，筛选和 limit 留给 Python。"""
+
+        for fragment in (
+                "left join example_knowledge_embedding e",
+                "e.deleted=false", "e.content_hash", "e.model", "e.entity_id",
+                "d.deleted=false", "d.status='pending'", "order by d.id", "e.id"):
+            cls.assertInSql(sql, fragment)
+        assert " limit " not in f" {sql} "
+
+    class _Transaction:
+        """异常时恢复向量状态，模拟 PostgreSQL 单 draft 事务回滚。"""
+
+        def __init__(self, conn):
+            self.conn = conn
+            self.snapshot = None
+
+        def __enter__(self):
+            self.snapshot = copy.deepcopy(self.conn.vectors)
+            return self
+
+        def __exit__(self, exc_type, exc_value, traceback):
+            if exc_type is not None:
+                self.conn.vectors = self.snapshot
+            return False
+
+
+def _draft(draft_id, text, source_id=1):
+    return draft_id, {"embed_text": text}, source_id
+
+
+def _vector(vector_id, draft_id, content_hash, *, model=embed.MODEL,
+            entity_id=None, deleted=False):
+    return {
+        "id": vector_id,
+        "tenant": embed.TENANT,
+        "draft_id": draft_id,
+        "content_hash": content_hash,
+        "model": model,
+        "entity_id": entity_id,
+        "deleted": deleted,
+    }
+
+
+class _EmbeddingResponse:
+    """模拟 embeddings HTTP 响应，仅提供生产代码实际读取的方法。"""
+
+    def __init__(self, data):
+        self.data = data
+
+    def raise_for_status(self):
+        return None
+
+    def json(self):
+        return {"data": self.data}
+
+
+class _EmbeddingSession:
+    """按请求文本动态返回带原始 index 的离线 embeddings 响应。"""
+
+    def __init__(self, responder):
+        self.responder = responder
+        self.inputs = []
+
+    def post(self, _url, *, json, timeout):
+        assert timeout == 120
+        inputs = list(json["input"])
+        self.inputs.append(inputs)
+        return _EmbeddingResponse(self.responder(inputs))
+
+
 class EmbedDraftsOfflineTest(unittest.TestCase):
     """覆盖 reset/embed 两种先后顺序与 hash owner 反例。"""
 
@@ -406,5 +629,321 @@ class EmbedDraftsOfflineTest(unittest.TestCase):
         self.assertEqual(conn.events, ["table_lock", "candidate_lock", "draft_vectors_lock"])
 
 
+class BulkSelfHealingOfflineTest(unittest.TestCase):
+    """直接覆盖 bulk 候选筛选、HTTP 边界和补嵌自愈结果。"""
+
+    VECTOR = [0.1, 0.2]
+
+    @staticmethod
+    def _candidate_ids(conn, limit=0):
+        candidates, failures = embed._load_bulk_candidates(conn, work_id=None, limit=limit)
+        return [candidate[0] for candidate in candidates], failures
+
+    def test无活向量入选且limit作用于Python筛选结果(self):
+        conn = _BulkConnection(
+            drafts=[_draft(100, TEXT), _draft(101, NEW_TEXT)],
+            vectors=[_vector(700, 100, CONTENT_HASH)],
+        )
+
+        candidate_ids, failures = self._candidate_ids(conn, limit=1)
+
+        self.assertEqual(candidate_ids, [101])
+        self.assertEqual(failures, [])
+
+    def test旧hash与旧model均入选并替换为唯一当前活向量(self):
+        conn = _BulkConnection(
+            drafts=[_draft(101, TEXT), _draft(102, NEW_TEXT)],
+            vectors=[
+                _vector(701, 101, "old-hash"),
+                _vector(702, 102, NEW_HASH, model="Old/Embedding-Model"),
+            ],
+        )
+
+        with patch.object(
+                embed, "embed_texts", return_value=([self.VECTOR, self.VECTOR], set())):
+            stats = embed._run_bulk(conn, sess=object(), work_id=None, limit=0)
+
+        self.assertEqual(stats, {"done": 2, "skip": 0, "fail": 0})
+        for draft_id, expected_hash in ((101, CONTENT_HASH), (102, NEW_HASH)):
+            active = [
+                row for row in conn.vectors
+                if row["draft_id"] == draft_id and not row["deleted"]
+            ]
+            self.assertEqual(len(active), 1)
+            self.assertEqual((active[0]["content_hash"], active[0]["model"]),
+                             (expected_hash, embed.MODEL))
+
+    def test批响应缺少index零时不压缩错写并逐条降级(self):
+        first_vector = [0.1]
+        second_vector = [0.2]
+
+        def respond(inputs):
+            if len(inputs) == 2:
+                # 服务端只返回第二条；若按排序后压缩，会被错误写给第一个 draft。
+                return [{"index": 1, "embedding": second_vector}]
+            if inputs == [TEXT]:
+                return []
+            if inputs == [NEW_TEXT]:
+                return [{"index": 0, "embedding": second_vector}]
+            raise AssertionError(f"未预期的输入：{inputs}")
+
+        conn = _BulkConnection(
+            drafts=[_draft(101, TEXT), _draft(102, NEW_TEXT)],
+            vectors=[
+                _vector(701, 101, "old-hash-1"),
+                _vector(702, 102, "old-hash-2"),
+            ],
+        )
+        sess = _EmbeddingSession(respond)
+
+        with patch.object(embed.time, "sleep"):
+            stats = embed._run_bulk(conn, sess=sess, work_id=None, limit=0)
+
+        self.assertEqual(stats, {"done": 1, "skip": 0, "fail": 1})
+        self.assertEqual(sess.inputs, [
+            [TEXT, NEW_TEXT], [TEXT, NEW_TEXT], [TEXT, NEW_TEXT], [TEXT], [NEW_TEXT],
+        ])
+        active_101 = [row for row in conn.vectors if row["draft_id"] == 101 and not row["deleted"]]
+        active_102 = [row for row in conn.vectors if row["draft_id"] == 102 and not row["deleted"]]
+        self.assertEqual([(row["content_hash"], row["model"]) for row in active_101],
+                         [("old-hash-1", embed.MODEL)])
+        self.assertEqual([(row["content_hash"], row["model"]) for row in active_102],
+                         [(NEW_HASH, embed.MODEL)])
+        candidate_ids, failures = self._candidate_ids(conn)
+        self.assertEqual(candidate_ids, [101])
+        self.assertEqual(failures, [])
+
+    def test乱序但完整的响应按原始index还原(self):
+        first_vector = [0.1]
+        second_vector = [0.2]
+        sess = _EmbeddingSession(lambda _inputs: [
+            {"index": 1, "embedding": second_vector},
+            {"index": 0, "embedding": first_vector},
+        ])
+
+        vectors, bad = embed.embed_texts(sess, [TEXT, NEW_TEXT])
+
+        self.assertEqual(vectors, [first_vector, second_vector])
+        self.assertEqual(bad, set())
+        self.assertEqual(sess.inputs, [[TEXT, NEW_TEXT]])
+
+    def test当前hash与model已有唯一活向量时跳过HTTP(self):
+        conn = _BulkConnection(
+            drafts=[_draft(101, TEXT)],
+            vectors=[_vector(701, 101, CONTENT_HASH)],
+        )
+        http = Mock()
+
+        with patch.object(embed, "embed_texts", http):
+            stats = embed._run_bulk(conn, sess=object(), work_id=None, limit=0)
+
+        self.assertEqual(stats, {"done": 0, "skip": 0, "fail": 0})
+        http.assert_not_called()
+
+    def test目标hash已有entity_owner时HTTP前失败且绝不迁移(self):
+        conn = _BulkConnection(
+            drafts=[_draft(101, TEXT)],
+            vectors=[_vector(700, 90, CONTENT_HASH, entity_id=9001)],
+        )
+        http = Mock()
+
+        with patch.object(embed, "embed_texts", http), \
+                patch.object(embed.click, "echo"):
+            with self.assertRaisesRegex(embed.EmbeddingOwnershipConflict, "entity"):
+                embed._run_bulk(conn, sess=object(), work_id=None, limit=0)
+
+        http.assert_not_called()
+        self.assertEqual(conn.vectors[0]["draft_id"], 90)
+        self.assertEqual(conn.vectors[0]["entity_id"], 9001)
+
+    def test目标hash已有其它active_draft_owner时向上抛且绝不迁移(self):
+        conn = _BulkConnection(
+            drafts=[_draft(90, TEXT), _draft(101, TEXT)],
+            vectors=[_vector(700, 90, CONTENT_HASH)],
+        )
+        conn.drafts[90]["status"] = "confirmed"
+        http = Mock()
+
+        with patch.object(embed, "embed_texts", http):
+            with self.assertRaisesRegex(embed.EmbeddingOwnershipConflict, "owner=90"):
+                embed._run_bulk(conn, sess=object(), work_id=None, limit=0)
+
+        http.assert_not_called()
+        self.assertEqual(conn.vectors[0]["draft_id"], 90)
+        self.assertFalse(conn.vectors[0]["deleted"])
+
+    def test_limit范围外候选被entity_owner占用时HTTP前向上抛(self):
+        conn = _BulkConnection(
+            drafts=[_draft(90, NEW_TEXT), _draft(101, TEXT), _draft(102, NEW_TEXT)],
+            vectors=[_vector(700, 90, NEW_HASH, entity_id=9001)],
+        )
+        conn.drafts[90]["status"] = "confirmed"
+        http = Mock()
+
+        with patch.object(embed, "embed_texts", http):
+            with self.assertRaisesRegex(embed.EmbeddingOwnershipConflict, "entity"):
+                embed._run_bulk(conn, sess=object(), work_id=None, limit=1)
+
+        http.assert_not_called()
+        self.assertEqual(len(conn.vectors), 1)
+        self.assertEqual((conn.vectors[0]["draft_id"], conn.vectors[0]["entity_id"]),
+                         (90, 9001))
+
+    def test_limit范围外候选被其它active_owner占用时HTTP前向上抛(self):
+        conn = _BulkConnection(
+            drafts=[_draft(90, NEW_TEXT), _draft(101, TEXT), _draft(102, NEW_TEXT)],
+            vectors=[_vector(700, 90, NEW_HASH)],
+        )
+        conn.drafts[90]["status"] = "confirmed"
+        http = Mock()
+
+        with patch.object(embed, "embed_texts", http):
+            with self.assertRaisesRegex(embed.EmbeddingOwnershipConflict, "owner=90"):
+                embed._run_bulk(conn, sess=object(), work_id=None, limit=1)
+
+        http.assert_not_called()
+        self.assertEqual(len(conn.vectors), 1)
+        self.assertEqual(conn.vectors[0]["draft_id"], 90)
+        self.assertFalse(conn.vectors[0]["deleted"])
+
+    def test_limit前预检全部owner但仅对限量候选发HTTP并写入(self):
+        conn = _BulkConnection(
+            drafts=[_draft(101, TEXT), _draft(102, NEW_TEXT)],
+        )
+        sess = object()
+
+        def respond(_sess, texts):
+            self.assertEqual(texts, [TEXT])
+            self.assertEqual(conn.events.count("owner_read"), 3)
+            self.assertEqual(conn.events[-1], "commit")
+            return [self.VECTOR], set()
+
+        with patch.object(embed, "embed_texts", side_effect=respond) as http:
+            stats = embed._run_bulk(conn, sess=sess, work_id=None, limit=1)
+
+        self.assertEqual(stats, {"done": 1, "skip": 0, "fail": 0})
+        http.assert_called_once()
+        first_commit = conn.events.index("commit")
+        self.assertEqual(conn.events[:first_commit].count("owner_read"), 2)
+        active = [row for row in conn.vectors if not row["deleted"]]
+        self.assertEqual([(row["draft_id"], row["content_hash"]) for row in active],
+                         [(101, CONTENT_HASH)])
+
+    def test读段发现同draft多条活向量时失败关闭并明确报告(self):
+        conn = _BulkConnection(
+            drafts=[_draft(100, TEXT), _draft(101, NEW_TEXT)],
+            vectors=[
+                _vector(701, 101, "old-hash-1"),
+                _vector(702, 101, "old-hash-2"),
+            ],
+        )
+        http = Mock()
+
+        with patch.object(embed, "embed_texts", http), \
+                patch.object(embed.click, "echo") as echo:
+            with self.assertRaisesRegex(embed.EmbeddingOwnershipConflict, "多条活向量"):
+                embed._run_bulk(conn, sess=object(), work_id=None, limit=1)
+
+        http.assert_not_called()
+        self.assertTrue(all(not row["deleted"] for row in conn.vectors))
+        self.assertTrue(any("多条活向量" in call.args[0] for call in echo.call_args_list))
+
+    def test_HTTP后写前出现多条活向量时事务失败关闭并明确报告(self):
+        conn = _BulkConnection(drafts=[_draft(101, TEXT)])
+
+        def race_after_http(_sess, _texts):
+            conn.vectors.extend([
+                _vector(701, 101, "race-old-1"),
+                _vector(702, 101, "race-old-2"),
+            ])
+            return [self.VECTOR], set()
+
+        with patch.object(embed, "embed_texts", side_effect=race_after_http) as http, \
+                patch.object(embed.click, "echo"):
+            with self.assertRaisesRegex(embed.EmbeddingOwnershipConflict, "多条活向量"):
+                embed._run_bulk(conn, sess=object(), work_id=None, limit=0)
+
+        http.assert_called_once()
+        self.assertEqual(len(conn.vectors), 2)
+        self.assertTrue(all(not row["deleted"] for row in conn.vectors))
+
+    def test_HTTP坏结果零写入且下一轮仍可入选(self):
+        bad_results = (
+            ([], set()),
+            ([None], set()),
+            ([self.VECTOR], {0}),
+        )
+        for result in bad_results:
+            with self.subTest(result=result):
+                conn = _BulkConnection(
+                    drafts=[_draft(101, TEXT)],
+                    vectors=[_vector(701, 101, "old-hash")],
+                )
+                with patch.object(embed, "embed_texts", return_value=result):
+                    stats = embed._run_bulk(conn, sess=object(), work_id=None, limit=0)
+
+                self.assertEqual(stats, {"done": 0, "skip": 0, "fail": 1})
+                self.assertFalse(conn.vectors[0]["deleted"])
+                candidate_ids, failures = self._candidate_ids(conn)
+                self.assertEqual(candidate_ids, [101])
+                self.assertEqual(failures, [])
+
+    def test_HTTP网络异常零写入且下一轮仍可入选(self):
+        conn = _BulkConnection(
+            drafts=[_draft(101, TEXT)],
+            vectors=[_vector(701, 101, "old-hash")],
+        )
+
+        with patch.object(embed, "embed_texts", side_effect=OSError("network down")):
+            stats = embed._run_bulk(conn, sess=object(), work_id=None, limit=0)
+
+        self.assertEqual(stats, {"done": 0, "skip": 0, "fail": 1})
+        self.assertFalse(conn.vectors[0]["deleted"])
+        candidate_ids, failures = self._candidate_ids(conn)
+        self.assertEqual(candidate_ids, [101])
+        self.assertEqual(failures, [])
+
+    def test同批目标hash冲突在limit前失败关闭且无人抢owner(self):
+        conn = _BulkConnection(
+            drafts=[_draft(101, TEXT), _draft(102, TEXT)],
+        )
+        http = Mock()
+
+        with patch.object(embed, "embed_texts", http), \
+                patch.object(embed.click, "echo") as echo:
+            with self.assertRaisesRegex(embed.EmbeddingOwnershipConflict, "目标 hash 冲突"):
+                embed._run_bulk(conn, sess=object(), work_id=None, limit=1)
+
+        http.assert_not_called()
+        self.assertEqual(conn.vectors, [])
+        messages = "\n".join(call.args[0] for call in echo.call_args_list)
+        self.assertIn("目标 hash 冲突", messages)
+        self.assertIn("101", messages)
+        self.assertIn("102", messages)
+
+
+class EmbedDraftsCliTest(unittest.TestCase):
+    """验证 CLI 参数在创建 session 和访问外部资源前完成校验。"""
+
+    def test_limit负数在任何session数据库或HTTP前被拒绝(self):
+        connection_context = MagicMock()
+        connection_context.__enter__.return_value = MagicMock()
+        connection_context.__exit__.return_value = False
+
+        with patch.object(embed, "_session") as session, \
+                patch.object(embed.psycopg, "connect", return_value=connection_context) as connect, \
+                patch.object(embed, "_run_bulk") as run_bulk, \
+                patch.object(embed, "embed_texts") as http:
+            result = CliRunner().invoke(embed.main, ["--limit", "-1"])
+
+        self.assertNotEqual(result.exit_code, 0)
+        self.assertIn("--limit", result.output)
+        self.assertIn("x>=0", result.output)
+        session.assert_not_called()
+        connect.assert_not_called()
+        run_bulk.assert_not_called()
+        http.assert_not_called()
+
+
 if __name__ == "__main__":
     unittest.main()

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

@@ -58,17 +58,17 @@ disable-model-invocation: true
 
 **窗行陷阱（放量首日实测）**：`--window` 参数变化后重切,旧窗行会按 from_order 占位,新的大窗被「已有大纲跳过」→ 中间章域永远漏出卡（验收期 1–3 章小窗占住 from_order=1,放量 1–34 章大窗被跳过）。**换窗参数重切前必须先删该书全部窗行**（窗行是可再生中间产物;卡挂「窗起」,cards 重出时按窗软删重出）。
 
-**作品面升格执行器 `scripts/parse_upgrade.py`（与上面范式拆书管线并行的另一条线,命令 `windows`/`run`/`status`）**：把参考书正文按窗抽成「会随剧情长大的实体卡」(升格卡,`source_type=upgrade_book`),设计见 `docs/2026-07-16-升格卡改造设计.md`。里程碑除真实章号外，模型输出必须临时携带所标章节正文短引；系统机械核验后删除证据，错章、缺证据或改写证据均拒收入库并留审计。模型给出的顶层出场章也必须由实体规范名或合法别名在对应章节正文中的实际出现机械证明；合法别名集合同时取卡内 payload 与独立 alias 表，单字规范名禁作章证据，规范名和别名均禁止通用称谓/关系称呼（如队长、舰长）作证据，常规二至四字专名保持精确子串命中。未命中章不得参与立卡、登场兜底或既有卡追加，无实证章时不补登场里程碑；机械登场兜底只允许用 debut 章正文存在性与实体名称/型构造中性台阶，禁止读取跨章摘要。既有卡每窗新增出场章必须审计完整旧值，redo/undo 精确恢复后再按新正文重算。显式 redo **仅允许当前 active 末窗**：执行前机械校验窗口号连续、章域合法且首尾相接，并校验目标窗逐章正文齐全且非空；任一失败必须在快照、undo 与任何写入前非零退出。历史窗修正必须全书前滚重建；后缀级联重算属于 P1，当前不支持。合法末窗 redo 仍须在撤销事务中清除所有 active `upgrade_book` 卡的本窗章域，保留窗外章并为历史无审计数据补旧值审计；redo 前建立完整恢复点（卡 payload/revision/deleted、别名、presence、水位、审计与窗状态），首次失败重新清理再试，最终失败完整恢复且重试初建卡软删。`--max-calls` 不得截断已经开始的即时重试，非显式 redo 不做该全局清理。`run` 默认**不发嵌入**;加 `--semantic-dedup` 开语义判重(治改名/跨型漏并)时,每窗按**读→算→写三段式短连接**跑——观察/近邻召回/M3 终判都在**无长连接**段发 LLM 与嵌入 HTTP(不再持窗级连接跨调用存活),写段只查预判结果落库;实体卡与关系卡的新建/更新都必须同步 state、逐字段审计并进入 touched；并**边抽边嵌**：本窗新建/更新卡在窗事务提交后增量嵌入落库(软删旧向量+upsert 新行),**后窗即可语义召回前窗刚长成的卡**,不再依赖"同书须预先全量 embed"。嵌段失败只告警、不牵连已 done 的窗。
+**作品面升格执行器 `scripts/parse_upgrade.py`（与上面范式拆书管线并行的另一条线,命令 `windows`/`run`/`status`）**：把参考书正文按窗抽成「会随剧情长大的实体卡」(升格卡,`source_type=upgrade_book`),设计见 `docs/2026-07-16-升格卡改造设计.md`。里程碑除真实章号外，模型输出必须临时携带所标章节正文短引；系统机械核验后删除证据，错章、缺证据或改写证据均拒收入库并留审计。模型给出的顶层出场章也必须由实体规范名或合法别名在对应章节正文中的实际出现机械证明；合法别名集合同时取卡内 payload 与独立 alias 表，单字规范名禁作章证据，规范名和别名均禁止通用称谓/关系称呼（如队长、舰长）作证据，常规二至四字专名保持精确子串命中。未命中章不得参与立卡、登场兜底或既有卡追加，无实证章时不补登场里程碑；机械登场兜底只允许用 debut 章正文存在性与实体名称/型构造中性台阶，禁止读取跨章摘要。既有卡每窗新增出场章必须审计完整旧值，redo/undo 精确恢复后再按新正文重算。任何送入模型的既有卡 payload 必须同步绑定读取时 revision；写事务锁行后必须仍为 `pending + upgrade_book` 且 revision 一致，否则整窗失败/重试，禁止以旧模型结果覆盖外部改版。关系 prompt 的每个参与角色也必须先读取 expected revision、再锁定并复验 status/source/revision，只有锁后 payload 可送模型，新建和更新关系都只能引用通过该门禁的角色。显式 redo **仅允许当前 active 末窗**：执行前机械校验窗口号连续、章域合法且首尾相接，并校验目标窗逐章正文齐全且非空；任一失败必须在快照、undo 与任何写入前非零退出。历史窗修正必须全书前滚重建；后缀级联重算属于 P1，当前不支持。合法末窗 redo 仍须在撤销事务中清除所有 active `upgrade_book` 卡的本窗章域，保留窗外章并为历史无审计数据补旧值审计；redo 前建立完整恢复点（卡 payload/status/revision/deleted、别名、presence、水位、审计与窗状态），首次失败重新清理再试，最终失败完整恢复且重试初建卡软删。`--max-calls` 不得截断已经开始的即时重试，非显式 redo 不做该全局清理。`run` 默认**不发嵌入**;加 `--semantic-dedup` 开语义判重(治改名/跨型漏并)时,每窗按**读→算→写三段式短连接**跑——观察/近邻召回/M3 终判都在**无长连接**段发 LLM 与嵌入 HTTP(不再持窗级连接跨调用存活),写段只查预判结果落库;实体卡与关系卡的新建/更新都必须同步 state、逐字段审计并进入 touched；并**边抽边嵌**：本窗新建/更新卡在窗事务提交后增量嵌入落库(软删旧向量+upsert 新行),**后窗即可语义召回前窗刚长成的卡**,不再依赖"同书须预先全量 embed"。普通嵌段服务失败只告警；确定性 owner 冲突先持久化 durable `compensation-failed` marker，再由普通窗按 touched fence、显式 redo 按整书六域全域 fence 补偿，任一外部漂移都禁止覆盖。marker 窗下次 run 禁止自动 undo。
 
 **登场中性台阶名称边界**：名称只能取 debut 章正文唯一实际命中的合法规范名或别名；若只命中旧别名就用旧别名，多个合法名称同时命中、仅命中通用称谓或无法唯一确定时退化为“人物登场/物件登场”等仅类型描述，禁止泄漏未来才形成的规范名。`new_card` 规范化名称后必须把实际 canonical 与合法 aliases 立即登记到本窗判重索引，不能继续使用模型原始括号名。
 
-**嵌入唯一键冲突失败关闭**：前述“嵌段失败只告警”仅指普通网络或服务异常；owner 预检必须在 `embed_texts` 前覆盖全部 todo hash。写连接第一条 SQL 必须按固定顺序以 `ROW EXCLUSIVE` 锁 `muse_knowledge_draft, example_knowledge_embedding`；随后按 did 锁 draft 行，机械确认 tenant、deleted=false、status=pending，并以当前 payload+MODEL 重算 hash 与 ready hash 一致，再按 did 锁全部活向量、按 hash 以 `FOR UPDATE` 二次校验 owner。若 HTTP 期间 payload 漂移，`(tenant_id, content_hash, model)` 唯一键被两个活跃 draft 的当前 payload 同时声明，唯一行 `entity_id` 非空，或同 draft 任一旧活向量已有 `entity_id`，必须在任何向量写入前硬停、把窗置 failed 并令命令非零；禁止迁移 entity owner、清空 `entity_id`、软删实体向量或打印窗口完成。仅当唯一行没有 entity owner 且命中旧软删 draft 时，才把行迁到当前 draft并刷新当前嵌入语义列；冲突更新及旧 hash 软删条件都必须保留 `entity_id IS NULL` 防并发竞态。
+**嵌入唯一键冲突失败关闭**：前述“嵌段失败只告警”仅指普通网络或服务异常；owner 预检必须在 `embed_texts` 前覆盖全部 todo hash。写连接第一条 SQL 必须按固定顺序以 `ROW EXCLUSIVE` 锁 `muse_knowledge_draft, example_knowledge_embedding`；随后按 did 锁 draft 行，机械确认 tenant、deleted=false、status=pending，并以当前 payload+MODEL 重算 hash 与 ready hash 一致，再按 did 锁全部活向量、按 hash 以 `FOR UPDATE` 二次校验 owner。若 HTTP 期间 payload 漂移，`(tenant_id, content_hash, model)` 唯一键被两个活跃 draft 的当前 payload 同时声明，唯一行 `entity_id` 非空，或同 draft 任一旧活向量已有 `entity_id`，必须在任何向量写入前硬停并令命令非零；禁止迁移 entity owner、清空 `entity_id`、软删实体向量或打印窗口完成。硬停后先以窗口 `done/null/upgrade` CAS 独立提交 durable `compensation-failed` marker；marker 失败不得继续补偿。普通窗随后锁定并 exact 复验 touched draft 的 status/source/revision/payload hash，通过后才允许同事务 `undo→最终 failed`；显式 redo 则按固定顺序阻写 draft/alias/presence/card_state/audit/window 六域，exact 复验 redo postcommit 全域快照（draft 含 payload/status/revision/updater/deleted），通过后才恢复 pre-redo 全域与原窗状态。任一外部确认、删除、改版或辅助域漂移都禁止 undo/restore，保留 marker；marker 窗下次 run 必须非零阻断，禁止自动 undo。仅当唯一行没有 entity owner 且命中旧软删 draft 时，才把行迁到当前 draft并刷新当前嵌入语义列；冲突更新及旧 hash 软删条件都必须保留 `entity_id IS NULL` 防并发竞态。
 
 **同书命令互斥**：`parse_upgrade.py run/windows`、`reset_upgrade_work.py` 的预览/执行，以及 `backup_upgrade_work.py backup/rehearse/restore` 均须先取得 `scripts/upgrade_work_lock.py` 的同租户同作品 PostgreSQL session advisory lock；失败必须在任何业务 SQL、文件 verify/写入、嵌入或 LLM 调用前非零退出。锁由独立 autocommit 连接持有到命令结束，该连接只执行加锁/解锁 SQL；`status` 只读且不取锁。所有调用方必须导入同一个 `upgrade_work_lock(...)` context manager，禁止另造不兼容锁键。
 
 **升格 work reset 前备份（只读）**：任何 `reset_upgrade_work.py --execute` 前，先用单个 `REPEATABLE READ READ ONLY` 事务导出该作品七域状态；目录必须是 `/private/tmp` 下尚不存在的新目录。input 摘要在同一快照中绑定不含密码的数据库 identity、Canonical 章节/block 内容、active 窗边界、升格七型 active 字段合同，以及计划参数 `MiniMax-M3 + semantic-dedup=true`；同时绑定本次实际执行的 `backup_upgrade_work.py`、`reset_upgrade_work.py`、`parse_upgrade.py`、`parse_llm.py`、`embed_drafts.py`、`upgrade_work_lock.py`、`llm.py` 和本 SKILL 的 fileSha。`gitCommit` 只记录 HEAD；当代码未提交时不得把它当执行代码身份，确认以 manifest 的 `inputSha + codeFiles + confirmationSha` 为准。SHA 仅用于发现传输、落盘或误操作造成的意外损坏，不宣称抵抗能同时改写工件和摘要的恶意篡改。命令默认不写数据库、不调用模型，生成后会关闭数据库连接并从磁盘独立复验；任一失败删除本次新建目录并非零退出。
 
-reset preview 不要求备份且只读；execute 必须同时提供 `--backup-dir`、与 manifest 精确匹配的 `--backup-id` 和 `--confirmation-sha`。取得统一同书 advisory lock 后，execute 须在同一业务事务内先以 `SHARE ROW EXCLUSIVE` 锁住 drafts/windows/aliases/presence/card_state/audits/embeddings 七表，再复用 `backup_upgrade_work` 的七域读取与摘要规则，将当前 `rowCount/primaryKeySha/contentSha` 和已离线复验 manifest 逐域 exact 比较；任一漂移必须在写入前失败关闭。
+reset preview 不要求备份且只读；execute 必须同时提供 `--backup-dir`、与 manifest 精确匹配的 `--backup-id` 和 `--confirmation-sha`。取得统一同书 advisory lock 后，execute 须在同一业务事务内以 `SHARE ROW EXCLUSIVE` 按固定顺序锁住 4 个输入源表 `muse_content_chapter→muse_content_block→muse_meta_schema→muse_meta_schema_version`，再锁 7 个产出表 `muse_knowledge_draft→example_upgrade_window→example_upgrade_alias→example_upgrade_presence→example_upgrade_card_state→example_upgrade_audit→example_knowledge_embedding`；必须在锁内重算完整 input，并复用 `backup_upgrade_work` 的七域读取与摘要规则，将当前完整 input 及七域 `rowCount/primaryKeySha/contentSha` 与已离线复验 manifest exact 比较；任一漂移必须在写入前失败关闭。
 
 reset execute 还必须在同书 advisory lock 内、创建业务连接前重新调用 `capture_code_identity`：manifest `codeFiles` 必须包含 `reset_upgrade_work.py`，并与当前实际完整 `codeFiles` 逐项一致；reset/parse/embed/llm/skill 任一 fileSha 漂移或集合变化均拒绝。旧备份仍可离线 `verify`，但缺少 reset fileSha 时不得用于 execute。
 

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

@@ -88,6 +88,21 @@ class EmbeddingOwnershipConflict(RuntimeError):
     """同一嵌入唯一键被两个活跃 draft 的当前 payload 同时声明。"""
 
 
+class UpgradeDraftWriteConflict(RuntimeError):
+    """升格卡已离开可写状态，或锁定后的 revision 写入失败。"""
+
+
+class AliasOwnershipConflict(RuntimeError):
+    """同书同别名已唯一映射到另一张 canonical 卡。"""
+
+
+class CompensationFenceConflict(RuntimeError):
+    """窗提交后的 draft 已被外部改动，禁止补偿覆盖。"""
+
+
+COMPENSATION_FAILED_PREFIX = "compensation-failed:"
+
+
 # ── 库内合同（元数据驱动公理：prompt 与守卫同源，禁手写合同）──
 
 def load_entity_contracts(conn):
@@ -176,15 +191,19 @@ def load_known(conn, work_id):
     aliases_by_draft 同时汇总 payload 与独立别名表，供后续正文实体命中过滤使用。"""
     name_map = {}
     aliases_by_draft = {}
-    for did, entity_type, name, brief, aliases in conn.execute(
+    revisions = {}
+    for did, entity_type, name, brief, aliases, revision in conn.execute(
             """SELECT id,
                       draft_payload->>'type',
                       draft_payload->>'名称',
                       COALESCE(draft_payload->>'一句话摘要',''),
-                      COALESCE(draft_payload->'别名','[]'::jsonb)
+                      COALESCE(draft_payload->'别名','[]'::jsonb),
+                      revision
                FROM muse_knowledge_draft
-               WHERE tenant_id=%s AND work_id=%s AND source_type=%s AND deleted=FALSE""",
+               WHERE tenant_id=%s AND work_id=%s AND source_type=%s
+                 AND status='pending' AND deleted=FALSE""",
             (TENANT, work_id, SOURCE_TYPE)).fetchall():
+        revisions[did] = revision
         t, nm = entity_type or "", (name or "").strip()
         if nm:
             name_map[nm] = (did, t, brief)
@@ -208,7 +227,7 @@ def load_known(conn, work_id):
             "WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE",
             (TENANT, work_id)).fetchall():
         presence.setdefault((t, nm), set()).add(ch)
-    return name_map, presence, aliases_by_draft
+    return name_map, presence, aliases_by_draft, revisions
 
 
 def prescan(name_map, text):
@@ -350,6 +369,38 @@ def _clean_alias(al):
     return al
 
 
+def _claim_alias(conn, work_id, canonical_name, alias, win_no, verdict_by):
+    """原子登记别名：仅活跃同 canonical 幂等，冲突或 tombstone 必须失败关闭。"""
+
+    alias = _clean_alias(alias)
+    if not alias or alias == canonical_name:
+        return None
+    row = conn.execute(
+        """INSERT INTO example_upgrade_alias
+               (work_id, canonical_name, alias, evidence_window, verdict_by, tenant_id)
+           VALUES (%s,%s,%s,%s,%s,%s)
+           ON CONFLICT (tenant_id, work_id, alias) DO UPDATE
+           SET canonical_name=EXCLUDED.canonical_name
+           WHERE example_upgrade_alias.canonical_name=EXCLUDED.canonical_name
+             AND example_upgrade_alias.deleted=FALSE
+           RETURNING canonical_name""",
+        (work_id, canonical_name, alias, win_no, verdict_by, TENANT),
+    ).fetchone()
+    if row and row[0] == canonical_name:
+        return alias
+    owner = conn.execute(
+        """SELECT canonical_name, deleted FROM example_upgrade_alias
+           WHERE tenant_id=%s AND work_id=%s AND alias=%s FOR UPDATE""",
+        (TENANT, work_id, alias),
+    ).fetchone()
+    owner_name = owner[0] if owner else "<唯一行缺失>"
+    owner_state = "deleted" if owner and owner[1] else "active"
+    raise AliasOwnershipConflict(
+        f"别名唯一映射冲突：work={work_id}，alias={alias}，"
+        f"现有canonical={owner_name}，现有状态={owner_state}，请求canonical={canonical_name}"
+    )
+
+
 def _is_garbage(text):
     """结构垃圾检测（抽检 H2 根治）：窗75 实测模型把原始变更包字符串化塞进字段值，
     7 张卡被 {'draft_id':...} 类程序结构污染。含结构特征的文本一律拒收留审计。"""
@@ -474,6 +525,8 @@ GENERIC_ADDRESS_ALIASES = frozenset({
 })
 TOP_LEVEL_APPEARANCE_AUDIT_FIELD = "顶层:出场章"
 # 关系卡顶层字段与卡水位不在 payload["字段"] 内，必须用明确 sentinel 让 undo 精确落回原位置。
+TOP_LEVEL_ALIASES_AUDIT_FIELD = "顶层:别名"
+TOP_LEVEL_SUMMARY_AUDIT_FIELD = "顶层:一句话摘要"
 TOP_LEVEL_RELATION_TYPE_AUDIT_FIELD = "顶层:关系类型"
 TOP_LEVEL_RELATION_EVOLUTION_AUDIT_FIELD = "顶层:演变轨迹"
 CARD_STATE_WATERMARK_AUDIT_FIELD = "状态:watermark_window"
@@ -558,6 +611,68 @@ def _append_verified_appearance_chapters(
     return True
 
 
+def _read_upgrade_draft_snapshot(conn, draft_id):
+    """读取送入模型的可信 payload/revision；不持锁跨模型调用。"""
+
+    row = conn.execute(
+        """SELECT draft_payload, status, revision, source_type
+           FROM muse_knowledge_draft
+           WHERE id=%s AND tenant_id=%s AND deleted=FALSE""",
+        (draft_id, TENANT),
+    ).fetchone()
+    if not row:
+        raise UpgradeDraftWriteConflict(f"升格 draft 不存在或已删除：draft={draft_id}")
+    payload, status, revision, source_type = row
+    if status != "pending" or source_type != SOURCE_TYPE:
+        raise UpgradeDraftWriteConflict(
+            f"升格 draft 不可写：draft={draft_id}，status={status}，source_type={source_type}"
+        )
+    return deepcopy(payload), revision
+
+
+def _lock_upgrade_draft(conn, draft_id, expected_revision=None):
+    """锁定当前 draft；状态、来源或模型所见 revision 漂移都失败关闭。"""
+
+    row = conn.execute(
+        """SELECT draft_payload, status, revision, source_type
+           FROM muse_knowledge_draft
+           WHERE id=%s AND tenant_id=%s AND deleted=FALSE
+           FOR UPDATE""",
+        (draft_id, TENANT),
+    ).fetchone()
+    if not row:
+        raise UpgradeDraftWriteConflict(f"升格 draft 不存在或已删除：draft={draft_id}")
+    payload, status, revision, source_type = row
+    if status != "pending" or source_type != SOURCE_TYPE:
+        raise UpgradeDraftWriteConflict(
+            f"升格 draft 不可写：draft={draft_id}，status={status}，source_type={source_type}"
+        )
+    if expected_revision is not None and revision != expected_revision:
+        raise UpgradeDraftWriteConflict(
+            f"升格 draft 模型版本冲突：draft={draft_id}，"
+            f"expected_revision={expected_revision}，current_revision={revision}"
+        )
+    return deepcopy(payload), revision
+
+
+def _write_locked_upgrade_draft(conn, draft_id, payload, revision):
+    """用锁后 revision 回写；任何状态或版本漂移都按确定性冲突失败关闭。"""
+
+    row = conn.execute(
+        """UPDATE muse_knowledge_draft SET draft_payload=%s,
+               revision=revision+1, updater='upgrade'
+           WHERE id=%s AND tenant_id=%s AND deleted=FALSE
+             AND status='pending' AND source_type=%s AND revision=%s
+           RETURNING revision""",
+        (json.dumps(payload, ensure_ascii=False), draft_id, TENANT, SOURCE_TYPE, revision),
+    ).fetchone()
+    if not row:
+        raise UpgradeDraftWriteConflict(
+            f"升格 draft 锁后写入冲突：draft={draft_id}，expected_revision={revision}"
+        )
+    return row[0]
+
+
 def _audit_relation_change(conn, draft_id, win_no, field_name, old_value, new_value):
     """记录关系卡一个真实落点的完整旧/新 JSON；值未变化时不制造空审计。"""
 
@@ -579,8 +694,8 @@ def _audit_relation_change(conn, draft_id, win_no, field_name, old_value, new_va
     return True
 
 
-def _update_relation_card_state(conn, draft_id, work_id, win_no):
-    """推进关系卡水位并审计旧值；undo 据 sentinel 恢复旧水位或删除本窗新 state。"""
+def _update_card_state(conn, draft_id, work_id, win_no):
+    """推进实体或关系卡水位并审计旧值；undo 可精确恢复或删除新 state。"""
 
     row = conn.execute(
         "SELECT watermark_window FROM example_upgrade_card_state WHERE draft_id=%s",
@@ -606,10 +721,10 @@ def _update_relation_card_state(conn, draft_id, work_id, win_no):
     )
 
 
-def _update_relation_card(conn, work_id, win_no, draft_id, payload, relation):
-    """按真实字段落点更新关系卡，并为顶层、逐字段、演变数组和水位分别留可撤销审计。"""
+def _update_relation_card(conn, work_id, win_no, draft_id, expected_revision, relation):
+    """锁后按当前 payload 更新关系卡，并校验模型读取时的 revision。"""
 
-    payload = deepcopy(payload)
+    payload, revision = _lock_upgrade_draft(conn, draft_id, expected_revision)
     fields = payload.setdefault("字段", {})
 
     # 历史数据可能把演变轨迹放在顶层；迁移本身也必须可撤销，不能只恢复字段侧。
@@ -662,12 +777,8 @@ def _update_relation_card(conn, work_id, win_no, draft_id, payload, relation):
                 new_relation_type):
             payload["关系类型"] = new_relation_type
 
-    conn.execute(
-        """UPDATE muse_knowledge_draft SET draft_payload=%s,
-               revision=revision+1, updater='upgrade' WHERE id=%s""",
-        (json.dumps(payload, ensure_ascii=False), draft_id),
-    )
-    _update_relation_card_state(conn, draft_id, work_id, win_no)
+    _write_locked_upgrade_draft(conn, draft_id, payload, revision)
+    _update_card_state(conn, draft_id, work_id, win_no)
     return draft_id
 
 
@@ -692,7 +803,7 @@ def _insert_relation_card(conn, work_id, win_no, payload):
     )
     for field_name, value in (payload.get("字段") or {}).items():
         _audit_relation_change(conn, draft_id, win_no, field_name, None, value)
-    _update_relation_card_state(conn, draft_id, work_id, win_no)
+    _update_card_state(conn, draft_id, work_id, win_no)
     return draft_id
 
 
@@ -919,13 +1030,13 @@ def merge_card(
     chapter_texts=None,
     appearance_chapters=None,
     known_aliases=None,
+    expected_revision=None,
 ):
     """按 5.1 三类规则合并变更字段：数组/白名单=追加（剥模型自带前缀+同文去重后带窗号），
     标量=覆写留审计（增量式假全量拦截转追加——抽检#4 信息回退病）。
     valid_keys：该型合同的合法字段 key 集——越合同 key 裁剪留审计（窗29实测模型把
     整段条目文本误当字段 key 写入，无校验会把卡体字段区打烂）。"""
-    payload = conn.execute("SELECT draft_payload FROM muse_knowledge_draft WHERE id=%s",
-                           (draft_id,)).fetchone()[0]
+    payload, revision = _lock_upgrade_draft(conn, draft_id, expected_revision)
     fields = payload.setdefault("字段", {})
     # 卡水位（该卡最后一次被更新的窗号）：补跑迟到窗（如窗41在窗80后补跑）的覆写类字段
     # 若直接落卡，会把书末态倒写回中期态（时间倒流污染，2026-07-15 补跑实测坐实）。
@@ -969,14 +1080,16 @@ def merge_card(
                         (draft_id, win_no, "迟到覆写弃用:一句话摘要",
                          json.dumps(v, ensure_ascii=False)[:2000], TENANT))
                 else:
-                    conn.execute(
-                        """INSERT INTO example_upgrade_audit
-                               (draft_id, window_no, field_name, old_value, new_value, tenant_id)
-                           VALUES (%s,%s,%s,%s,%s,%s)""",
-                        (draft_id, win_no, "一句话摘要",
-                         json.dumps(payload.get("一句话摘要"), ensure_ascii=False),
-                         json.dumps(v.strip(), ensure_ascii=False), TENANT))
-                    payload["一句话摘要"] = v.strip()
+                    old_summary = deepcopy(payload.get("一句话摘要")) \
+                        if "一句话摘要" in payload else None
+                    if _audit_relation_change(
+                            conn,
+                            draft_id,
+                            win_no,
+                            TOP_LEVEL_SUMMARY_AUDIT_FIELD,
+                            old_summary,
+                            v.strip()):
+                        payload["一句话摘要"] = v.strip()
             continue
         if k in MILESTONE_FIELDS:
             # 里程碑对象数组（演变历程）：走对象合并路径——去重按台阶内容、排序按真实章号，
@@ -1056,30 +1169,26 @@ def merge_card(
                  json.dumps(v, ensure_ascii=False), TENANT))
             # 覆写值同样剥尾部拼接残渣（批9 样张走查同款病灶，覆写路一并守住）
             fields[k] = _strip_tail(v) if isinstance(v, str) else v
-    for al0 in alias_new or []:
-        al = _clean_alias(al0)
-        if al and al != payload.get("名称"):
-            payload.setdefault("别名", [])
-            if al not in payload["别名"]:
-                payload["别名"].append(al)
-            conn.execute(
-                """INSERT INTO example_upgrade_alias
-                       (work_id, canonical_name, alias, evidence_window, verdict_by, tenant_id)
-                   VALUES (%s,%s,%s,%s,'ai',%s)
-                   ON CONFLICT (tenant_id, work_id, alias) DO NOTHING""",
-                (payload.get("_work_id") or 0, payload.get("名称"), al, win_no, TENANT))
-    conn.execute(
-        """UPDATE muse_knowledge_draft SET draft_payload=%s, revision=revision+1,
-               updater='upgrade' WHERE id=%s""",
-        (json.dumps(payload, ensure_ascii=False), draft_id))
-    conn.execute(
-        """INSERT INTO example_upgrade_card_state (draft_id, work_id, watermark_window, tenant_id)
-           VALUES (%s,%s,%s,%s)
-           ON CONFLICT (draft_id) DO UPDATE
-           SET watermark_window=GREATEST(example_upgrade_card_state.watermark_window,
-                                         EXCLUDED.watermark_window),
-               update_time=now()""",
-        (draft_id, payload.get("_work_id") or 0, win_no, TENANT))
+    old_aliases = deepcopy(payload.get("别名")) if "别名" in payload else None
+    merged_aliases = list(old_aliases) if isinstance(old_aliases, list) else []
+    work_id = payload.get("_work_id") or 0
+    canonical_name = payload.get("名称")
+    alias_added = False
+    for raw_alias in alias_new or []:
+        alias = _claim_alias(conn, work_id, canonical_name, raw_alias, win_no, "ai")
+        if alias and alias not in merged_aliases:
+            merged_aliases.append(alias)
+            alias_added = True
+    if alias_added and _audit_relation_change(
+            conn,
+            draft_id,
+            win_no,
+            TOP_LEVEL_ALIASES_AUDIT_FIELD,
+            old_aliases,
+            merged_aliases):
+        payload["别名"] = merged_aliases
+    _write_locked_upgrade_draft(conn, draft_id, payload, revision)
+    _update_card_state(conn, draft_id, work_id, win_no)
 
 
 def new_card(
@@ -1159,6 +1268,9 @@ def new_card(
                "来源": f"升格@窗{win_no}", "状态": "草稿",
                "目标库": "本书作品库", "可见范围": "本书私有",
                "_work_id": work_id}
+    # 先原子占用全部 alias；任一不同 canonical 冲突都会让窗事务在写 payload 前失败关闭。
+    for alias in payload["别名"]:
+        _claim_alias(conn, work_id, raw, alias, win_no, "init")
     did = conn.execute(
         """INSERT INTO muse_knowledge_draft
                (work_id, draft_type, draft_payload, status, source_type, source_id,
@@ -1172,15 +1284,6 @@ def new_card(
                    (draft_id, window_no, field_name, old_value, new_value, tenant_id)
                VALUES (%s,%s,%s,NULL,%s,%s)""",
             (did, win_no, k, json.dumps(v, ensure_ascii=False), TENANT))
-    # 初始别名同步入 alias 表（抽检 L3 根源：立卡别名只写卡内不入表，
-    # 别名表覆盖率仅 26%，"别名精确判重"层大面积空转导致同人两卡漏并）
-    for al in payload["别名"]:
-        conn.execute(
-            """INSERT INTO example_upgrade_alias
-                   (work_id, canonical_name, alias, evidence_window, verdict_by, tenant_id)
-               VALUES (%s,%s,%s,%s,'init',%s)
-               ON CONFLICT (tenant_id, work_id, alias) DO NOTHING""",
-            (work_id, raw, al, win_no, TENANT))
     conn.execute(
         """INSERT INTO example_upgrade_card_state (draft_id, work_id, watermark_window, tenant_id)
            VALUES (%s,%s,%s,%s) ON CONFLICT (draft_id) DO NOTHING""",
@@ -1194,29 +1297,33 @@ def _snapshot_redo_window(conn, work_id, win_no):
     snapshot = {
         "window_no": win_no,
         "drafts": conn.execute(
-            """SELECT id, draft_payload, revision, updater, deleted
+            """SELECT id, draft_payload, status, revision, updater, deleted
                FROM muse_knowledge_draft
-               WHERE tenant_id=%s AND work_id=%s AND source_type=%s""",
+               WHERE tenant_id=%s AND work_id=%s AND source_type=%s
+               ORDER BY id""",
             (TENANT, work_id, SOURCE_TYPE),
         ).fetchall(),
         "aliases": conn.execute(
             """SELECT id, canonical_name, alias, evidence_window, verdict_by,
                       creator, create_time, updater, update_time, deleted
                FROM example_upgrade_alias
-               WHERE tenant_id=%s AND work_id=%s""",
+               WHERE tenant_id=%s AND work_id=%s
+               ORDER BY id""",
             (TENANT, work_id),
         ).fetchall(),
         "presence": conn.execute(
             """SELECT id, window_no, chapter_no, entity_type, name, observation,
                       creator, create_time, deleted
                FROM example_upgrade_presence
-               WHERE tenant_id=%s AND work_id=%s""",
+               WHERE tenant_id=%s AND work_id=%s
+               ORDER BY id""",
             (TENANT, work_id),
         ).fetchall(),
         "card_states": conn.execute(
             """SELECT draft_id, watermark_window, update_time
                FROM example_upgrade_card_state
-               WHERE tenant_id=%s AND work_id=%s""",
+               WHERE tenant_id=%s AND work_id=%s
+               ORDER BY draft_id""",
             (TENANT, work_id),
         ).fetchall(),
         "audits": conn.execute(
@@ -1224,7 +1331,8 @@ def _snapshot_redo_window(conn, work_id, win_no):
                       a.old_value, a.new_value, a.create_time
                FROM example_upgrade_audit a
                JOIN muse_knowledge_draft d ON d.id=a.draft_id
-               WHERE a.tenant_id=%s AND d.work_id=%s AND d.source_type=%s""",
+               WHERE a.tenant_id=%s AND d.work_id=%s AND d.source_type=%s
+               ORDER BY a.id""",
             (TENANT, work_id, SOURCE_TYPE),
         ).fetchall(),
         "window": conn.execute(
@@ -1236,6 +1344,26 @@ def _snapshot_redo_window(conn, work_id, win_no):
     return deepcopy(snapshot)
 
 
+def _assert_redo_postcommit_fence(conn, work_id, expected_snapshot):
+    """阻写 redo 恢复覆盖域并 exact 复验提交态；任一域漂移都禁止整书恢复。"""
+
+    if expected_snapshot is None:
+        raise CompensationFenceConflict("redo 补偿缺少 postcommit 全域围栏")
+    # 固定表序一次取得阻写锁；等待中的外部写事务提交后，READ COMMITTED 重读必能看到并拒绝漂移。
+    conn.execute(
+        """LOCK TABLE muse_knowledge_draft, example_upgrade_alias,
+                      example_upgrade_presence, example_upgrade_card_state,
+                      example_upgrade_audit, example_upgrade_window
+           IN SHARE ROW EXCLUSIVE MODE"""
+    )
+    current_snapshot = _snapshot_redo_window(
+        conn, work_id, expected_snapshot["window_no"]
+    )
+    for domain in ("drafts", "aliases", "presence", "card_states", "audits", "window"):
+        if current_snapshot.get(domain) != expected_snapshot.get(domain):
+            raise CompensationFenceConflict(f"redo 补偿全域围栏冲突：domain={domain}")
+
+
 def _restore_redo_window(conn, work_id, snapshot):
     """原子恢复 redo 前状态；重试中新建卡只做软删，保留全局编号与引用安全。"""
 
@@ -1250,12 +1378,12 @@ def _restore_redo_window(conn, work_id, snapshot):
     }
     for draft_id in current_draft_ids - old_draft_ids:
         conn.execute("UPDATE muse_knowledge_draft SET deleted=TRUE WHERE id=%s", (draft_id,))
-    for draft_id, payload, revision, updater, deleted in snapshot["drafts"]:
+    for draft_id, payload, status, revision, updater, deleted in snapshot["drafts"]:
         conn.execute(
             """UPDATE muse_knowledge_draft
-               SET draft_payload=%s, revision=%s, updater=%s, deleted=%s
+               SET draft_payload=%s, status=%s, revision=%s, updater=%s, deleted=%s
                WHERE id=%s AND tenant_id=%s AND work_id=%s AND source_type=%s""",
-            (json.dumps(payload, ensure_ascii=False), revision, updater, deleted,
+            (json.dumps(payload, ensure_ascii=False), status, revision, updater, deleted,
              draft_id, TENANT, work_id, SOURCE_TYPE),
         )
 
@@ -1356,6 +1484,18 @@ def undo_window(conn, work_id, win_no, *, from_chapter=None, to_chapter=None):
                 payload.pop("出场章", None)
             else:
                 payload["出场章"] = json.loads(old) if isinstance(old, str) else old
+        elif fname == TOP_LEVEL_ALIASES_AUDIT_FIELD:
+            # 顶层别名与唯一表必须一起撤销；旧值为空表示原 payload 连该键都不存在。
+            if old is None:
+                payload.pop("别名", None)
+            else:
+                payload["别名"] = json.loads(old) if isinstance(old, str) else old
+        elif fname == TOP_LEVEL_SUMMARY_AUDIT_FIELD:
+            # 一句话摘要住顶层，禁止按普通字段恢复成 payload["字段"] 内的影子摘要。
+            if old is None:
+                payload.pop("一句话摘要", None)
+            else:
+                payload["一句话摘要"] = json.loads(old) if isinstance(old, str) else old
         elif fname == TOP_LEVEL_RELATION_TYPE_AUDIT_FIELD:
             # 关系类型住 payload 顶层；普通字段恢复会错误写到 payload["字段"]。
             if old is None:
@@ -1372,8 +1512,12 @@ def undo_window(conn, work_id, win_no, *, from_chapter=None, to_chapter=None):
             payload.get("字段", {}).pop(fname, None)
         else:
             payload.setdefault("字段", {})[fname] = json.loads(old)
-        conn.execute("UPDATE muse_knowledge_draft SET draft_payload=%s WHERE id=%s",
-                     (json.dumps(payload, ensure_ascii=False), did))
+        conn.execute(
+            """UPDATE muse_knowledge_draft
+               SET draft_payload=%s, revision=revision+1, updater='upgrade-undo'
+               WHERE id=%s""",
+            (json.dumps(payload, ensure_ascii=False), did),
+        )
     conn.execute("""DELETE FROM example_upgrade_audit WHERE tenant_id=%s AND window_no=%s
                     AND draft_id IN (SELECT id FROM muse_knowledge_draft WHERE work_id=%s)""",
                  (TENANT, win_no, work_id))
@@ -1384,7 +1528,12 @@ def undo_window(conn, work_id, win_no, *, from_chapter=None, to_chapter=None):
                WHERE tenant_id=%s AND work_id=%s AND source_type=%s AND deleted=FALSE""",
             (TENANT, work_id, SOURCE_TYPE)).fetchall():
         if payload.get("来源") == f"升格@窗{win_no}":
-            conn.execute("UPDATE muse_knowledge_draft SET deleted=TRUE WHERE id=%s", (did,))
+            conn.execute(
+                """UPDATE muse_knowledge_draft
+                   SET deleted=TRUE, revision=revision+1, updater='upgrade-undo'
+                   WHERE id=%s""",
+                (did,),
+            )
             # 同步清卡水位行（fable 复验实证：深空回滚删 400+ 初卡后 card_state 僵尸行
             # 残留，按 draft JOIN 不过滤 deleted 的查询会捞出僵尸卡）
             conn.execute("DELETE FROM example_upgrade_card_state WHERE draft_id=%s", (did,))
@@ -1422,8 +1571,12 @@ def undo_window(conn, work_id, win_no, *, from_chapter=None, to_chapter=None):
                 if len(nv) != len(v):
                     payload["字段"][k], changed = nv, True
         if changed:
-            conn.execute("UPDATE muse_knowledge_draft SET draft_payload=%s WHERE id=%s",
-                         (json.dumps(payload, ensure_ascii=False), did))
+            conn.execute(
+                """UPDATE muse_knowledge_draft
+                   SET draft_payload=%s, revision=revision+1, updater='upgrade-undo'
+                   WHERE id=%s""",
+                (json.dumps(payload, ensure_ascii=False), did),
+            )
     conn.execute("DELETE FROM example_upgrade_presence WHERE tenant_id=%s AND work_id=%s AND window_no=%s",
                  (TENANT, work_id, win_no))
     conn.execute("""DELETE FROM example_upgrade_alias WHERE tenant_id=%s AND work_id=%s
@@ -1689,8 +1842,7 @@ def _assert_ready_draft_current(conn, draft_id, expected_hash):
             f"待写 draft 状态非法：draft={draft_id}，tenant={tenant_id}，"
             f"deleted={deleted}，status={status}"
         )
-    current_text = build_embed_text(payload or {})
-    current_hash = hashlib.sha256(f"{current_text}|{EMBED_MODEL}".encode()).hexdigest()
+    current_hash = _payload_embed_hash(payload)
     if current_hash != expected_hash:
         raise EmbeddingOwnershipConflict(
             f"HTTP 期间 draft payload 已漂移：draft={draft_id}，"
@@ -1698,6 +1850,13 @@ def _assert_ready_draft_current(conn, draft_id, expected_hash):
         )
 
 
+def _payload_embed_hash(payload):
+    """按嵌入唯一键同源公式计算当前 payload hash。"""
+
+    text = build_embed_text(payload or {})
+    return hashlib.sha256(f"{text}|{EMBED_MODEL}".encode()).hexdigest()
+
+
 def embed_touched_cards(sess, work_id, touched):
     """嵌段（洞① 三段式连接）：窗事务 commit **之后**，把本窗新建/更新的卡增量嵌入落库，
     让后窗语义判重能召回前窗刚长成的卡（治「起了对不上正文的名→机械预扫认不出→成长线静默断」）。
@@ -1728,7 +1887,7 @@ def embed_touched_cards(sess, work_id, touched):
         todo = []                            # [(did, text, hash)…]
         for did, payload in payloads.items():
             text = build_embed_text(payload or {})
-            h = hashlib.sha256(f"{text}|{EMBED_MODEL}".encode()).hexdigest()
+            h = _payload_embed_hash(payload)
             if h not in live_hashes.get(did, set()):
                 todo.append((did, text, h))
         if not todo:
@@ -1820,6 +1979,64 @@ def embed_touched_cards(sess, work_id, touched):
         return 0
 
 
+def _capture_touched_fence(conn, touched):
+    """捕获窗提交后 touched 卡的 revision/hash，作为冲突补偿唯一许可。"""
+
+    fence = {}
+    for draft_id in sorted(touched):
+        payload, revision = _read_upgrade_draft_snapshot(conn, draft_id)
+        fence[draft_id] = (revision, _payload_embed_hash(payload))
+    return fence
+
+
+def _assert_touched_fence(conn, work_id, fence):
+    """锁定并复验提交后围栏；任何外部确认、改版、删除都禁止 undo。"""
+
+    for draft_id in sorted(fence):
+        row = conn.execute(
+            """SELECT work_id, draft_payload, status, revision, source_type, deleted
+               FROM muse_knowledge_draft
+               WHERE id=%s AND tenant_id=%s FOR UPDATE""",
+            (draft_id, TENANT),
+        ).fetchone()
+        expected_revision, expected_hash = fence[draft_id]
+        if not row:
+            raise CompensationFenceConflict(f"补偿围栏 draft 缺失：draft={draft_id}")
+        current_work, payload, status, revision, source_type, deleted = row
+        current_hash = _payload_embed_hash(payload)
+        if current_work != work_id or deleted or status != "pending" \
+                or source_type != SOURCE_TYPE or revision != expected_revision \
+                or current_hash != expected_hash:
+            raise CompensationFenceConflict(
+                f"补偿围栏冲突：draft={draft_id}，work={current_work}，status={status}，"
+                f"source_type={source_type}，deleted={deleted}，revision={revision}，"
+                f"expected_revision={expected_revision}，hash_match={current_hash == expected_hash}"
+            )
+
+
+def _persist_compensation_failed(work_id, win_no, owner_error, compensation_error):
+    """补偿事务失败后另开事务落永久断点，避免旧 done 被后续运行跳过。"""
+
+    message = (
+        f"{COMPENSATION_FAILED_PREFIX} owner={owner_error}; compensation={compensation_error}"
+    )[:500]
+    with psycopg.connect(DSN) as conn:
+        row = conn.execute(
+            """UPDATE example_upgrade_window SET status='failed', error_message=%s,
+                   updater='upgrade'
+               WHERE tenant_id=%s AND work_id=%s AND window_no=%s
+                 AND status='done' AND error_message IS NULL AND updater='upgrade'
+               RETURNING status""",
+            (message, TENANT, work_id, win_no),
+        ).fetchone()
+        if not row:
+            raise CompensationFenceConflict(
+                f"补偿 marker 窗口 CAS 冲突：work={work_id}，window={win_no}"
+            )
+        conn.commit()
+    return message
+
+
 # ── 命令 ──
 
 @click.group()
@@ -1847,6 +2064,7 @@ def _run(work_id, max_windows, max_calls, model, redo_window, semantic_on):
     embed_sess = _embed_session() if semantic_on else None
     redo_snapshot = None
     redo_material = None
+    redo_postcommit_fence = None
 
     def call(prompt, need_keys):
         calls["n"] += 1
@@ -1914,9 +2132,13 @@ def _run(work_id, max_windows, max_calls, model, redo_window, semantic_on):
             wins = [(r[0], r[1], r[2]) for r in wins]
             # failed 窗重跑前必须先撤销旧写入（部分写入直接重跑会 double-append 追加字段）
             for r0 in conn.execute(
-                    """SELECT window_no FROM example_upgrade_window
+                    """SELECT window_no, error_message FROM example_upgrade_window
                        WHERE tenant_id=%s AND work_id=%s AND status='failed' AND deleted=FALSE""",
                     (TENANT, work_id)).fetchall():
+                if str(r0[1] or "").startswith(COMPENSATION_FAILED_PREFIX):
+                    raise click.ClickException(
+                        f"failed 窗{r0[0]} 存在未完成补偿断点，禁止自动 undo：{r0[1]}"
+                    )
                 click.echo(f"[撤销] failed 窗{r0[0]} 旧写入回滚后重跑")
                 undo_window(conn, work_id, r0[0])
     done_n = 0
@@ -1940,7 +2162,7 @@ def _run(work_id, max_windows, max_calls, model, redo_window, semantic_on):
                     text, chapter_texts = redo_material
                 else:
                     text, chapter_texts = load_window_material(conn, work_id, a, b)
-                name_map, presence, aliases_by_draft = load_known(conn, work_id)
+                name_map, presence, aliases_by_draft, expected_revisions = load_known(conn, work_id)
             onstage = prescan(name_map, text)          # ⓪ 机械预扫纯内存，无连接
             # ── 算1（无连接）：① 实体观察 M3 调用 + 输出清洗，全程不持连接 ──
             obs, usage = call(observe_prompt(contracts, title, a, b, text, onstage),
@@ -1977,6 +2199,7 @@ def _run(work_id, max_windows, max_calls, model, redo_window, semantic_on):
             # ── 写（窗事务，保持现单事务原子性）：②-⑥ 照旧，唯一改动=判重换查预判 verdicts；update/relation
             #    的 M3 仍在窗事务内——既有已接受模式（keepalives 兜底），本次明确不动其连接边界 ──
             new_ids = []   # 本窗新立卡 id（洞①：与 to_update 键并成 touched 供嵌段增量嵌入）
+            touched_fence = {}
             with psycopg.connect(DSN) as conn:
                 # ② 判重+立卡门槛（机械；分类走 _classify_new_name 纯函数，与预判段同判据防两处漂移）
                 to_update = {}   # draft_id -> [观察点…]
@@ -1992,15 +2215,13 @@ def _run(work_id, max_windows, max_calls, model, redo_window, semantic_on):
                     if kind == "alias":                      # 疑似别名：初卡转观察材料（G3；洞②材料不截里程碑）
                         did = name_map[key][0]
                         canonical_name = _resolve_canonical_name(name_map, aliases_by_draft, key)
+                        # alias 是分类副作用，也必须先通过模型所见目标卡的 pending/source/revision 门禁。
+                        _lock_upgrade_draft(conn, did, expected_revisions[did])
                         to_update.setdefault(did, []).append(
                             f"（别名「{nm}」并入）初卡材料：{_merge_material(ent)}")
-                        conn.execute(
-                            """INSERT INTO example_upgrade_alias
-                                   (work_id, canonical_name, alias, evidence_window, verdict_by, tenant_id)
-                               VALUES (%s,%s,%s,%s,'ai',%s)
-                               ON CONFLICT (tenant_id, work_id, alias) DO NOTHING""",
-                            (work_id, canonical_name, nm, win_no, TENANT))
-                        legal_alias = _clean_alias(nm)
+                        legal_alias = _claim_alias(
+                            conn, work_id, canonical_name, nm, win_no, "ai"
+                        )
                         if legal_alias:
                             aliases_by_draft.setdefault(did, set()).add(legal_alias)
                         continue
@@ -2008,16 +2229,13 @@ def _run(work_id, max_windows, max_calls, model, redo_window, semantic_on):
                         # 不直接立卡，转观察材料并入既有卡由更新步 AI 甄别（软防护，留复核标记；洞②材料不截里程碑）
                         did = name_map[key][0]
                         canonical_name = _resolve_canonical_name(name_map, aliases_by_draft, key)
+                        _lock_upgrade_draft(conn, did, expected_revisions[did])
                         to_update.setdefault(did, []).append(
                             f"（名称疑似同一实体「{nm}」≈「{key}」，请甄别后再并入）"
                             f"初卡材料：{_merge_material(ent)}")
-                        conn.execute(
-                            """INSERT INTO example_upgrade_alias
-                                   (work_id, canonical_name, alias, evidence_window, verdict_by, tenant_id)
-                               VALUES (%s,%s,%s,%s,'substr',%s)
-                               ON CONFLICT (tenant_id, work_id, alias) DO NOTHING""",
-                            (work_id, canonical_name, nm, win_no, TENANT))
-                        legal_alias = _clean_alias(nm)
+                        legal_alias = _claim_alias(
+                            conn, work_id, canonical_name, nm, win_no, "substr"
+                        )
                         if legal_alias:
                             aliases_by_draft.setdefault(did, set()).add(legal_alias)
                         continue
@@ -2031,15 +2249,12 @@ def _run(work_id, max_windows, max_calls, model, redo_window, semantic_on):
                             verdict, vd = verdicts.get(nm, ("new", None))
                             if verdict == "merge":           # 同型同一实体：并入既有卡，不另立
                                 did0, canon = vd
+                                _lock_upgrade_draft(conn, did0, expected_revisions[did0])
                                 to_update.setdefault(did0, []).append(
                                     f"（语义判重·「{nm}」并入同一实体）初卡材料：{_merge_material(ent)}")
-                                conn.execute(
-                                    """INSERT INTO example_upgrade_alias
-                                           (work_id, canonical_name, alias, evidence_window, verdict_by, tenant_id)
-                                       VALUES (%s,%s,%s,%s,'semantic',%s)
-                                       ON CONFLICT (tenant_id, work_id, alias) DO NOTHING""",
-                                    (work_id, canon, nm, win_no, TENANT))
-                                legal_alias = _clean_alias(nm)
+                                legal_alias = _claim_alias(
+                                    conn, work_id, canon, nm, win_no, "semantic"
+                                )
                                 if legal_alias:
                                     aliases_by_draft.setdefault(did0, set()).add(legal_alias)
                                 continue
@@ -2054,6 +2269,7 @@ def _run(work_id, max_windows, max_calls, model, redo_window, semantic_on):
                                     known_chapters=hist,
                                 )
                                 new_ids.append(did)
+                                _, expected_revisions[did] = _read_upgrade_draft_snapshot(conn, did)
                                 target = (did, ent.get("型", ""), ent.get("一句话摘要", ""))
                                 name_map[canonical_name] = target
                                 for card_alias in card_aliases:
@@ -2079,6 +2295,7 @@ def _run(work_id, max_windows, max_calls, model, redo_window, semantic_on):
                             known_chapters=hist,
                         )
                         new_ids.append(did)
+                        _, expected_revisions[did] = _read_upgrade_draft_snapshot(conn, did)
                         target = (did, ent.get("型", ""), ent.get("一句话摘要", ""))
                         name_map[canonical_name] = target
                         for card_alias in card_aliases:
@@ -2105,9 +2322,11 @@ def _run(work_id, max_windows, max_calls, model, redo_window, semantic_on):
                 for i in range(0, len(items), UPDATE_BATCH):
                     batch = items[i:i + UPDATE_BATCH]
                     cards = []
+                    batch_revisions = {}
                     for did, obs_pts in batch:
-                        p = conn.execute("SELECT draft_payload FROM muse_knowledge_draft WHERE id=%s",
-                                         (did,)).fetchone()[0]
+                        p, prompt_revision = _read_upgrade_draft_snapshot(conn, did)
+                        batch_revisions[did] = prompt_revision
+                        expected_revisions[did] = prompt_revision
                         cards.append((did, p, obs_pts))
                     # 缺"更新"键宽容为空批：prompt 教"无变化的卡不输出"，某批恰好全无变化时
                     # 模型会顺势连键一起省（批7实测 6 窗全死于此）。语义上缺键≈空批，按空批放行
@@ -2139,30 +2358,33 @@ def _run(work_id, max_windows, max_calls, model, redo_window, semantic_on):
                                        valid_keys=did2keys.get(u["draft_id"]),
                                        chapter_texts=chapter_texts,
                                        appearance_chapters=pres_add.pop(u["draft_id"], set()),
-                                       known_aliases=aliases_by_draft.get(u["draft_id"], set()))
+                                       known_aliases=aliases_by_draft.get(u["draft_id"], set()),
+                                       expected_revision=batch_revisions[u["draft_id"]])
                 # ⑤ 关系增量（核心角色=本窗有更新的 character + 在场 character，≤8）
                 relation_touched = set()
                 char_cards = []
                 seen = set()
                 for did in list(to_update.keys()):
-                    p = conn.execute("SELECT draft_payload FROM muse_knowledge_draft WHERE id=%s",
-                                     (did,)).fetchone()[0]
-                    if p.get("type") == "character" and did not in seen:
+                    candidate, participant_revision = _read_upgrade_draft_snapshot(conn, did)
+                    if candidate.get("type") == "character" and did not in seen:
+                        p, _ = _lock_upgrade_draft(conn, did, participant_revision)
                         char_cards.append((did, p))
                         seen.add(did)
                 for nm, (did, t, _) in onstage.items():
                     if t == "character" and did and did not in seen and len(char_cards) < 8:
-                        p = conn.execute("SELECT draft_payload FROM muse_knowledge_draft WHERE id=%s",
-                                         (did,)).fetchone()[0]
+                        _, participant_revision = _read_upgrade_draft_snapshot(conn, did)
+                        p, _ = _lock_upgrade_draft(conn, did, participant_revision)
                         char_cards.append((did, p))
                         seen.add(did)
                 char_cards = char_cards[:8]
                 if len(char_cards) >= 2:
-                    rels = conn.execute(
-                        """SELECT id, draft_payload FROM muse_knowledge_draft
+                    rel_rows = conn.execute(
+                        """SELECT id, draft_payload, revision FROM muse_knowledge_draft
                            WHERE tenant_id=%s AND work_id=%s AND source_type=%s AND deleted=FALSE
-                             AND draft_payload->>'type'=%s""",
+                             AND status='pending' AND draft_payload->>'type'=%s""",
                         (TENANT, work_id, SOURCE_TYPE, RELATION_TYPE)).fetchall()
+                    rels = [(rid, payload) for rid, payload, _ in rel_rows]
+                    relation_revisions = {rid: revision for rid, _, revision in rel_rows}
                     rel_out, _ = call(relation_prompt(contracts, title, a, b, text,
                                                       char_cards, rels), ("关系",))
                     id2name = {d: p.get("名称") for d, p in char_cards}
@@ -2178,7 +2400,9 @@ def _run(work_id, max_windows, max_calls, model, redo_window, semantic_on):
                         if key in exist:
                             rid, rp = exist[key]
                             relation_touched.add(
-                                _update_relation_card(conn, work_id, win_no, rid, rp, r)
+                                _update_relation_card(
+                                    conn, work_id, win_no, rid, relation_revisions[rid], r
+                                )
                             )
                         else:
                             f2 = dict(r.get("其他字段") or {})
@@ -2202,32 +2426,39 @@ def _run(work_id, max_windows, max_calls, model, redo_window, semantic_on):
                         pres_add.setdefault(name_map[nm][0], set()).update(
                             _int_chaps(it.get("出场章"))
                         )
+                presence_touched = set()
                 for did, chs in pres_add.items():
                     if not chs:
                         continue
-                    p = conn.execute("SELECT draft_payload FROM muse_knowledge_draft WHERE id=%s",
-                                     (did,)).fetchone()[0]
+                    payload, revision = _lock_upgrade_draft(
+                        conn, did, expected_revisions[did]
+                    )
                     # 纯出场与无字段变化卡复用同一「过滤+审计追加」路径，确保 redo 能撤销上一轮顶层章号。
                     changed = _append_verified_appearance_chapters(
                         conn,
                         did,
                         win_no,
-                        p,
+                        payload,
                         chs,
                         chapter_texts,
                         known_aliases=aliases_by_draft.get(did, set()),
                     )
                     if not changed:
                         continue
-                    conn.execute("UPDATE muse_knowledge_draft SET draft_payload=%s WHERE id=%s",
-                                 (json.dumps(p, ensure_ascii=False), did))
+                    _write_locked_upgrade_draft(conn, did, payload, revision)
+                    presence_touched.add(did)
                 conn.execute(
                     """UPDATE example_upgrade_window SET status='done', error_message=NULL,
                            updater='upgrade' WHERE tenant_id=%s AND work_id=%s AND window_no=%s""",
                     (TENANT, work_id, win_no))
+                touched = set(new_ids) | set(to_update.keys()) | relation_touched | presence_touched
+                if semantic_on and touched:
+                    # 在提交事务内捕获即将提交的精确版本，消除 commit 后被外部改版再采样的竞态。
+                    touched_fence = _capture_touched_fence(conn, touched)
+                if redo_snapshot is not None:
+                    # restore 会覆盖整书六域；在提交前捕获其完整提交态，供冲突补偿 exact 复验。
+                    redo_postcommit_fence = _snapshot_redo_window(conn, work_id, win_no)
                 conn.commit()
-            # 写段成功：实体与关系的新建/更新 id 一并进入提交后嵌段，确保当前 payload 只有一个活向量。
-            touched = set(new_ids) | set(to_update.keys()) | relation_touched
         except SensitiveHardStop as e:
             with psycopg.connect(DSN) as conn:
                 if redo_snapshot is not None:
@@ -2285,13 +2516,45 @@ def _run(work_id, max_windows, max_calls, model, redo_window, semantic_on):
                 embed_n = embed_touched_cards(embed_sess, work_id, touched)
             except EmbeddingOwnershipConflict as exc:
                 error_message = f"嵌入唯一 owner 冲突：{exc}"[:500]
-                with psycopg.connect(DSN) as conn:
-                    conn.execute(
-                        """UPDATE example_upgrade_window SET status='failed', error_message=%s,
-                               updater='upgrade' WHERE tenant_id=%s AND work_id=%s AND window_no=%s""",
-                        (error_message, TENANT, work_id, win_no),
+                # 先把 done 改成 durable 失败断点；补偿成功后普通窗会覆盖成最终 failed，
+                # redo 会恢复原窗状态。若补偿异常，此 marker 天然保留，不依赖第二次数据库写入。
+                try:
+                    durable_message = _persist_compensation_failed(
+                        work_id, win_no, exc, "pending"
                     )
-                    conn.commit()
+                except Exception as marker_exc:
+                    raise click.ClickException(
+                        f"嵌入冲突后无法持久化补偿断点：{marker_exc}"
+                    ) from marker_exc
+                try:
+                    with psycopg.connect(DSN) as conn:
+                        if redo_snapshot is not None:
+                            # redo restore 覆盖整书六域，必须阻写并 exact 复验完整提交态。
+                            expected_redo_fence = deepcopy(redo_postcommit_fence)
+                            expected_redo_fence["window"] = (
+                                "failed", durable_message, "upgrade"
+                            )
+                            _assert_redo_postcommit_fence(
+                                conn, work_id, expected_redo_fence
+                            )
+                            _restore_redo_window(conn, work_id, redo_snapshot)
+                        else:
+                            # 普通前滚只会撤销本窗写入，维持 touched draft 围栏边界。
+                            _assert_touched_fence(conn, work_id, touched_fence)
+                            # 普通窗必须在同一事务先撤销五域，再标 failed，禁止留下已提交半窗。
+                            undo_window(conn, work_id, win_no)
+                            conn.execute(
+                                """UPDATE example_upgrade_window SET status='failed', error_message=%s,
+                                       updater='upgrade'
+                                   WHERE tenant_id=%s AND work_id=%s AND window_no=%s""",
+                                (error_message, TENANT, work_id, win_no),
+                            )
+                        conn.commit()
+                except Exception as compensation_exc:
+                    # 补偿事务回滚，先前独立提交的 marker 保留；不依赖数据库仍可写的二次更新。
+                    raise click.ClickException(
+                        f"{durable_message}; compensation={compensation_exc}"[:500]
+                    ) from compensation_exc
                 raise click.ClickException(error_message) from exc
         new_n = len(obs.get('新名字', []))
         extra = f" 判重:并{dedup_merge_n}/链{dedup_chain_n} 嵌入:{embed_n}" if semantic_on else ""

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

@@ -37,8 +37,15 @@ from upgrade_work_lock import UpgradeWorkLockUnavailable, upgrade_work_lock  # n
 
 SOURCE_TYPE = "upgrade_book"
 
-# execute 在读取备份快照前锁住七域表，阻断未接统一 advisory lock 的 embed/confirm 并发写入。
+# capture_input_snapshot 实际读取 chapter/block 与 schema/schema_version；它们必须和七域一起先锁定，
+# 否则完整输入核对后仍可能被正文或 active 合同写入穿透。固定顺序采用“输入源→派生域”：潜在写者
+# 可能先持输入源表再写派生域，若 reset 反向先持派生表再等待输入源会形成互等。派生域内部沿用既有
+# 顺序，尤其保持 draft→embedding，与 embed/parse 写事务的固定表锁顺序一致。
 RESET_TABLE_LOCK_SQL = """LOCK TABLE
+    muse_content_chapter,
+    muse_content_block,
+    muse_meta_schema,
+    muse_meta_schema_version,
     muse_knowledge_draft,
     example_upgrade_window,
     example_upgrade_alias,
@@ -103,16 +110,28 @@ def _assert_code_identity_matches_manifest(manifest, current_identity):
         )
 
 
-def _reset(work_id, execute, manifest=None):
+def _reset(work_id, execute, manifest=None, code_identity=None):
     """在调用方已持有同书锁时预览或执行重抽清理。"""
 
     with psycopg.connect(DSN) as conn:
         if execute:
-            # 表锁、当前七域快照和后续 reset 全部位于同一事务，摘要核对后不存在写竞态窗口。
+            # 输入源与七域表锁、七域核对及完整输入绑定均在同一事务内，先于 destructive SQL。
             conn.execute(RESET_TABLE_LOCK_SQL)
             with conn.cursor(row_factory=dict_row) as cursor:
                 title, current_domains = backup._read_snapshot(cursor, work_id, TENANT)
-            _assert_snapshot_matches_manifest(manifest, current_domains)
+                _assert_snapshot_matches_manifest(manifest, current_domains)
+                current_input = backup.capture_input_snapshot(
+                    cursor,
+                    work_id,
+                    TENANT,
+                    code_identity,
+                    manifest["input"]["expectedChapters"],
+                    manifest["input"]["expectedWindows"],
+                    windows=current_domains["windows"],
+                )
+            backup.assert_restore_input(
+                manifest, current_input, work_id=work_id, tenant=TENANT
+            )
         else:
             title = conn.execute("SELECT title FROM muse_content_work WHERE id=%s",
                                  (work_id,)).fetchone()[0]
@@ -265,6 +284,7 @@ def main(work_id, execute, backup_dir, backup_id, confirmation_sha):
     try:
         with upgrade_work_lock(DSN, TENANT, work_id):
             manifest = None
+            code_identity = None
             if execute:
                 manifest = backup.verify_backup(backup_dir)
                 backup.validate_execute_confirmation(
@@ -277,10 +297,9 @@ def main(work_id, execute, backup_dir, backup_id, confirmation_sha):
                 if manifest.get("tenant") != TENANT or manifest.get("work") != work_id:
                     raise click.ClickException("备份 manifest 的 tenant/work 与 reset 目标不一致")
                 # 仍在同书 advisory lock 内，且尚未创建业务连接或执行任何 reset SQL。
-                _assert_code_identity_matches_manifest(
-                    manifest, backup.capture_code_identity()
-                )
-            return _reset(work_id, execute, manifest)
+                code_identity = backup.capture_code_identity()
+                _assert_code_identity_matches_manifest(manifest, code_identity)
+            return _reset(work_id, execute, manifest, code_identity)
     except UpgradeWorkLockUnavailable as exc:
         raise click.ClickException(str(exc)) from exc
     except click.ClickException:

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

@@ -27,6 +27,28 @@ BACKUP_ARGS = [
     "--backup-id", BACKUP_ID,
     "--confirmation-sha", CONFIRMATION_SHA,
 ]
+EXPECTED_RESET_LOCK_TABLES = (
+    "muse_content_chapter",
+    "muse_content_block",
+    "muse_meta_schema",
+    "muse_meta_schema_version",
+    "muse_knowledge_draft",
+    "example_upgrade_window",
+    "example_upgrade_alias",
+    "example_upgrade_presence",
+    "example_upgrade_card_state",
+    "example_upgrade_audit",
+    "example_knowledge_embedding",
+)
+
+
+def _locked_tables(sql):
+    """从规范化 LOCK TABLE SQL 中提取有序表名。"""
+
+    prefix = "lock table "
+    suffix = " in share row exclusive mode"
+    assert sql.startswith(prefix) and sql.endswith(suffix)
+    return tuple(table.strip() for table in sql[len(prefix):-len(suffix)].split(","))
 
 
 def _lock_failure(*_args, **_kwargs):
@@ -50,10 +72,11 @@ class _Result:
 
 
 class _CursorContext:
-    """记录 dict_row 游标请求；快照读取本身由备份模块替身返回。"""
+    """模拟 dict_row cursor；同一 SQL 在默认 connection 上仍返回 tuple。"""
 
-    def __init__(self, conn):
+    def __init__(self, conn, row_factory):
         self.conn = conn
+        self.row_factory = row_factory
 
     def __enter__(self):
         return self
@@ -61,12 +84,19 @@ class _CursorContext:
     def __exit__(self, exc_type, exc, traceback):
         return False
 
+    def execute(self, sql, params=None):
+        normalized = " ".join(sql.split()).lower()
+        self.conn.executions.append((normalized, params))
+        return self.conn._input_query_result(normalized, mapping=True)
+
 
 class _ResetConnection:
     """有状态 fake DB：验证向量 SQL 边界，并保留不应被 reset 误伤的反例。"""
 
-    def __init__(self, *, postcheck_failure=None, confirmed_draft_id=None):
+    def __init__(self, *, postcheck_failure=None, confirmed_draft_id=None,
+                 input_drift=None):
         self.postcheck_failure = postcheck_failure
+        self.input_drift = input_drift
         self.commits = 0
         self.executions = []
         self.cursor_row_factories = []
@@ -120,7 +150,60 @@ class _ResetConnection:
         """只允许实现代码显式请求 dict_row 快照游标。"""
 
         self.cursor_row_factories.append(row_factory)
-        return _CursorContext(self)
+        return _CursorContext(self, row_factory)
+
+    def _input_query_result(self, normalized, *, mapping):
+        """为完整输入查询提供真实列形；默认 connection 故意返回 tuple。"""
+
+        if normalized.startswith("select current_database() as database"):
+            row = {
+                "database": "other-db" if self.input_drift == "database" else "muse-example",
+                "server_addr": "100.64.0.8",
+                "server_port": 5433,
+                "server_version": "160000",
+            }
+            return _Result(row=row if mapping else tuple(row.values()))
+        if normalized.startswith("select c.id as chapter_id"):
+            rows = [
+                {
+                    "chapter_id": 11,
+                    "chapter_order": 1,
+                    "chapter_title": "第一章",
+                    "chapter_revision": 1,
+                    "block_id": 101,
+                    "block_order": 1,
+                    "block_type": "scene",
+                    "block_title": None,
+                    "content_doc": None,
+                    "content_text": "漂移正文" if self.input_drift == "canonical" else "第一章正文",
+                    "block_revision": 1,
+                },
+                {
+                    "chapter_id": 12,
+                    "chapter_order": 2,
+                    "chapter_title": "第二章",
+                    "chapter_revision": 1,
+                    "block_id": 102,
+                    "block_order": 1,
+                    "block_type": "scene",
+                    "block_title": None,
+                    "content_doc": None,
+                    "content_text": "第二章正文",
+                    "block_revision": 1,
+                },
+            ]
+            return _Result(rows=rows if mapping else [tuple(row.values()) for row in rows])
+        if normalized.startswith("select s.schema_key, s.active_version_id"):
+            rows = [
+                {
+                    "schema_key": schema_key,
+                    "active_version_id": index,
+                    "field_contract_snapshot": {"字段": []},
+                }
+                for index, schema_key in enumerate(sorted(backup.UPGRADE_SCHEMA_KEYS), start=1)
+            ]
+            return _Result(rows=rows if mapping else [tuple(row.values()) for row in rows])
+        raise AssertionError(f"未覆盖的完整输入 SQL：{normalized}")
 
     @staticmethod
     def _is_target_draft(draft):
@@ -156,14 +239,14 @@ class _ResetConnection:
         normalized = " ".join(sql.split()).lower()
         self.executions.append((normalized, params))
         if normalized.startswith("lock table"):
-            assert "share row exclusive mode" in normalized
-            for table in (
-                    "muse_knowledge_draft", "example_upgrade_window",
-                    "example_upgrade_alias", "example_upgrade_presence",
-                    "example_upgrade_card_state", "example_upgrade_audit",
-                    "example_knowledge_embedding"):
-                assert table in normalized
+            assert _locked_tables(normalized) == EXPECTED_RESET_LOCK_TABLES
             return _Result()
+        if normalized.startswith((
+                "select current_database() as database",
+                "select c.id as chapter_id",
+                "select s.schema_key, s.active_version_id",
+        )):
+            return self._input_query_result(normalized, mapping=False)
         if normalized.startswith("select title from muse_content_work"):
             return _Result(("离线测试书",))
         if normalized.startswith("select count(*) from muse_knowledge_draft"):
@@ -277,6 +360,40 @@ def _code_files():
     }
 
 
+def _input_snapshot(*, code_files=None):
+    """构造与真实备份 manifest 同形的完整 reset 输入身份。"""
+
+    return {
+        "databaseIdentity": {
+            "database": "muse-example",
+            "serverAddr": "100.64.0.8",
+            "serverPort": 5433,
+            "serverVersion": "160000",
+        },
+        "canonicalContent": {
+            "chapterCount": 2,
+            "blockCount": 2,
+            "sha": "c" * 64,
+        },
+        "windowBoundary": {"count": 1, "sha": "w" * 64},
+        "activeFieldContracts": {
+            "count": 7,
+            "schemaKeys": sorted(backup.UPGRADE_SCHEMA_KEYS),
+            "sha": "f" * 64,
+        },
+        "codeFiles": dict(code_files or _code_files()),
+        "expectedChapters": 2,
+        "expectedWindows": 1,
+        "plan": {
+            "model": backup.PLANNED_MODEL,
+            "semanticDedup": backup.PLANNED_SEMANTIC_DEDUP,
+        },
+        "sourceType": backup.SOURCE_TYPE,
+        "tenant": 1,
+        "work": 8,
+    }
+
+
 def _manifest_for(domains, *, code_files=None):
     """按备份模块的排序、主键摘要和内容摘要构造已离线 verify 的清单替身。"""
 
@@ -290,12 +407,16 @@ def _manifest_for(domains, *, code_files=None):
             "primaryKeySha": backup._primary_key_sha(spec, rows),
             "contentSha": backup._content_sha(rows),
         }
+    input_snapshot = _input_snapshot(code_files=code_files)
     return {
         "backup_id": BACKUP_ID,
         "confirmationSha": CONFIRMATION_SHA,
         "work": 8,
         "tenant": 1,
-        "input": {"codeFiles": dict(code_files or _code_files())},
+        "input": input_snapshot,
+        "inputSha": backup._sha256(
+            backup.canonical_json(input_snapshot).encode("utf-8")
+        ),
         "domains": domain_manifest,
     }, normalized
 
@@ -382,12 +503,15 @@ class UpgradeCommandLockOfflineTest(unittest.TestCase):
 
     def _invoke_reset_execute(self, conn, *, manifest=None, current_domains=None,
                               backup_id=BACKUP_ID, confirmation_sha=CONFIRMATION_SHA,
-                              lock_context=_lock_success, current_code_files=None):
+                              lock_context=_lock_success, current_code_files=None,
+                              current_input=None, capture_input=None):
         """在锁、磁盘 verify 与同事务快照均为离线替身时调用 execute。"""
 
         default_manifest, default_domains = _manifest_for(_snapshot_domains())
         manifest = manifest or default_manifest
         current_domains = current_domains or default_domains
+        current_input = manifest["input"] if current_input is None else current_input
+        capture_input = capture_input or Mock(return_value=current_input)
         args = [
             "--work-id", "8", "--execute",
             "--backup-dir", "/private/tmp",
@@ -400,6 +524,7 @@ class UpgradeCommandLockOfflineTest(unittest.TestCase):
                 patch.object(backup, "capture_code_identity", return_value={
                     "codeFiles": dict(current_code_files or manifest["input"]["codeFiles"]),
                 }), \
+                patch.object(backup, "capture_input_snapshot", capture_input), \
                 patch.object(backup, "_read_snapshot",
                              return_value=("离线测试书", current_domains)):
             return self.runner.invoke(reset.main, args)
@@ -435,10 +560,12 @@ class UpgradeCommandLockOfflineTest(unittest.TestCase):
             result = self.runner.invoke(reset.main, ["--work-id", "8"])
 
         self.assertEqual(result.exit_code, 0, result.output)
+        self.assertIn("《离线测试书》", result.output)
         self.assertIn("软删活向量 2", result.output)
         self.assertFalse(any(sql.startswith(("update ", "delete "))
                              for sql, _params in conn.executions))
         self.assertEqual(conn.commits, 0)
+        self.assertEqual(conn.cursor_row_factories, [])
 
     def test_reset_execute_soft_deletes_only_target_upgrade_vectors(self):
         """执行只软删当前租户、作品、升格来源向量，并用明确 actor 留痕。"""
@@ -464,8 +591,8 @@ class UpgradeCommandLockOfflineTest(unittest.TestCase):
         self.assertTrue(vector_writes[0].startswith("update "))
         self.assertIn("e.entity_id is null", vector_writes[0])
 
-    def test_reset_execute_verifies_backup_inside_advisory_lock(self):
-        """磁盘 verify 必须发生在同书 advisory lock 内，且早于业务连接。"""
+    def test_reset_execute_verifies_backup_and_input_inside_advisory_lock(self):
+        """磁盘 verify 与完整输入读取均须位于同书 advisory lock 内。"""
 
         events = []
 
@@ -485,12 +612,19 @@ class UpgradeCommandLockOfflineTest(unittest.TestCase):
             events.append("backup-verified")
             return manifest
 
+        def captured_input(*_args, **_kwargs):
+            self.assertEqual(events, ["lock-enter", "backup-verified"])
+            events.append("input-captured")
+            return manifest["input"]
+
         with patch.object(reset, "upgrade_work_lock", tracked_lock), \
                 patch.object(reset.psycopg, "connect", return_value=conn), \
                 patch.object(backup, "verify_backup", side_effect=verified_manifest), \
                 patch.object(backup, "capture_code_identity", return_value={
                     "codeFiles": _code_files(),
                 }), \
+                patch.object(backup, "capture_input_snapshot",
+                             side_effect=captured_input), \
                 patch.object(backup, "_read_snapshot",
                              return_value=("离线测试书", domains)):
             result = self.runner.invoke(
@@ -498,7 +632,9 @@ class UpgradeCommandLockOfflineTest(unittest.TestCase):
             )
 
         self.assertEqual(result.exit_code, 0, result.output)
-        self.assertEqual(events, ["lock-enter", "backup-verified", "lock-exit"])
+        self.assertEqual(
+            events, ["lock-enter", "backup-verified", "input-captured", "lock-exit"]
+        )
 
     def test_reset_execute_rejects_snapshot_drift_before_writes(self):
         """七域任一摘要漂移必须在所有 reset 写语句前失败。"""
@@ -517,6 +653,119 @@ class UpgradeCommandLockOfflineTest(unittest.TestCase):
         self.assertEqual(len(conn.executions), 1)
         self.assertTrue(conn.executions[0][0].startswith("lock table"))
 
+    def test_reset_execute_captures_complete_input_inside_table_lock(self):
+        """输入源与七域须按固定顺序先锁定，再读完整输入和执行 destructive SQL。"""
+
+        manifest, domains = _manifest_for(_snapshot_domains())
+        current_identity = {"codeFiles": dict(manifest["input"]["codeFiles"])}
+        conn = _ResetConnection()
+
+        def capture_input(*args, **kwargs):
+            self.assertTrue(conn.executions)
+            self.assertTrue(conn.executions[0][0].startswith("lock table"))
+            self.assertEqual(
+                _locked_tables(conn.executions[0][0]), EXPECTED_RESET_LOCK_TABLES
+            )
+            self.assertFalse(any(sql.startswith(("update ", "delete "))
+                                 for sql, _params in conn.executions))
+            capture_cursor = args[0]
+            self.assertIsInstance(capture_cursor, _CursorContext)
+            self.assertIs(capture_cursor.conn, conn)
+            self.assertIs(capture_cursor.row_factory, reset.dict_row)
+            self.assertEqual(args[1:], (8, 1, current_identity, 2, 1))
+            self.assertEqual(kwargs, {"windows": domains["windows"]})
+            return manifest["input"]
+
+        capture = Mock(side_effect=capture_input)
+        result = self._invoke_reset_execute(
+            conn,
+            manifest=manifest,
+            current_domains=domains,
+            current_code_files=current_identity["codeFiles"],
+            capture_input=capture,
+        )
+
+        self.assertEqual(result.exit_code, 0, result.output)
+        capture.assert_called_once()
+        self.assertEqual(conn.commits, 1)
+
+    def test_reset_execute_real_input_capture_uses_dict_cursor_and_guards_drift(self):
+        """默认 connection 为 tuple 时，真实 input capture 仍须成功并保持写前漂移闸门。"""
+
+        domains = backup.sort_rows("windows", _snapshot_domains()["windows"])
+        all_domains = _snapshot_domains()
+        all_domains["windows"] = domains
+        manifest, normalized_domains = _manifest_for(all_domains)
+        identity = {"codeFiles": dict(manifest["input"]["codeFiles"])}
+        baseline_conn = _ResetConnection()
+        with baseline_conn.cursor(row_factory=reset.dict_row) as cursor:
+            expected_input = backup.capture_input_snapshot(
+                cursor, 8, 1, identity, 2, 1, windows=domains
+            )
+        manifest["input"] = expected_input
+        manifest["inputSha"] = backup._sha256(
+            backup.canonical_json(expected_input).encode("utf-8")
+        )
+
+        for case, input_drift, expected_exit in (
+                ("consistent", None, 0),
+                ("canonical-drift", "canonical", 1)):
+            with self.subTest(case=case):
+                conn = _ResetConnection(input_drift=input_drift)
+                capture = Mock(wraps=backup.capture_input_snapshot)
+                result = self._invoke_reset_execute(
+                    conn,
+                    manifest=manifest,
+                    current_domains=normalized_domains,
+                    current_code_files=identity["codeFiles"],
+                    capture_input=capture,
+                )
+
+                self.assertEqual(result.exit_code, expected_exit, result.output)
+                capture.assert_called_once()
+                capture_cursor = capture.call_args.args[0]
+                self.assertIsInstance(capture_cursor, _CursorContext)
+                self.assertIs(capture_cursor.row_factory, reset.dict_row)
+                writes = [sql for sql, _params in conn.executions
+                          if sql.startswith(("update ", "delete "))]
+                if input_drift:
+                    self.assertIn("input 漂移", result.output)
+                    self.assertEqual(writes, [])
+                    self.assertEqual(conn.commits, 0)
+                else:
+                    self.assertTrue(writes)
+                    self.assertEqual(conn.commits, 1)
+
+    def test_reset_execute_rejects_complete_input_drift_before_writes(self):
+        """正文、字段合同或数据库身份漂移均必须在 destructive SQL 前失败。"""
+
+        cases = (
+            ("canonical", ("canonicalContent", "sha"), "d" * 64, "input 漂移"),
+            ("contracts", ("activeFieldContracts", "sha"), "e" * 64, "input 漂移"),
+            ("database", ("databaseIdentity", "database"), "other-db", "数据库 identity"),
+        )
+        for case, path, value, expected_message in cases:
+            with self.subTest(case=case):
+                manifest, domains = _manifest_for(_snapshot_domains())
+                current_input = copy.deepcopy(manifest["input"])
+                current_input[path[0]][path[1]] = value
+                conn = _ResetConnection()
+
+                result = self._invoke_reset_execute(
+                    conn,
+                    manifest=manifest,
+                    current_domains=domains,
+                    current_input=current_input,
+                )
+
+                self.assertNotEqual(result.exit_code, 0)
+                self.assertIn(expected_message, result.output)
+                self.assertEqual(conn.commits, 0)
+                self.assertEqual(len(conn.executions), 1)
+                self.assertTrue(conn.executions[0][0].startswith("lock table"))
+                self.assertFalse(any(sql.startswith(("update ", "delete "))
+                                     for sql, _params in conn.executions))
+
     def test_reset_execute_rejects_missing_or_drifted_code_identity_before_database(self):
         """七域不变时，旧 manifest 缺 reset 或 reset/parse 任一 fileSha 漂移仍须拒绝。"""