From daac544211738f639cbb02e34dd4a15a67e59576 Mon Sep 17 00:00:00 2001 From: zizi Date: Tue, 21 Jul 2026 17:02:11 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=9E=E7=8E=B0:=20=E4=B8=BA=E5=8D=87?= =?UTF-8?q?=E6=A0=BC=E5=85=A8=E4=B9=A6=E9=87=8D=E5=BB=BA=E5=BB=BA=E7=AB=8B?= =?UTF-8?q?=E5=8F=AF=E6=81=A2=E5=A4=8D=E5=A4=87=E4=BB=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 在统一作品锁下导出并校验七域状态，绑定正文、窗口、字段合同与代码身份；提供事务回滚演练和双确认恢复入口，防止 reset 前后出现并发混写或不可恢复状态。 --- .claude/skills/parse-book/SKILL.md | 19 + .../parse-book/scripts/backup_upgrade_work.py | 1333 +++++++++++++++++ .../parse-book/scripts/parse_upgrade.py | 47 +- .../parse-book/scripts/reset_upgrade_work.py | 21 +- .../test_backup_upgrade_work_offline.py | 650 ++++++++ .../scripts/test_parse_upgrade_offline.py | 8 +- .../test_reset_upgrade_work_offline.py | 102 ++ .../scripts/test_upgrade_work_lock_offline.py | 133 ++ .../parse-book/scripts/upgrade_work_lock.py | 65 + 9 files changed, 2355 insertions(+), 23 deletions(-) create mode 100644 .claude/skills/parse-book/scripts/backup_upgrade_work.py create mode 100644 .claude/skills/parse-book/scripts/test_backup_upgrade_work_offline.py create mode 100644 .claude/skills/parse-book/scripts/test_reset_upgrade_work_offline.py create mode 100644 .claude/skills/parse-book/scripts/test_upgrade_work_lock_offline.py create mode 100644 .claude/skills/parse-book/scripts/upgrade_work_lock.py diff --git a/.claude/skills/parse-book/SKILL.md b/.claude/skills/parse-book/SKILL.md index ca14f23..6f26385 100644 --- a/.claude/skills/parse-book/SKILL.md +++ b/.claude/skills/parse-book/SKILL.md @@ -60,6 +60,25 @@ disable-model-invocation: true **作品面升格执行器 `scripts/parse_upgrade.py`（与上面范式拆书管线并行的另一条线,命令 `windows`/`run`/`status`）**：把参考书正文按窗抽成「会随剧情长大的实体卡」(升格卡,`source_type=upgrade_book`),设计见 `docs/2026-07-16-升格卡改造设计.md`。里程碑除真实章号外，模型输出必须临时携带所标章节正文短引；系统机械核验后删除证据，错章、缺证据或改写证据均拒收入库并留审计。模型给出的顶层出场章也必须由实体规范名或合法别名在对应章节正文中的实际出现机械证明；合法别名集合同时取卡内 payload 与独立 alias 表，单字规范名禁作章证据，规范名和别名均禁止通用称谓/关系称呼（如队长、舰长）作证据，常规二至四字专名保持精确子串命中。未命中章不得参与立卡、登场兜底或既有卡追加，无实证章时不补登场里程碑；既有卡每窗新增出场章必须审计完整旧值，redo/undo 精确恢复后再按新正文重算。显式 redo **仅允许当前 active 末窗**：执行前机械校验窗口号连续、章域合法且首尾相接，并校验目标窗逐章正文齐全且非空；任一失败必须在快照、undo 与任何写入前非零退出。历史窗修正必须全书前滚重建；后缀级联重算属于 P1，当前不支持。合法末窗 redo 仍须在撤销事务中清除所有 active `upgrade_book` 卡的本窗章域，保留窗外章并为历史无审计数据补旧值审计；redo 前建立完整恢复点（卡 payload/revision/deleted、别名、presence、水位、审计与窗状态），首次失败重新清理再试，最终失败完整恢复且重试初建卡软删。`--max-calls` 不得截断已经开始的即时重试，非显式 redo 不做该全局清理。`run` 默认**不发嵌入**;加 `--semantic-dedup` 开语义判重(治改名/跨型漏并)时,每窗按**读→算→写三段式短连接**跑——观察/近邻召回/M3 终判都在**无长连接**段发 LLM 与嵌入 HTTP(不再持窗级连接跨调用存活),写段只查预判结果落库;并**边抽边嵌**：本窗新建/更新卡在窗事务提交后增量嵌入落库(软删旧向量+upsert 新行),**后窗即可语义召回前窗刚长成的卡**,不再依赖"同书须预先全量 embed"。嵌段失败只告警、不牵连已 done 的窗。 +**同书命令互斥**：`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`、`parse_upgrade.py`、`llm.py` 和本 SKILL 的 fileSha。`gitCommit` 只记录 HEAD；当代码未提交时不得把它当执行代码身份，确认以 manifest 的 `inputSha + codeFiles + confirmationSha` 为准。SHA 仅用于发现传输、落盘或误操作造成的意外损坏，不宣称抵抗能同时改写工件和摘要的恶意篡改。命令默认不写数据库、不调用模型，生成后会关闭数据库连接并从磁盘独立复验；任一失败删除本次新建目录并非零退出。 + +```bash +# 创建 work 8 七域一致性备份；expected 值进入 input 摘要并机械核对 594 章/116 窗 +.venv/bin/python .claude/skills/parse-book/scripts/backup_upgrade_work.py backup --work-id 8 --expected-chapters 594 --expected-windows 116 +# 显式指定 /private/tmp 下的新目录 +.venv/bin/python .claude/skills/parse-book/scripts/backup_upgrade_work.py backup --work-id 8 --expected-chapters 594 --expected-windows 116 --output-dir /private/tmp/muse-upgrade-work8-before-reset +# 已有备份纯离线复验（不连接数据库、不调用模型） +.venv/bin/python .claude/skills/parse-book/scripts/backup_upgrade_work.py verify --backup-dir /private/tmp/muse-upgrade-work8-before-reset +# 在真实 public 表的单事务中应用七域恢复、逐域读回，最后强制 ROLLBACK；不调整 sequence +.venv/bin/python .claude/skills/parse-book/scripts/backup_upgrade_work.py rehearse --work-id 8 --backup-dir /private/tmp/muse-upgrade-work8-before-reset --confirmation-sha +# 演练通过并取得本次明确确认后才允许提交；backup-id 必须与 manifest 精确一致 +.venv/bin/python .claude/skills/parse-book/scripts/backup_upgrade_work.py restore --work-id 8 --backup-dir /private/tmp/muse-upgrade-work8-before-reset --execute --backup-id --confirmation-sha +``` + +backup 在同一只读快照内机械验证 active `window_no=1..expected-windows`、各窗章域无缝覆盖 Canonical 连续章域，并要求每章至少一个非空 block；任一不符不产备份。恢复前固定先离线 `verify`，核对 `confirmationSha`，并在事务内按 manifest 的 expected 值重算 input；work、tenant、数据库 identity 或 Canonical/窗/合同/代码任一漂移均拒绝。恢复内核按备份 ID exact upsert drafts/windows/embeddings；额外 active 窗直接拒绝，reset 后新增升格 draft 及其额外 embedding 只软删；aliases/presence/card_state/audits 按 work 边界先清后插。为避免 `updated_at` trigger 改写备份时间戳，恢复写段使用事务局部 replica trigger 模式，仍按 drafts→windows→aliases/presence→card_state/audits→embeddings 的父子顺序执行，事务结束自动复原。`rehearse` 永不调用非事务性的 sequence 调整并强制回滚；`restore --execute` 还必须同时精确匹配 `backup_id + confirmationSha`，逐域读回通过后才把 sequence 仅向前推进并提交。 + **范式线失败关闭（2026-07-21 假绿复盘后现行）**：`parse_outline.py window`、`parse_llm.py cards`、`parse_outline.py check` 是同一本作品的顺序流水线。普通网络/429/5xx 由 `chat_governed` 在单模型内短重试后沿 `MiniMax-M3 → MiniMax-M2.7 → glm-5.2 → deepseek-v4-flash` 降级；整条链耗尽，或功能层结构输出有限重试仍失败，必须**立即停止当前作品并保留断点**，禁止跳窗/跳型继续。`cards` 和 `check` 开始前机械验证大纲窗对全书章域无头部、中部、尾部缝隙；任一缝隙非零退出。补插窗后所有终检按 `from_order/to_order` 绝对章号排序并重新编号阶段，`window_no` 只作运行展示号，不得参与时序判断。 ## 步骤(自底向上,与创作期规划的自顶向下互为镜像) diff --git a/.claude/skills/parse-book/scripts/backup_upgrade_work.py b/.claude/skills/parse-book/scripts/backup_upgrade_work.py new file mode 100644 index 0000000..c7794bc --- /dev/null +++ b/.claude/skills/parse-book/scripts/backup_upgrade_work.py @@ -0,0 +1,1333 @@ +#!/usr/bin/env python3 +"""reset 前一致性备份任意作品的升格派生状态，并支持纯离线复验。 + +备份命令只在一个 ``REPEATABLE READ READ ONLY`` 事务中读取数据库，不写数据库， +也不调用任何模型。七个状态域先按主键稳定排序，再写入独立 JSONL；每个文件和清单均先写 +同目录临时文件，再用原子 rename 发布。数据库连接关闭后，脚本会从磁盘重新读取全部工件， +独立重算行数、主键摘要、内容摘要与文件摘要；失败时删除本次新建目录并非零退出。 + +``verify`` 子命令只读本地文件，不加载数据库连接配置，适合在离线环境复验已有备份。 +""" + +import base64 +import datetime as dt +import hashlib +import json +import math +import os +import pathlib +import re +import shutil +import subprocess +import uuid +from dataclasses import dataclass +from decimal import Decimal +from typing import Any, Callable, Mapping, Sequence + +import click + +from upgrade_work_lock import UpgradeWorkLockUnavailable, upgrade_work_lock + + +SOURCE_TYPE = "upgrade_book" +PLANNED_MODEL = "MiniMax-M3" +PLANNED_SEMANTIC_DEDUP = True +BACKUP_ROOT = pathlib.Path("/private/tmp") +MANIFEST_FILE = "manifest.json" +MANIFEST_SHA_FILE = "manifest.sha256" +UPGRADE_SCHEMA_KEYS = ( + "character", "location", "item", "faction", "power_system", "event", + "character_relation", +) + + +class BackupError(RuntimeError): + """备份输入、数据库快照或工件写入不满足合同。""" + + +class VerificationError(BackupError): + """磁盘备份的结构或摘要复验失败。""" + + +@dataclass(frozen=True) +class DomainSpec: + """单个备份域的表、主键、文件名和边界查询。""" + + table: str + primary_keys: tuple[str, ...] + file_name: str + sql: str + params: tuple[str, ...] + + +# 七域查询都显式携带租户和作品边界。drafts 不过滤 deleted；windows 按需求只导出 active 行； +# audits/embeddings 不信任自身缺失的 work_id，而是通过同租户、同作品、同来源 draft 子查询定位。 +DOMAIN_SPECS: dict[str, DomainSpec] = { + "drafts": DomainSpec( + table="muse_knowledge_draft", + primary_keys=("id",), + file_name="drafts.jsonl", + sql="""SELECT d.* + FROM muse_knowledge_draft d + WHERE d.tenant_id = %s AND d.work_id = %s AND d.source_type = %s + ORDER BY d.id""", + params=("tenant", "work", "source"), + ), + "windows": DomainSpec( + table="example_upgrade_window", + primary_keys=("id",), + file_name="windows.jsonl", + sql="""SELECT w.* + FROM example_upgrade_window w + WHERE w.tenant_id = %s AND w.work_id = %s AND w.deleted = FALSE + ORDER BY w.id""", + params=("tenant", "work"), + ), + "aliases": DomainSpec( + table="example_upgrade_alias", + primary_keys=("id",), + file_name="aliases.jsonl", + sql="""SELECT a.* + FROM example_upgrade_alias a + WHERE a.tenant_id = %s AND a.work_id = %s + ORDER BY a.id""", + params=("tenant", "work"), + ), + "presence": DomainSpec( + table="example_upgrade_presence", + primary_keys=("id",), + file_name="presence.jsonl", + sql="""SELECT p.* + FROM example_upgrade_presence p + WHERE p.tenant_id = %s AND p.work_id = %s + ORDER BY p.id""", + params=("tenant", "work"), + ), + "card_state": DomainSpec( + table="example_upgrade_card_state", + primary_keys=("draft_id",), + file_name="card_state.jsonl", + sql="""SELECT s.* + FROM example_upgrade_card_state s + WHERE s.tenant_id = %s AND s.work_id = %s + ORDER BY s.draft_id""", + params=("tenant", "work"), + ), + "audits": DomainSpec( + table="example_upgrade_audit", + primary_keys=("id",), + file_name="audits.jsonl", + sql="""SELECT a.* + FROM example_upgrade_audit a + WHERE a.tenant_id = %s AND a.draft_id IN ( + SELECT d.id + FROM muse_knowledge_draft d + WHERE d.tenant_id = %s AND d.work_id = %s AND d.source_type = %s + ) + ORDER BY a.id""", + params=("tenant", "tenant", "work", "source"), + ), + "embeddings": DomainSpec( + table="example_knowledge_embedding", + primary_keys=("id",), + file_name="embeddings.jsonl", + # vector 显式转 text，避免不同 pgvector/psycopg 适配器返回不同 Python 类型。 + sql="""SELECT e.id, e.draft_id, e.entity_id, e.content_hash, e.embed_text, + e.model, e.dimensions, e.embedding::text AS embedding, + e.creator, e.create_time, e.updater, e.update_time, + e.deleted, e.tenant_id + FROM example_knowledge_embedding e + WHERE e.tenant_id = %s AND e.draft_id IN ( + SELECT d.id + FROM muse_knowledge_draft d + WHERE d.tenant_id = %s AND d.work_id = %s AND d.source_type = %s + ) + ORDER BY e.id""", + params=("tenant", "tenant", "work", "source"), + ), +} + + +def canonicalize(value: Any) -> Any: + """把数据库常见类型转成无歧义、可重复 JSON 序列化的值。""" + + if value is None or isinstance(value, (str, bool, int)): + return value + if isinstance(value, Decimal): + # Decimal 使用字符串保留尾零和完整精度，避免二进制浮点改变摘要。 + return str(value) + if isinstance(value, float): + if not math.isfinite(value): + raise ValueError("JSON 不允许 NaN 或 Infinity") + return value + if isinstance(value, dt.datetime): + if value.tzinfo is not None: + utc_value = value.astimezone(dt.timezone.utc) + return utc_value.isoformat().replace("+00:00", "Z") + return value.isoformat() + if isinstance(value, (dt.date, dt.time)): + return value.isoformat() + if isinstance(value, uuid.UUID): + return str(value) + if isinstance(value, (bytes, bytearray, memoryview)): + return {"$base64": base64.b64encode(bytes(value)).decode("ascii")} + if isinstance(value, Mapping): + return {str(key): canonicalize(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [canonicalize(item) for item in value] + # 某些 vector 适配器返回 ndarray；先转普通列表，再递归处理 Decimal 等元素。 + to_list = getattr(value, "tolist", None) + if callable(to_list): + return canonicalize(to_list()) + # PostgreSQL 扩展类型若没有专用适配器，字符串输出是其稳定文本表示。 + return str(value) + + +def canonical_json(value: Any) -> str: + """生成键排序、无多余空格、禁止非标准浮点值的规范 JSON。""" + + return json.dumps( + canonicalize(value), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ) + + +def _sha256(data: bytes) -> str: + """返回字节内容的小写 SHA-256。""" + + return hashlib.sha256(data).hexdigest() + + +def _primary_key_values(spec: DomainSpec, rows: Sequence[Mapping[str, Any]]) -> list[list[Any]]: + """提取稳定排序后的主键元组，缺字段时立即失败。""" + + values = [] + for row in rows: + try: + values.append([canonicalize(row[key]) for key in spec.primary_keys]) + except KeyError as error: + raise VerificationError(f"{spec.file_name} 缺少主键字段：{error.args[0]}") from error + return values + + +def sort_rows(domain: str, rows: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]: + """按域主键稳定排序并规范化整行，消除数据库返回顺序和类型差异。""" + + if domain not in DOMAIN_SPECS: + raise ValueError(f"未知备份域：{domain}") + spec = DOMAIN_SPECS[domain] + normalized = [] + for row in rows: + if not isinstance(row, Mapping): + raise BackupError( + f"{domain} 查询返回 {type(row).__name__}，连接必须使用 psycopg.rows.dict_row" + ) + normalized.append(canonicalize(dict(row))) + + def sort_key(row: Mapping[str, Any]) -> tuple[Any, ...]: + try: + return tuple(row[key] for key in spec.primary_keys) + except KeyError as error: + raise VerificationError(f"{domain} 缺少主键字段：{error.args[0]}") from error + + return sorted(normalized, key=sort_key) + + +def _content_sha(rows: Sequence[Mapping[str, Any]]) -> str: + """对规范 JSON 数组计算内容摘要，避免 JSONL 行连接的边界歧义。""" + + return _sha256(canonical_json(list(rows)).encode("utf-8")) + + +def _primary_key_sha(spec: DomainSpec, rows: Sequence[Mapping[str, Any]]) -> str: + """对稳定主键元组数组计算摘要。""" + + return _sha256(canonical_json(_primary_key_values(spec, rows)).encode("utf-8")) + + +def _jsonl_bytes(rows: Sequence[Mapping[str, Any]]) -> bytes: + """生成 UTF-8 JSONL；空域对应空文件，非空文件以换行结尾。""" + + if not rows: + return b"" + return ("\n".join(canonical_json(row) for row in rows) + "\n").encode("utf-8") + + +def _window_boundary_sha(rows: Sequence[Mapping[str, Any]]) -> str: + """只摘要 active 窗的业务序号与章边界，不让状态和时间字段污染边界身份。""" + + boundaries = [ + { + "windowNo": row.get("window_no"), + "fromChapter": row.get("from_chapter"), + "toChapter": row.get("to_chapter"), + } + for row in rows + ] + boundaries.sort(key=lambda row: (row["fromChapter"], row["toChapter"], row["windowNo"])) + return _sha256(canonical_json(boundaries).encode("utf-8")) + + +def _is_under_backup_root(path: pathlib.Path) -> bool: + """使用解析后的路径判断，防止 ``..`` 或符号链接绕过目录边界。""" + + try: + path.relative_to(BACKUP_ROOT.resolve()) + return path != BACKUP_ROOT.resolve() + except ValueError: + return False + + +def validate_new_output_dir(output_dir: pathlib.Path) -> pathlib.Path: + """校验目标是 /private/tmp 下尚不存在的新目录，并返回绝对解析路径。""" + + expanded = pathlib.Path(output_dir).expanduser().absolute() + try: + expanded.relative_to(BACKUP_ROOT) + except ValueError as error: + raise ValueError("输出目录必须显式位于 /private/tmp 下") from error + resolved = expanded.resolve(strict=False) + if not _is_under_backup_root(resolved): + raise ValueError("输出目录必须是 /private/tmp 下的子目录") + if resolved.exists() or resolved.is_symlink(): + raise FileExistsError(f"输出目录已存在，禁止覆盖：{resolved}") + return resolved + + +def _validate_existing_backup_dir(backup_dir: pathlib.Path) -> pathlib.Path: + """校验离线复验目标仍位于允许的备份根目录。""" + + resolved = pathlib.Path(backup_dir).expanduser().resolve(strict=True) + if not _is_under_backup_root(resolved) or not resolved.is_dir(): + raise VerificationError("备份目录必须是 /private/tmp 下的现有目录") + return resolved + + +def _atomic_write(path: pathlib.Path, data: bytes) -> None: + """同目录写临时文件、落盘后原子替换，任何异常不遗留半文件。""" + + temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") + try: + descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(descriptor, "wb") as stream: + stream.write(data) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + finally: + temporary.unlink(missing_ok=True) + + +def _git_commit() -> str: + """读取脚本所在仓库当前提交；不把工作区 diff 或远端信息写入备份。""" + + repository = pathlib.Path(__file__).resolve().parents[4] + result = subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=repository, + check=True, + capture_output=True, + text=True, + ) + return result.stdout.strip() + + +def _utc_created_at() -> str: + """生成固定 UTC 语义的 ISO-8601 时间。""" + + return dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z") + + +def _query_params(names: Sequence[str], tenant: int, work_id: int) -> tuple[Any, ...]: + """按域声明组装参数，避免 SQL 与边界参数错位。""" + + values = {"tenant": tenant, "work": work_id, "source": SOURCE_TYPE} + return tuple(values[name] for name in names) + + +def _require_mapping(row: Any, context: str) -> Mapping[str, Any]: + """数据库读取统一要求 dict_row，防止 tuple 被位置式误解。""" + + if not isinstance(row, Mapping): + raise BackupError(f"{context} 未返回 dict_row") + return row + + +def _file_sha(path: pathlib.Path) -> str: + """计算本地代码或 skill 文件摘要。""" + + return _sha256(path.read_bytes()) + + +def _repository_root() -> pathlib.Path: + """返回 agent-example Git 根目录。""" + + return pathlib.Path(__file__).resolve().parents[4] + + +def capture_code_identity(git_commit: str | None = None) -> dict[str, Any]: + """绑定本次实际执行文件；未提交代码用 fileSha 表达，不借 HEAD 冒充。""" + + here = pathlib.Path(__file__).resolve().parent + root = _repository_root() + files = { + "backup_upgrade_work.py": pathlib.Path(__file__).resolve(), + "parse_upgrade.py": here / "parse_upgrade.py", + "parse_llm.py": here / "parse_llm.py", + "embed_drafts.py": here.parents[1] / "embed" / "scripts" / "embed_drafts.py", + "upgrade_work_lock.py": here / "upgrade_work_lock.py", + "llm.py": here.parents[1] / "llm" / "scripts" / "llm.py", + "parse-book/SKILL.md": here.parent / "SKILL.md", + } + commit = git_commit or _git_commit() + file_shas = {name: _file_sha(path) for name, path in files.items()} + head_matches = {} + for name, path in files.items(): + relative = path.relative_to(root).as_posix() + result = subprocess.run( + ["git", "show", f"{commit}:{relative}"], + cwd=root, + capture_output=True, + ) + head_matches[name] = result.returncode == 0 and _sha256(result.stdout) == file_shas[name] + return { + "gitCommit": commit, + "gitHeadCoversCodeFiles": all(head_matches.values()), + "codeFiles": file_shas, + "headMatches": head_matches, + } + + +def _database_identity(conn: Any) -> dict[str, Any]: + """读取不含用户名和密码的数据库身份。""" + + row = conn.execute( + """SELECT current_database() AS database, + inet_server_addr()::text AS server_addr, + inet_server_port() AS server_port, + current_setting('server_version_num') AS server_version""" + ).fetchone() + row = _require_mapping(row, "database identity") + return { + "database": row["database"], + "serverAddr": row["server_addr"], + "serverPort": row["server_port"], + "serverVersion": row["server_version"], + } + + +def _canonical_content_rows(conn: Any, work_id: int, tenant: int) -> list[dict[str, Any]]: + """按章节和 block 业务顺序读取 Canonical 正文。""" + + rows = conn.execute( + """SELECT c.id AS chapter_id, c.order_no AS chapter_order, c.title AS chapter_title, + c.revision AS chapter_revision, + b.id AS block_id, b.order_no AS block_order, b.block_type, + b.title AS block_title, b.content_doc, b.content_text, + b.revision AS block_revision + FROM muse_content_chapter c + LEFT JOIN muse_content_block b + ON b.tenant_id = c.tenant_id AND b.work_id = c.work_id + AND b.chapter_id = c.id AND b.deleted = FALSE + WHERE c.tenant_id = %s AND c.work_id = %s AND c.deleted = FALSE + ORDER BY c.order_no, c.id, b.order_no NULLS FIRST, b.id NULLS FIRST""", + (tenant, work_id), + ).fetchall() + return [canonicalize(dict(_require_mapping(row, "Canonical content"))) for row in rows] + + +def _canonical_content_identity(normalized: Sequence[Mapping[str, Any]]) -> dict[str, Any]: + """摘要已按业务顺序读取的 Canonical 行。""" + + chapter_ids = {row["chapter_id"] for row in normalized} + block_ids = {row["block_id"] for row in normalized if row.get("block_id") is not None} + return { + "chapterCount": len(chapter_ids), + "blockCount": len(block_ids), + "sha": _sha256(canonical_json(normalized).encode("utf-8")), + } + + +def _nonempty_block(row: Mapping[str, Any]) -> bool: + """block 有 ID，且文本或结构化文档至少一项非空。""" + + if row.get("block_id") is None: + return False + text = row.get("content_text") + document = row.get("content_doc") + return bool(isinstance(text, str) and text.strip()) or document not in (None, {}, [], "") + + +def validate_expected_scope( + canonical_rows: Sequence[Mapping[str, Any]], + windows: Sequence[Mapping[str, Any]], + *, + expected_chapters: int, + expected_windows: int, +) -> None: + """机械验证章节/block 与窗口号、章域完整连续。""" + + if expected_chapters <= 0 or expected_windows <= 0: + raise BackupError("expected-chapters/expected-windows 必须为正整数") + chapter_orders: dict[Any, int] = {} + chapter_has_content: dict[Any, bool] = {} + for row in canonical_rows: + row = _require_mapping(row, "expected Canonical row") + chapter_id = row.get("chapter_id") + order = row.get("chapter_order") + chapter_orders[chapter_id] = order + chapter_has_content[chapter_id] = chapter_has_content.get(chapter_id, False) or _nonempty_block(row) + if len(chapter_orders) != expected_chapters: + raise BackupError( + f"Canonical 章节数 {len(chapter_orders)} != expected {expected_chapters}" + ) + ordered_chapters = sorted(chapter_orders.values()) + if ordered_chapters != list(range(ordered_chapters[0], ordered_chapters[-1] + 1)): + raise BackupError("Canonical 章节 order_no 不连续") + missing_blocks = sorted( + chapter_orders[chapter_id] + for chapter_id, has_content in chapter_has_content.items() if not has_content + ) + if missing_blocks: + raise BackupError(f"章节缺少非空 block：{missing_blocks[:20]}") + + if len(windows) != expected_windows: + raise BackupError(f"active 窗数 {len(windows)} != expected {expected_windows}") + ordered_windows = sorted(windows, key=lambda row: row["window_no"]) + window_numbers = [row["window_no"] for row in ordered_windows] + if window_numbers != list(range(1, expected_windows + 1)): + raise BackupError("active window_no 必须从 1 连续到 expected-windows") + cursor = ordered_chapters[0] + for row in ordered_windows: + start, end = row["from_chapter"], row["to_chapter"] + if start != cursor or end < start: + raise BackupError("active 窗章域存在头部、中部缝隙或非法倒序") + cursor = end + 1 + if cursor - 1 != ordered_chapters[-1]: + raise BackupError("active 窗章域未覆盖 Canonical 尾章") + + +def _active_contract_identity(conn: Any, tenant: int) -> dict[str, Any]: + """摘要升格七型当前激活字段合同和版本。""" + + rows = conn.execute( + """SELECT s.schema_key, s.active_version_id, v.field_contract_snapshot + FROM muse_meta_schema s + JOIN muse_meta_schema_version v ON v.id = s.active_version_id + WHERE s.tenant_id = %s AND s.schema_key = ANY(%s) + ORDER BY s.schema_key""", + (tenant, list(UPGRADE_SCHEMA_KEYS)), + ).fetchall() + normalized = [ + canonicalize(dict(_require_mapping(row, "active field contracts"))) for row in rows + ] + keys = [row["schema_key"] for row in normalized] + if keys != sorted(UPGRADE_SCHEMA_KEYS): + raise BackupError(f"升格 active 字段合同不完整：{keys}") + return { + "count": len(normalized), + "schemaKeys": keys, + "sha": _sha256(canonical_json(normalized).encode("utf-8")), + } + + +def _load_active_windows(conn: Any, work_id: int, tenant: int) -> list[dict[str, Any]]: + """恢复前重算输入时读取 active 窗边界。""" + + spec = DOMAIN_SPECS["windows"] + rows = conn.execute(spec.sql, _query_params(spec.params, tenant, work_id)).fetchall() + return sort_rows("windows", rows) + + +def capture_input_snapshot( + conn: Any, + work_id: int, + tenant: int, + code_identity: Mapping[str, Any], + expected_chapters: int, + expected_windows: int, + windows: Sequence[Mapping[str, Any]] | None = None, +) -> dict[str, Any]: + """在当前只读快照中绑定 DB、Canonical、窗、合同、代码和计划参数。""" + + active_windows = list(windows) if windows is not None else _load_active_windows( + conn, work_id, tenant + ) + canonical_rows = _canonical_content_rows(conn, work_id, tenant) + validate_expected_scope( + canonical_rows, + active_windows, + expected_chapters=expected_chapters, + expected_windows=expected_windows, + ) + return { + "databaseIdentity": _database_identity(conn), + "canonicalContent": _canonical_content_identity(canonical_rows), + "windowBoundary": { + "count": len(active_windows), + "sha": _window_boundary_sha(active_windows), + }, + "activeFieldContracts": _active_contract_identity(conn, tenant), + "codeFiles": dict(code_identity["codeFiles"]), + "expectedChapters": expected_chapters, + "expectedWindows": expected_windows, + "plan": {"model": PLANNED_MODEL, "semanticDedup": PLANNED_SEMANTIC_DEDUP}, + "sourceType": SOURCE_TYPE, + "tenant": tenant, + "work": work_id, + } + + +def _read_snapshot(conn: Any, work_id: int, tenant: int) -> tuple[str, dict[str, list[dict[str, Any]]]]: + """在调用方已开启的只读事务内读取作品标题和七域完整快照。""" + + work = conn.execute( + """SELECT title FROM muse_content_work + WHERE tenant_id = %s AND id = %s""", + (tenant, work_id), + ).fetchone() + if not work: + raise BackupError(f"作品不存在或不属于 tenant={tenant}：work={work_id}") + title = _require_mapping(work, "work title")["title"] + domains = {} + for name, spec in DOMAIN_SPECS.items(): + rows = conn.execute(spec.sql, _query_params(spec.params, tenant, work_id)).fetchall() + domains[name] = sort_rows(name, rows) + return str(title), domains + + +def _build_manifest( + backup_id: str, + work_id: int, + tenant: int, + title: str, + git_commit: str, + input_summary: Mapping[str, Any], + git_head_covers_code: bool, + domains: Mapping[str, Sequence[Mapping[str, Any]]], + files: Mapping[str, bytes], +) -> dict[str, Any]: + """构造不含 DSN/密码的自描述清单。""" + + domain_manifest = {} + for name, spec in DOMAIN_SPECS.items(): + rows = domains[name] + file_data = files[name] + domain_manifest[name] = { + "file": spec.file_name, + "rowCount": len(rows), + "primaryKeySha": _primary_key_sha(spec, rows), + "contentSha": _content_sha(rows), + "fileSha": _sha256(file_data), + } + input_sha = _sha256(canonical_json(input_summary).encode("utf-8")) + confirmation = { + "backup_id": backup_id, + "codeFiles": input_summary["codeFiles"], + "inputSha": input_sha, + "tenant": tenant, + "work": work_id, + } + return { + "schemaVersion": 1, + "backup_id": backup_id, + "work": work_id, + "tenant": tenant, + "title": title, + "createdAt": _utc_created_at(), + "gitCommit": git_commit, + "gitHeadCoversCodeFiles": git_head_covers_code, + "windowBoundarySha": input_summary["windowBoundary"]["sha"], + "input": dict(input_summary), + "inputSha": input_sha, + "confirmationSha": _sha256(canonical_json(confirmation).encode("utf-8")), + "domains": domain_manifest, + } + + +def _write_artifacts( + output_dir: pathlib.Path, + manifest: Mapping[str, Any], + files: Mapping[str, bytes], +) -> None: + """按七域、清单、清单摘要的顺序原子发布所有工件。""" + + for name, spec in DOMAIN_SPECS.items(): + _atomic_write(output_dir / spec.file_name, files[name]) + manifest_data = (canonical_json(manifest) + "\n").encode("utf-8") + _atomic_write(output_dir / MANIFEST_FILE, manifest_data) + manifest_sha = f"{_sha256(manifest_data)} {MANIFEST_FILE}\n".encode("ascii") + _atomic_write(output_dir / MANIFEST_SHA_FILE, manifest_sha) + + +def _read_jsonl(path: pathlib.Path) -> tuple[bytes, list[dict[str, Any]]]: + """从磁盘读取并严格解析 JSONL，不接受空行或非对象行。""" + + data = path.read_bytes() + rows = [] + for line_number, line in enumerate(data.splitlines(), start=1): + if not line.strip(): + raise VerificationError(f"{path.name} 第 {line_number} 行为空") + try: + row = json.loads(line) + except json.JSONDecodeError as error: + raise VerificationError(f"{path.name} 第 {line_number} 行不是合法 JSON") from error + if not isinstance(row, dict): + raise VerificationError(f"{path.name} 第 {line_number} 行不是 JSON 对象") + rows.append(row) + return data, rows + + +def verify_backup(backup_dir: pathlib.Path) -> dict[str, Any]: + """只依赖磁盘工件，重算清单及七域全部摘要，成功时返回清单。""" + + root = _validate_existing_backup_dir(backup_dir) + manifest_path = root / MANIFEST_FILE + manifest_sha_path = root / MANIFEST_SHA_FILE + try: + manifest_data = manifest_path.read_bytes() + expected_manifest_sha = manifest_sha_path.read_text(encoding="ascii").split()[0] + except (FileNotFoundError, IndexError, UnicodeError) as error: + raise VerificationError("缺少或无法读取 manifest 工件") from error + if _sha256(manifest_data) != expected_manifest_sha: + raise VerificationError("manifest.json 文件摘要不一致") + try: + manifest = json.loads(manifest_data) + except json.JSONDecodeError as error: + raise VerificationError("manifest.json 不是合法 JSON") from error + if not isinstance(manifest, dict): + raise VerificationError("manifest.json 顶层必须是对象") + required = { + "backup_id", "work", "tenant", "title", "createdAt", "gitCommit", + "gitHeadCoversCodeFiles", "windowBoundarySha", "input", "inputSha", + "confirmationSha", "domains", + } + missing = sorted(required - set(manifest)) + if missing: + raise VerificationError(f"manifest 缺少字段：{', '.join(missing)}") + try: + uuid.UUID(str(manifest["backup_id"])) + except ValueError as error: + raise VerificationError("backup_id 不是合法 UUID") from error + if set(manifest["domains"]) != set(DOMAIN_SPECS): + raise VerificationError("manifest 七域集合不完整或包含未知域") + input_sha = _sha256(canonical_json(manifest["input"]).encode("utf-8")) + if input_sha != manifest["inputSha"]: + raise VerificationError("inputSha 与输入摘要不一致") + input_summary = manifest["input"] + if not isinstance(input_summary, dict): + raise VerificationError("input 摘要必须是对象") + if input_summary.get("sourceType") != SOURCE_TYPE \ + or input_summary.get("tenant") != manifest["tenant"] \ + or input_summary.get("work") != manifest["work"]: + raise VerificationError("input 摘要与 work/tenant/sourceType 不一致") + required_input = { + "databaseIdentity", "canonicalContent", "windowBoundary", + "activeFieldContracts", "codeFiles", "expectedChapters", "expectedWindows", "plan", + } + if not required_input.issubset(input_summary): + raise VerificationError("input 摘要缺少数据库、Canonical、窗、合同、代码或计划身份") + if input_summary["plan"] != { + "model": PLANNED_MODEL, "semanticDedup": PLANNED_SEMANTIC_DEDUP, + }: + raise VerificationError("input 计划模型或 semantic-dedup 与恢复合同不一致") + if input_summary["canonicalContent"].get("chapterCount") != input_summary["expectedChapters"] \ + or input_summary["windowBoundary"].get("count") != input_summary["expectedWindows"]: + raise VerificationError("input 实际章节/窗口数与 expected 值不一致") + required_code = { + "backup_upgrade_work.py", "parse_upgrade.py", "parse_llm.py", "embed_drafts.py", + "upgrade_work_lock.py", "llm.py", "parse-book/SKILL.md", + } + if not required_code.issubset(input_summary["codeFiles"]): + raise VerificationError("input codeFiles 未覆盖升格实际依赖") + confirmation = { + "backup_id": manifest["backup_id"], + "codeFiles": input_summary["codeFiles"], + "inputSha": manifest["inputSha"], + "tenant": manifest["tenant"], + "work": manifest["work"], + } + if _sha256(canonical_json(confirmation).encode("utf-8")) != manifest["confirmationSha"]: + raise VerificationError("confirmationSha 未绑定 backup/input/code fileSha") + + disk_rows = {} + for name, spec in DOMAIN_SPECS.items(): + expected = manifest["domains"].get(name) + if not isinstance(expected, dict) or expected.get("file") != spec.file_name: + raise VerificationError(f"{name} 的文件声明不合法") + try: + file_data, rows = _read_jsonl(root / spec.file_name) + except FileNotFoundError as error: + raise VerificationError(f"缺少域文件：{spec.file_name}") from error + ordered = sort_rows(name, rows) + if rows != ordered: + raise VerificationError(f"{spec.file_name} 未按主键稳定排序") + actual = { + "file": spec.file_name, + "rowCount": len(rows), + "primaryKeySha": _primary_key_sha(spec, rows), + "contentSha": _content_sha(rows), + "fileSha": _sha256(file_data), + } + if actual != expected: + raise VerificationError(f"{spec.file_name} 行数或摘要不一致") + disk_rows[name] = rows + boundary_sha = _window_boundary_sha(disk_rows["windows"]) + if boundary_sha != manifest["windowBoundarySha"]: + raise VerificationError("windowBoundarySha 与 windows.jsonl 不一致") + if boundary_sha != input_summary["windowBoundary"]["sha"] \ + or len(disk_rows["windows"]) != input_summary["windowBoundary"]["count"]: + raise VerificationError("input 窗边界摘要与 windows.jsonl 不一致") + return manifest + + +def _default_connect(dsn: str, **kwargs: Any) -> Any: + """仅 backup 命令真正执行时才加载 psycopg，保持 verify 可纯离线运行。""" + + import psycopg + from psycopg.rows import dict_row + + return psycopg.connect(dsn, row_factory=dict_row, **kwargs) + + +def _load_db_config() -> tuple[str, int]: + """延迟复用 parse-book 既有数据库配置；本脚本和清单均不复制密码。""" + + from parse_llm import DSN, TENANT + + return DSN, TENANT + + +def backup_work( + work_id: int, + output_dir: pathlib.Path | None = None, + *, + tenant: int, + dsn: str, + expected_chapters: int, + expected_windows: int, + connect: Callable[..., Any] = _default_connect, + git_commit: str | None = None, + backup_id: str | None = None, + code_identity: Mapping[str, Any] | None = None, + lock_context: Callable[..., Any] = upgrade_work_lock, + lock_connect: Callable[..., Any] | None = None, +) -> dict[str, Any]: + """持同书锁创建备份；锁失败时不创建业务连接或输出目录。""" + + with lock_context(dsn, tenant, work_id, connect=lock_connect): + return _backup_work_locked( + work_id, + output_dir, + tenant=tenant, + dsn=dsn, + expected_chapters=expected_chapters, + expected_windows=expected_windows, + connect=connect, + git_commit=git_commit, + backup_id=backup_id, + code_identity=code_identity, + ) + + +def _backup_work_locked( + work_id: int, + output_dir: pathlib.Path | None = None, + *, + tenant: int, + dsn: str, + expected_chapters: int, + expected_windows: int, + connect: Callable[..., Any] = _default_connect, + git_commit: str | None = None, + backup_id: str | None = None, + code_identity: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + """创建并立即离线复验备份；任何失败只删除本次新建目录。""" + + identifier = backup_id or str(uuid.uuid4()) + try: + uuid.UUID(identifier) + except ValueError as error: + raise BackupError("backup_id 必须是合法 UUID") from error + requested = output_dir or (BACKUP_ROOT / f"upgrade-work-{work_id}-{identifier}") + target = validate_new_output_dir(pathlib.Path(requested)) + created = False + try: + target.mkdir(mode=0o700, parents=False, exist_ok=False) + created = True + actual_code = dict(code_identity or capture_code_identity(git_commit)) + # autocommit=True 让事务边界完全由下面唯一一组 BEGIN/COMMIT 控制，不产生隐式第二事务。 + with connect(dsn, autocommit=True) as conn: + conn.execute("BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY") + try: + title, domains = _read_snapshot(conn, work_id, tenant) + input_summary = capture_input_snapshot( + conn, + work_id, + tenant, + actual_code, + expected_chapters, + expected_windows, + windows=domains["windows"], + ) + conn.execute("COMMIT") + except Exception: + conn.execute("ROLLBACK") + raise + files = {name: _jsonl_bytes(rows) for name, rows in domains.items()} + manifest = _build_manifest( + backup_id=identifier, + work_id=work_id, + tenant=tenant, + title=title, + git_commit=actual_code["gitCommit"], + input_summary=input_summary, + git_head_covers_code=actual_code["gitHeadCoversCodeFiles"], + domains=domains, + files=files, + ) + _write_artifacts(target, manifest, files) + # 必须退出连接上下文后再走只读磁盘复验，避免误用内存结果或仍存活的 DB 连接。 + return verify_backup(target) + except Exception: + if created: + shutil.rmtree(target, ignore_errors=True) + raise + + +def read_backup_rows(backup_dir: pathlib.Path) -> dict[str, list[dict[str, Any]]]: + """离线 verify 通过后，从七个固定文件加载恢复输入。""" + + root = _validate_existing_backup_dir(backup_dir) + return { + name: _read_jsonl(root / spec.file_name)[1] for name, spec in DOMAIN_SPECS.items() + } + + +def assert_restore_input( + manifest: Mapping[str, Any], + current_input: Mapping[str, Any], + *, + work_id: int, + tenant: int, +) -> None: + """恢复前拒绝 work、tenant、数据库身份或任一输入摘要漂移。""" + + if manifest.get("work") != work_id or manifest.get("tenant") != tenant: + raise BackupError("恢复 work/tenant 与备份 manifest 不一致") + if current_input.get("databaseIdentity") != manifest["input"].get("databaseIdentity"): + raise BackupError("当前数据库 identity 与备份不一致") + current_sha = _sha256(canonical_json(current_input).encode("utf-8")) + if current_sha != manifest.get("inputSha"): + raise BackupError("Canonical、窗、字段合同、代码文件或计划参数发生 input 漂移") + + +def validate_execute_confirmation( + execute: bool, + supplied_backup_id: str | None, + manifest_backup_id: str, + supplied_confirmation_sha: str | None, + manifest_confirmation_sha: str, +) -> None: + """rehearse/restore 都核对 confirmation；execute 再核对 backup_id。""" + + if not supplied_confirmation_sha or supplied_confirmation_sha != manifest_confirmation_sha: + raise BackupError("必须提供与 manifest 精确匹配的 --confirmation-sha") + if not supplied_backup_id or supplied_backup_id != manifest_backup_id: + if execute: + raise BackupError("--execute 必须同时提供与 manifest 匹配的 --backup-id") + + +def _sql_identifier(name: str) -> str: + """恢复 SQL 只允许固定的小写标识符，禁止从工件注入 SQL。""" + + if not re.fullmatch(r"[a-z][a-z0-9_]*", name): + raise BackupError(f"非法 SQL 标识符：{name}") + return name + + +JSON_COLUMNS = { + "muse_knowledge_draft": { + "proposed_changes", "draft_payload", "current_canonical_snapshot", + "risk_marker_snapshot", + }, +} + + +def _upsert_row(conn: Any, spec: DomainSpec, row: Mapping[str, Any]) -> None: + """按备份行全部列 exact upsert；JSON/vector 使用显式目标类型。""" + + table = _sql_identifier(spec.table) + columns = [_sql_identifier(column) for column in row] + if not columns: + raise BackupError(f"{table} 恢复行为空") + placeholders = [] + values = [] + for column in columns: + value = row[column] + if column in JSON_COLUMNS.get(table, set()): + placeholders.append("CAST(%s AS jsonb)") + values.append(canonical_json(value) if value is not None else None) + elif table == "example_knowledge_embedding" and column == "embedding": + placeholders.append("CAST(%s AS vector)") + values.append(value) + else: + placeholders.append("%s") + values.append(value) + conflict = ", ".join(spec.primary_keys) + updates = [column for column in columns if column not in spec.primary_keys] + update_sql = ", ".join(f"{column} = EXCLUDED.{column}" for column in updates) + overriding = " OVERRIDING SYSTEM VALUE" if table in { + "muse_knowledge_draft", "example_knowledge_embedding" + } else "" + sql = ( + f"INSERT INTO {table} ({', '.join(columns)}){overriding} " + f"VALUES ({', '.join(placeholders)}) ON CONFLICT ({conflict}) DO UPDATE SET {update_sql}" + ) + conn.execute(sql, tuple(values)) + + +def _restore_ids(rows: Sequence[Mapping[str, Any]], key: str) -> list[Any]: + """提取恢复主键；空集合仍以空数组传给 ANY。""" + + return [row[key] for row in rows] + + +def apply_restore_kernel( + conn: Any, + manifest: Mapping[str, Any], + rows: Mapping[str, Sequence[Mapping[str, Any]]], +) -> None: + """rehearse/restore 共用七域恢复内核，顺序兼容未来外键收紧。""" + + work_id = manifest["work"] + tenant = manifest["tenant"] + draft_ids = _restore_ids(rows["drafts"], "id") + window_ids = _restore_ids(rows["windows"], "id") + embedding_ids = _restore_ids(rows["embeddings"], "id") + + # 额外 active 窗不是 reset 的合法副作用，必须在任何恢复写入前拒绝。 + extra_window = conn.execute( + """SELECT w.id FROM example_upgrade_window w + WHERE w.tenant_id = %s AND w.work_id = %s AND w.deleted = FALSE + AND NOT (w.id = ANY(%s)) LIMIT 1""", + (tenant, work_id, window_ids), + ).fetchone() + if extra_window: + raise BackupError("当前存在备份之外的 active 窗，拒绝恢复") + + # 先清引用域，再恢复父行；audit 通过同 work 的全部升格 draft 定位。 + conn.execute( + """DELETE FROM example_upgrade_audit a + WHERE a.tenant_id = %s AND a.draft_id IN ( + SELECT d.id FROM muse_knowledge_draft d + WHERE d.tenant_id = %s AND d.work_id = %s AND d.source_type = %s + )""", + (tenant, tenant, work_id, SOURCE_TYPE), + ) + for table in ("example_upgrade_card_state", "example_upgrade_alias", "example_upgrade_presence"): + conn.execute( + f"DELETE FROM {_sql_identifier(table)} WHERE tenant_id = %s AND work_id = %s", + (tenant, work_id), + ) + + # reset 后新增的同 work 升格卡不物删，只软删；备份 ID 随后 exact upsert 恢复原 deleted 值。 + conn.execute( + """UPDATE muse_knowledge_draft d SET deleted = TRUE + WHERE d.tenant_id = %s AND d.work_id = %s AND d.source_type = %s + AND NOT (d.id = ANY(%s))""", + (tenant, work_id, SOURCE_TYPE, draft_ids), + ) + # 额外向量同样只软删，范围只覆盖当前 work 的升格 draft。 + conn.execute( + """UPDATE example_knowledge_embedding e SET deleted = TRUE + WHERE e.tenant_id = %s AND NOT (e.id = ANY(%s)) AND EXISTS ( + SELECT 1 FROM muse_knowledge_draft d + WHERE d.id = e.draft_id AND d.tenant_id = %s + AND d.work_id = %s AND d.source_type = %s + )""", + (tenant, embedding_ids, tenant, work_id, SOURCE_TYPE), + ) + + for name in ("drafts", "windows", "aliases", "presence", "card_state", "audits", "embeddings"): + spec = DOMAIN_SPECS[name] + for row in rows[name]: + _upsert_row(conn, spec, row) + + +def verify_restored_domains( + conn: Any, + manifest: Mapping[str, Any], + expected: Mapping[str, Sequence[Mapping[str, Any]]], +) -> None: + """事务内逐域读回；备份 ID exact，新增 draft/embedding 只能是软删态。""" + + _title, actual = _read_snapshot(conn, manifest["work"], manifest["tenant"]) + for name in ("windows", "aliases", "presence", "card_state", "audits"): + if actual[name] != list(expected[name]): + raise BackupError(f"恢复读回不一致：{name}") + for name in ("drafts", "embeddings"): + key = DOMAIN_SPECS[name].primary_keys[0] + expected_by_id = {row[key]: row for row in expected[name]} + actual_by_id = {row[key]: row for row in actual[name]} + for row_id, expected_row in expected_by_id.items(): + if actual_by_id.get(row_id) != expected_row: + raise BackupError(f"恢复读回不一致：{name} id={row_id}") + extras = [row for row_id, row in actual_by_id.items() if row_id not in expected_by_id] + if any(not row.get("deleted") for row in extras): + raise BackupError(f"恢复后存在未软删的额外 {name}") + + +SEQUENCE_TABLES = ( + "muse_knowledge_draft", "example_upgrade_window", "example_upgrade_alias", + "example_upgrade_presence", "example_upgrade_audit", "example_knowledge_embedding", +) + + +def advance_sequences_forward(conn: Any) -> None: + """仅 execute 成功读回后向前推进 identity/serial，绝不 setval 回退。""" + + for table in SEQUENCE_TABLES: + sequence_row = conn.execute( + "SELECT pg_get_serial_sequence(%s, 'id') AS sequence_name", (table,) + ).fetchone() + sequence_row = _require_mapping(sequence_row, f"{table} sequence") + sequence_name = sequence_row.get("sequence_name") + if not sequence_name: + continue + schema, name = str(sequence_name).split(".", 1) + last_row = conn.execute( + """SELECT last_value FROM pg_sequences + WHERE schemaname = %s AND sequencename = %s""", + (schema, name), + ).fetchone() + max_row = conn.execute( + f"SELECT MAX(id) AS max_id FROM {_sql_identifier(table)}" + ).fetchone() + current = _require_mapping(last_row, f"{table} sequence value").get("last_value") or 0 + maximum = _require_mapping(max_row, f"{table} max id").get("max_id") or 0 + if maximum > current: + conn.execute("SELECT setval(%s::regclass, %s, TRUE)", (sequence_name, maximum)) + + +def restore_backup( + backup_dir: pathlib.Path, + *, + work_id: int, + tenant: int, + dsn: str, + connect: Callable[..., Any] = _default_connect, + execute: bool = False, + backup_id: str | None = None, + confirmation_sha: str | None = None, + code_identity: Mapping[str, Any] | None = None, + lock_context: Callable[..., Any] = upgrade_work_lock, + lock_connect: Callable[..., Any] | None = None, +) -> dict[str, Any]: + """持同书锁完成离线 verify、演练回滚或显式恢复提交。""" + + with lock_context(dsn, tenant, work_id, connect=lock_connect): + return _restore_backup_locked( + backup_dir, + work_id=work_id, + tenant=tenant, + dsn=dsn, + connect=connect, + execute=execute, + backup_id=backup_id, + confirmation_sha=confirmation_sha, + code_identity=code_identity, + ) + + +def _restore_backup_locked( + backup_dir: pathlib.Path, + *, + work_id: int, + tenant: int, + dsn: str, + connect: Callable[..., Any], + execute: bool, + backup_id: str | None, + confirmation_sha: str | None, + code_identity: Mapping[str, Any] | None, +) -> dict[str, Any]: + """锁内先离线 verify，再在真实 public 表演练或恢复。""" + + manifest = verify_backup(backup_dir) + rows = read_backup_rows(backup_dir) + validate_execute_confirmation( + execute, + backup_id, + manifest["backup_id"], + confirmation_sha, + manifest["confirmationSha"], + ) + actual_code = dict(code_identity or capture_code_identity()) + with connect(dsn, autocommit=True) as conn: + conn.execute("BEGIN ISOLATION LEVEL REPEATABLE READ") + conn.execute("SET LOCAL search_path TO public") + finished = False + try: + current_input = capture_input_snapshot( + conn, + work_id, + tenant, + actual_code, + manifest["input"]["expectedChapters"], + manifest["input"]["expectedWindows"], + ) + assert_restore_input(manifest, current_input, work_id=work_id, tenant=tenant) + # updated_at trigger 会改写备份时间戳；事务局部 replica 模式保证 exact 恢复，结束即复原。 + conn.execute("SET LOCAL session_replication_role TO replica") + apply_restore_kernel(conn, manifest, rows) + verify_restored_domains(conn, manifest, rows) + if execute: + advance_sequences_forward(conn) + conn.execute("COMMIT") + else: + # PostgreSQL sequence 非事务性；rehearse 不调用 sequence 内核，只回滚表数据。 + conn.execute("ROLLBACK") + finished = True + finally: + if not finished: + conn.execute("ROLLBACK") + return manifest + + +@click.group() +def cli() -> None: + """升格作品 reset 前备份与离线校验。""" + + +@cli.command("backup") +@click.option("--work-id", type=click.IntRange(min=1), required=True, help="待备份作品 ID") +@click.option("--expected-chapters", type=click.IntRange(min=1), required=True) +@click.option("--expected-windows", type=click.IntRange(min=1), required=True) +@click.option( + "--output-dir", + type=click.Path(path_type=pathlib.Path, file_okay=False, resolve_path=False), + help="/private/tmp 下尚不存在的目录；不填则按 backup_id 自动生成", +) +def backup_command( + work_id: int, + expected_chapters: int, + expected_windows: int, + output_dir: pathlib.Path | None, +) -> None: + """在单个只读一致性事务中创建七域备份并立即复验。""" + + try: + dsn, tenant = _load_db_config() + manifest = backup_work( + work_id, + output_dir, + tenant=tenant, + dsn=dsn, + expected_chapters=expected_chapters, + expected_windows=expected_windows, + ) + except Exception as error: + raise click.ClickException(str(error)) from error + click.echo( + f"备份并离线复验通过：backup_id={manifest['backup_id']} " + f"work={manifest['work']} title=《{manifest['title']}》 " + f"inputSha={manifest['inputSha']} confirmationSha={manifest['confirmationSha']}" + ) + click.echo("代码 fileSha：" + canonical_json(manifest["input"]["codeFiles"])) + final_dir = output_dir or BACKUP_ROOT / f"upgrade-work-{work_id}-{manifest['backup_id']}" + click.echo(f"目录：{final_dir}") + + +@cli.command("verify") +@click.option( + "--backup-dir", + type=click.Path(path_type=pathlib.Path, exists=True, file_okay=False, resolve_path=False), + required=True, + help="待离线复验的备份目录", +) +def verify_command(backup_dir: pathlib.Path) -> None: + """不连接数据库、不调用模型，复验现有磁盘备份。""" + + try: + manifest = verify_backup(backup_dir) + except Exception as error: + raise click.ClickException(str(error)) from error + counts = ", ".join( + f"{name}={manifest['domains'][name]['rowCount']}" for name in DOMAIN_SPECS + ) + click.echo(f"离线复验通过：backup_id={manifest['backup_id']}；{counts}") + + +@cli.command("rehearse") +@click.option("--work-id", type=click.IntRange(min=1), required=True, help="恢复作品 ID") +@click.option("--confirmation-sha", required=True, help="必须与 manifest 精确一致") +@click.option( + "--backup-dir", + type=click.Path(path_type=pathlib.Path, exists=True, file_okay=False), + required=True, +) +def rehearse_command(work_id: int, confirmation_sha: str, backup_dir: pathlib.Path) -> None: + """在真实 public 表事务内恢复并读回，最终强制 ROLLBACK。""" + + try: + dsn, tenant = _load_db_config() + manifest = restore_backup( + backup_dir, + work_id=work_id, + tenant=tenant, + dsn=dsn, + execute=False, + confirmation_sha=confirmation_sha, + ) + except Exception as error: + raise click.ClickException(str(error)) from error + click.echo( + f"恢复演练读回通过并已回滚：backup_id={manifest['backup_id']} " + f"confirmationSha={manifest['confirmationSha']}" + ) + + +@cli.command("restore") +@click.option("--work-id", type=click.IntRange(min=1), required=True, help="恢复作品 ID") +@click.option( + "--backup-dir", + type=click.Path(path_type=pathlib.Path, exists=True, file_okay=False), + required=True, +) +@click.option("--backup-id", help="必须与 manifest backup_id 精确一致") +@click.option("--confirmation-sha", required=True, help="必须与 manifest 精确一致") +@click.option("--execute", is_flag=True, help="显式提交恢复；缺少时拒绝执行") +def restore_command( + work_id: int, + backup_dir: pathlib.Path, + backup_id: str | None, + confirmation_sha: str, + execute: bool, +) -> None: + """经双确认后提交恢复；不提供 --execute 时不会降级成隐式演练。""" + + if not execute: + raise click.ClickException("restore 必须显式提供 --execute；演练请使用 rehearse") + try: + dsn, tenant = _load_db_config() + manifest = restore_backup( + backup_dir, + work_id=work_id, + tenant=tenant, + dsn=dsn, + execute=True, + backup_id=backup_id, + confirmation_sha=confirmation_sha, + ) + except Exception as error: + raise click.ClickException(str(error)) from error + click.echo(f"恢复已提交：backup_id={manifest['backup_id']} work={manifest['work']}") + + +if __name__ == "__main__": + cli() diff --git a/.claude/skills/parse-book/scripts/parse_upgrade.py b/.claude/skills/parse-book/scripts/parse_upgrade.py index 9fb43b5..d2824f8 100644 --- a/.claude/skills/parse-book/scripts/parse_upgrade.py +++ b/.claude/skills/parse-book/scripts/parse_upgrade.py @@ -37,6 +37,7 @@ import psycopg sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2] / "llm" / "scripts")) from parse_llm import m3_json, SensitiveHardStop, IDENTITY, TENANT, DSN # noqa: E402 +from upgrade_work_lock import UpgradeWorkLockUnavailable, upgrade_work_lock # noqa: E402 # 语义判重（P1）复用 embed skill 的嵌入通道（同模型同维、与检索端语义对齐）—— # 只在开启 --semantic-dedup 时才真调，默认关（试跑期嵌入延后，见文件头注释）； # build_embed_text/MODEL/DIM/ACTOR 供嵌段（洞①边抽边嵌）复用检索端同源文本构造器与列常量。 @@ -1460,24 +1461,16 @@ def cli(): @click.option("--work-id", type=int, required=True) def windows(work_id): """机械切正文窗（幂等）。""" - with psycopg.connect(DSN) as conn: - total, new = cut_windows(conn, work_id) - click.echo(f"work={work_id} 切窗完成：全书 {total} 窗（本次新建 {new} 行）") + try: + with upgrade_work_lock(DSN, TENANT, work_id): + with psycopg.connect(DSN) as conn: + total, new = cut_windows(conn, work_id) + click.echo(f"work={work_id} 切窗完成：全书 {total} 窗（本次新建 {new} 行）") + except UpgradeWorkLockUnavailable as exc: + raise click.ClickException(str(exc)) from exc -@cli.command() -@click.option("--work-id", type=int, required=True) -@click.option("--max-windows", type=click.IntRange(min=0), default=0, - help="本次最多跑几个窗（0=不限）") -@click.option("--max-calls", type=click.IntRange(min=0), default=0, - help="本次 LLM 调用上限（0=不限，含敏感失败）") -@click.option("--model", default="MiniMax-M3", show_default=True) -@click.option("--redo-window", type=int, default=0, - help="仅允许当前末窗强制重跑；历史窗须全书前滚重建，后缀级联重算(P1)暂不支持") -@click.option("--semantic-dedup", "semantic_on", is_flag=True, - help="开启语义判重(P1)：立卡前召回同书近邻+M3终判治改名/跨型漏并；" - "边抽边嵌（本窗新建/更新卡窗事务后增量嵌入落库，后窗即可召回前窗的卡）；默认关") -def run(work_id, max_windows, max_calls, model, redo_window, semantic_on): +def _run(work_id, max_windows, max_calls, model, redo_window, semantic_on): """按窗顺序跑升格；redo 仅限当前末窗，历史窗须全书前滚重建。""" calls = {"n": 0} # 调用计数（含敏感失败换模型的次数由 m3_json 内部消化，此处计成功轮次） # 语义判重嵌入会话（仅开启时建；禁系统代理，走内网直连） @@ -1945,6 +1938,28 @@ def run(work_id, max_windows, max_calls, model, redo_window, semantic_on): click.echo(f"《{title}》本次完成 {done_n} 窗，LLM 调用 {calls['n']} 次") +@cli.command("run") +@click.option("--work-id", type=int, required=True) +@click.option("--max-windows", type=click.IntRange(min=0), default=0, + help="本次最多跑几个窗（0=不限）") +@click.option("--max-calls", type=click.IntRange(min=0), default=0, + help="本次 LLM 调用上限（0=不限，含敏感失败）") +@click.option("--model", default="MiniMax-M3", show_default=True) +@click.option("--redo-window", type=int, default=0, + help="仅允许当前末窗强制重跑；历史窗须全书前滚重建，后缀级联重算(P1)暂不支持") +@click.option("--semantic-dedup", "semantic_on", is_flag=True, + help="开启语义判重(P1)：立卡前召回同书近邻+M3终判治改名/跨型漏并；" + "边抽边嵌（本窗新建/更新卡窗事务后增量嵌入落库，后窗即可召回前窗的卡）；默认关") +def run(work_id, max_windows, max_calls, model, redo_window, semantic_on): + """持有同书会话锁后执行升格，锁连接可安全跨越模型调用。""" + + try: + with upgrade_work_lock(DSN, TENANT, work_id): + return _run(work_id, max_windows, max_calls, model, redo_window, semantic_on) + except UpgradeWorkLockUnavailable as exc: + raise click.ClickException(str(exc)) from exc + + @cli.command() @click.option("--work-id", type=int, required=True) def status(work_id): diff --git a/.claude/skills/parse-book/scripts/reset_upgrade_work.py b/.claude/skills/parse-book/scripts/reset_upgrade_work.py index 58aadb7..7f72ef0 100644 --- a/.claude/skills/parse-book/scripts/reset_upgrade_work.py +++ b/.claude/skills/parse-book/scripts/reset_upgrade_work.py @@ -30,14 +30,14 @@ import psycopg # 复用管线的连接与租户常量（与 parse_upgrade 同源） sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) from parse_llm import DSN, TENANT # noqa: E402 +from upgrade_work_lock import UpgradeWorkLockUnavailable, upgrade_work_lock # noqa: E402 SOURCE_TYPE = "upgrade_book" -@click.command() -@click.option("--work-id", type=int, required=True) -@click.option("--execute", is_flag=True, help="真执行（默认只预览统计）") -def main(work_id, execute): +def _reset(work_id, execute): + """在调用方已持有同书锁时预览或执行重抽清理。""" + with psycopg.connect(DSN) as conn: title = conn.execute("SELECT title FROM muse_content_work WHERE id=%s", (work_id,)).fetchone()[0] @@ -101,5 +101,18 @@ def main(work_id, execute): click.echo(f" ✅ 清后复核：活升格卡 {left}（应0）｜ 窗 pending {pend[0]}/{pend[1]}（应全 pending）") +@click.command() +@click.option("--work-id", type=int, required=True) +@click.option("--execute", is_flag=True, help="真执行（默认只预览统计）") +def main(work_id, execute): + """先获取同书会话锁，再预览或执行升格全量重抽准备。""" + + try: + with upgrade_work_lock(DSN, TENANT, work_id): + return _reset(work_id, execute) + except UpgradeWorkLockUnavailable as exc: + raise click.ClickException(str(exc)) from exc + + if __name__ == "__main__": main() diff --git a/.claude/skills/parse-book/scripts/test_backup_upgrade_work_offline.py b/.claude/skills/parse-book/scripts/test_backup_upgrade_work_offline.py new file mode 100644 index 0000000..b81c3d1 --- /dev/null +++ b/.claude/skills/parse-book/scripts/test_backup_upgrade_work_offline.py @@ -0,0 +1,650 @@ +#!/usr/bin/env python3 +"""升格作品备份脚本的纯离线回归测试：不连接数据库、不调用模型。""" + +import datetime as dt +import json +import pathlib +import shutil +import sys +import uuid +from contextlib import contextmanager +from decimal import Decimal +from unittest import mock + + +HERE = pathlib.Path(__file__).resolve().parent +sys.path.insert(0, str(HERE)) + +import backup_upgrade_work as backup # noqa: E402 + + +class _Result: + """模拟 psycopg 查询结果，仅暴露脚本使用的读取接口。""" + + def __init__(self, rows): + self.rows = rows + + def fetchone(self): + return self.rows[0] if self.rows else None + + def fetchall(self): + return list(self.rows) + + +class _OfflineConnection: + """按表名返回固定数据，并记录完整 SQL 与连接关闭时机。""" + + def __init__(self): + self.closed = False + self.executions = [] + self.rows = { + "drafts": [ + { + "id": 2, + "work_id": 8, + "source_type": "upgrade_book", + "confidence": Decimal("0.8200"), + "draft_payload": {"名称": "乙"}, + "create_time": dt.datetime(2026, 7, 21, 9, 1, 2, 3000), + "deleted": True, + "tenant_id": 1, + }, + { + "id": 1, + "work_id": 8, + "source_type": "upgrade_book", + "confidence": Decimal("0.90"), + "draft_payload": {"名称": "甲"}, + "create_time": dt.datetime(2026, 7, 21, 9, 1, 1), + "deleted": False, + "tenant_id": 1, + }, + ], + "windows": [ + {"id": 12, "work_id": 8, "window_no": 2, "from_chapter": 2, + "to_chapter": 2, "deleted": False, "tenant_id": 1}, + {"id": 11, "work_id": 8, "window_no": 1, "from_chapter": 1, + "to_chapter": 1, "deleted": False, "tenant_id": 1}, + ], + "aliases": [{"id": 21, "work_id": 8, "alias": "阿甲", "tenant_id": 1}], + "presence": [{"id": 31, "work_id": 8, "name": "路人", "tenant_id": 1}], + "card_state": [{"draft_id": 2, "work_id": 8, "watermark_window": 2, + "tenant_id": 1}], + "audits": [{"id": 41, "draft_id": 2, "window_no": 2, "tenant_id": 1}], + "embeddings": [{"id": 51, "draft_id": 2, "embedding": "[0.1,0.20]", + "content_hash": "a" * 64, "tenant_id": 1}], + } + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, traceback): + self.closed = True + + def execute(self, sql, params=None): + normalized = " ".join(sql.split()).lower() + self.executions.append((normalized, params)) + if normalized.startswith("begin") or normalized in {"commit", "rollback"}: + return _Result([]) + if "current_database()" in normalized: + return _Result([{"database": "muse-example", "server_addr": "100.64.0.8", + "server_port": 5433, "server_version": "160000"}]) + if "from muse_content_chapter c" in normalized: + return _Result([ + {"chapter_id": 1, "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": "正文一", "block_revision": 1}, + {"chapter_id": 2, "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}, + ]) + if "from muse_meta_schema s" in normalized: + return _Result([ + {"schema_key": key, "active_version_id": index, + "field_contract_snapshot": {"key": key}} + for index, key in enumerate(sorted(backup.UPGRADE_SCHEMA_KEYS), start=1) + ]) + if "from muse_content_work" in normalized: + return _Result([{"title": "离线测试书"}]) + table_to_domain = { + "from muse_knowledge_draft d": "drafts", + "from example_upgrade_window w": "windows", + "from example_upgrade_alias a": "aliases", + "from example_upgrade_presence p": "presence", + "from example_upgrade_card_state s": "card_state", + "from example_upgrade_audit a": "audits", + "from example_knowledge_embedding e": "embeddings", + } + for marker, domain in table_to_domain.items(): + if marker in normalized: + return _Result(self.rows[domain]) + raise AssertionError(f"未覆盖的离线 SQL：{normalized}") + + +class _RestoreConnection: + """记录恢复事务 SQL；查询默认返回空集合，便于只验证恢复边界。""" + + def __init__(self, sequence_last=100, table_max=50, extra_window=False): + self.closed = False + self.executions = [] + self.sequence_last = sequence_last + self.table_max = table_max + self.extra_window = extra_window + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, traceback): + self.closed = True + + def execute(self, sql, params=None): + normalized = " ".join(sql.split()).lower() + self.executions.append((normalized, params)) + if "from pg_sequences" in normalized: + return _Result([{"last_value": self.sequence_last}]) + if "select max(id) as max_id" in normalized: + return _Result([{"max_id": self.table_max}]) + if "pg_get_serial_sequence" in normalized: + table = params[0] if params else "unknown" + return _Result([{"sequence_name": f"public.{table}_id_seq"}]) + if "from example_upgrade_window w" in normalized and "not (w.id = any" in normalized: + return _Result([{"id": 999}] if self.extra_window else []) + return _Result([]) + + +def _new_backup_dir(): + """返回 /private/tmp 下尚不存在的测试目录。""" + + return pathlib.Path("/private/tmp") / f"backup-upgrade-offline-{uuid.uuid4()}" + + +def _code_identity(): + """离线测试固定代码身份，避免读取工作区状态影响摘要。""" + + return { + "gitCommit": "0123456789abcdef", + "gitHeadCoversCodeFiles": False, + "codeFiles": { + "backup_upgrade_work.py": "b" * 64, + "parse_upgrade.py": "u" * 64, + "parse_llm.py": "p" * 64, + "embed_drafts.py": "e" * 64, + "upgrade_work_lock.py": "k" * 64, + "llm.py": "l" * 64, + "parse-book/SKILL.md": "s" * 64, + }, + "headMatches": {}, + } + + +@contextmanager +def _no_lock(*_args, **_kwargs): + """离线测试锁替身，不连接 PostgreSQL。""" + + yield (1, 2) + + +def _check(name, condition): + """沿用 parse-book 离线测试的简明输出格式。""" + + if not condition: + raise AssertionError(name) + print(f"[PASS] {name}") + + +def test_stable_sort_and_json_types(): + """相同主键集合不受查询返回顺序影响，特殊数据库类型可稳定 JSON 化。""" + + rows = [ + {"id": 10, "at": dt.datetime(2026, 7, 21, 8, 0), "score": Decimal("1.2300")}, + {"id": 2, "at": dt.date(2026, 7, 20), "score": Decimal("0.10")}, + ] + ordered = backup.sort_rows("drafts", rows) + _check("排序稳定-按域主键升序", [row["id"] for row in ordered] == [2, 10]) + encoded = json.loads(backup.canonical_json(ordered[1])) + _check("JSON类型-datetime转ISO字符串", encoded["at"] == "2026-07-21T08:00:00") + _check("JSON类型-Decimal保留精度字符串", encoded["score"] == "1.2300") + vector = json.loads(backup.canonical_json({"embedding": [Decimal("0.10"), Decimal("-2.0")]})) + _check("JSON类型-vector元素稳定序列化", vector["embedding"] == ["0.10", "-2.0"]) + + +def test_path_guards(): + """输出只能是 /private/tmp 下的新目录。""" + + for invalid in ("/tmp/not-explicit", "/private/tmp", "/private/tmp/../etc/backup"): + try: + backup.validate_new_output_dir(pathlib.Path(invalid)) + except ValueError: + pass + else: + raise AssertionError(f"路径越界未拒绝：{invalid}") + _check("路径越界-全部拒绝", True) + + existing = _new_backup_dir() + existing.mkdir(mode=0o700) + try: + try: + backup.validate_new_output_dir(existing) + except FileExistsError: + _check("重复目录-拒绝覆盖", True) + else: + raise AssertionError("已存在目录未拒绝") + finally: + existing.rmdir() + + +def test_backup_transaction_queries_and_round_trip(): + """七域在单个只读一致性事务中读取，关库后再从磁盘复验。""" + + output = _new_backup_dir() + conn = _OfflineConnection() + original_verify = backup.verify_backup + + def verify_after_close(path): + if not conn.closed: + raise AssertionError("数据库连接尚未关闭就开始磁盘复验") + return original_verify(path) + + try: + with mock.patch.object(backup, "verify_backup", side_effect=verify_after_close): + manifest = backup.backup_work( + work_id=8, + output_dir=output, + tenant=1, + dsn="postgresql://user:secret@example.invalid/db", + connect=lambda *_args, **_kwargs: conn, + git_commit="0123456789abcdef", + code_identity=_code_identity(), + expected_chapters=2, + expected_windows=2, + lock_context=_no_lock, + ) + + _check("事务-连接关闭后复验", conn.closed) + statements = [sql for sql, _params in conn.executions] + _check("事务-仅一个REPEATABLE READ READ ONLY", statements.count( + "begin isolation level repeatable read read only") == 1) + _check("事务-正常提交一次", statements.count("commit") == 1) + _check("事务-无数据库写语句", not any( + sql.startswith(("insert ", "update ", "delete ", "alter ", "drop ")) + for sql in statements + )) + + _check("七域-文件与清单齐全", set(manifest["domains"]) == set(backup.DOMAIN_SPECS)) + # 作品标题一条 SELECT，加七域各一条 SELECT；draft 子查询不应被误算成独立域查询。 + _check("七域与input-各查询只执行一次", + sum(sql.startswith("select") for sql in statements) == 11) + sql_by_domain = { + name: next(sql for sql in statements if f"from {spec.table} " in sql) + for name, spec in backup.DOMAIN_SPECS.items() + } + _check("查询边界-drafts含软删", "d.deleted" not in sql_by_domain["drafts"]) + _check("查询边界-windows仅active", "w.deleted = false" in sql_by_domain["windows"]) + _check("查询边界-audits指向目标draft", "a.draft_id in" in sql_by_domain["audits"] + and "source_type" in sql_by_domain["audits"]) + _check("查询边界-embeddings指向目标draft", "e.draft_id in" in sql_by_domain["embeddings"] + and "source_type" in sql_by_domain["embeddings"]) + _check("清单-身份与边界摘要齐全", manifest["work"] == 8 + and manifest["tenant"] == 1 + and manifest["title"] == "离线测试书" + and len(manifest["windowBoundarySha"]) == 64 + and len(manifest["inputSha"]) == 64) + _check("input-同快照绑定Canonical窗合同代码与计划", + manifest["input"]["canonicalContent"]["chapterCount"] == 2 + and manifest["input"]["windowBoundary"]["count"] == 2 + and manifest["input"]["activeFieldContracts"]["count"] == 7 + and manifest["input"]["codeFiles"]["llm.py"] == "l" * 64 + and manifest["input"]["plan"] == { + "model": "MiniMax-M3", "semanticDedup": True}) + _check("清单-不泄露数据库密码", "secret" not in json.dumps(manifest, ensure_ascii=False)) + _check("排序稳定-JSONL按主键", [ + json.loads(line)["id"] + for line in (output / "drafts.jsonl").read_text(encoding="utf-8").splitlines() + ] == [1, 2]) + _check("原子写-无临时文件残留", not list(output.glob("*.tmp"))) + _check("离线复验-七域校验通过", original_verify(output)["backup_id"] == manifest["backup_id"]) + finally: + shutil.rmtree(output, ignore_errors=True) + + +def test_accidental_corruption_detection(): + """任一 JSONL 意外损坏后离线 verify 必须失败；不模拟恶意重写摘要。""" + + output = _new_backup_dir() + conn = _OfflineConnection() + try: + backup.backup_work( + work_id=8, + output_dir=output, + tenant=1, + dsn="unused", + connect=lambda *_args, **_kwargs: conn, + git_commit="fedcba9876543210", + code_identity=_code_identity(), + expected_chapters=2, + expected_windows=2, + lock_context=_no_lock, + ) + with (output / "aliases.jsonl").open("ab") as stream: + stream.write(b'{"id":999}\n') + try: + backup.verify_backup(output) + except backup.VerificationError: + _check("意外损坏检测-文件变化被拒绝", True) + else: + raise AssertionError("意外损坏后的备份仍通过 verify") + finally: + shutil.rmtree(output, ignore_errors=True) + + +def test_dict_row_and_tuple_rejection(): + """真实连接必须请求 dict_row，tuple 行不得被误转成 dict。""" + + with mock.patch("psycopg.connect") as connect: + backup._default_connect("postgresql://example.invalid/db", autocommit=True) + from psycopg.rows import dict_row + _check("连接-dict_row", connect.call_args.kwargs["row_factory"] is dict_row) + + try: + backup.sort_rows("drafts", [(1, 8)]) + except backup.BackupError: + _check("真实rows-tuple拒绝", True) + else: + raise AssertionError("tuple 数据行被错误接受") + + +def _restore_manifest(): + """构造恢复 SQL 测试所需的最小已验证 manifest。""" + + input_summary = { + "databaseIdentity": {"database": "muse-example", "serverAddr": "100.64.0.8", + "serverPort": 5433, "serverVersion": "160000"}, + "canonicalContent": {"chapterCount": 594, "blockCount": 594, "sha": "c" * 64}, + "windowBoundary": {"count": 116, "sha": "w" * 64}, + "activeFieldContracts": {"count": 7, "sha": "f" * 64}, + "codeFiles": { + "backup_upgrade_work.py": "b" * 64, + "parse_upgrade.py": "u" * 64, + "parse_llm.py": "p" * 64, + "embed_drafts.py": "e" * 64, + "upgrade_work_lock.py": "k" * 64, + "llm.py": "l" * 64, + "parse-book/SKILL.md": "s" * 64, + }, + "plan": {"model": "MiniMax-M3", "semanticDedup": True}, + "expectedChapters": 594, + "expectedWindows": 116, + "sourceType": "upgrade_book", + "tenant": 1, + "work": 8, + } + manifest = { + "backup_id": "11111111-1111-4111-8111-111111111111", + "work": 8, + "tenant": 1, + "input": input_summary, + "inputSha": backup._sha256(backup.canonical_json(input_summary).encode("utf-8")), + } + confirmation = { + "backup_id": manifest["backup_id"], + "codeFiles": input_summary["codeFiles"], + "inputSha": manifest["inputSha"], + "tenant": manifest["tenant"], + "work": manifest["work"], + } + manifest["confirmationSha"] = backup._sha256( + backup.canonical_json(confirmation).encode("utf-8") + ) + return manifest + + +def _restore_rows(): + """七域各给一行，驱动 upsert 与清理顺序测试。""" + + return { + "drafts": [{"id": 1, "work_id": 8, "source_type": "upgrade_book", + "draft_payload": {"名称": "甲"}, "deleted": False, "tenant_id": 1}], + "windows": [{"id": 11, "work_id": 8, "window_no": 1, "from_chapter": 1, + "to_chapter": 5, "deleted": False, "tenant_id": 1}], + "aliases": [{"id": 21, "work_id": 8, "canonical_name": "甲", "alias": "阿甲", + "tenant_id": 1}], + "presence": [{"id": 31, "work_id": 8, "window_no": 1, "chapter_no": 1, + "entity_type": "character", "name": "甲", "tenant_id": 1}], + "card_state": [{"draft_id": 1, "work_id": 8, "watermark_window": 1, + "tenant_id": 1}], + "audits": [{"id": 41, "draft_id": 1, "window_no": 1, "field_name": "名称", + "tenant_id": 1}], + "embeddings": [{"id": 51, "draft_id": 1, "content_hash": "a" * 64, + "embed_text": "甲", "model": "Qwen", "dimensions": 2, + "embedding": "[0.1,0.2]", "tenant_id": 1}], + } + + +def test_input_identity_and_drift_guard(): + """input 必须绑定完整身份，任一 Canonical 输入漂移均拒绝恢复。""" + + manifest = _restore_manifest() + expected = manifest["input"] + _check("input-模型与语义判重绑定", expected["plan"] == { + "model": "MiniMax-M3", "semanticDedup": True}) + _check("input-594章与116窗绑定", expected["canonicalContent"]["chapterCount"] == 594 + and expected["windowBoundary"]["count"] == 116) + changed = json.loads(json.dumps(expected)) + changed["canonicalContent"]["sha"] = "d" * 64 + try: + backup.assert_restore_input(manifest, changed, work_id=8, tenant=1) + except backup.BackupError: + _check("input漂移-拒绝恢复", True) + else: + raise AssertionError("Canonical input 漂移未拒绝") + wrong_database = json.loads(json.dumps(expected)) + wrong_database["databaseIdentity"]["database"] = "other-db" + try: + backup.assert_restore_input(manifest, wrong_database, work_id=8, tenant=1) + except backup.BackupError: + _check("数据库identity漂移-拒绝恢复", True) + else: + raise AssertionError("数据库 identity 不符未拒绝") + for work_id, tenant in ((9, 1), (8, 2)): + try: + backup.assert_restore_input(manifest, expected, work_id=work_id, tenant=tenant) + except backup.BackupError: + pass + else: + raise AssertionError("work/tenant 不符未拒绝") + _check("恢复范围-work与tenant不符拒绝", True) + + +def test_restore_sql_boundaries_and_rehearse_rollback(): + """共享恢复内核按七域边界操作，演练无条件回滚。""" + + manifest = _restore_manifest() + rows = _restore_rows() + conn = _RestoreConnection() + with mock.patch.object(backup, "verify_backup", return_value=manifest), \ + mock.patch.object(backup, "read_backup_rows", return_value=rows), \ + mock.patch.object(backup, "capture_input_snapshot", return_value=manifest["input"]), \ + mock.patch.object(backup, "verify_restored_domains"): + backup.restore_backup( + pathlib.Path("/private/tmp/offline-restore"), work_id=8, tenant=1, + dsn="unused", connect=lambda *_args, **_kwargs: conn, execute=False, + confirmation_sha=manifest["confirmationSha"], lock_context=_no_lock, + ) + statements = [sql for sql, _params in conn.executions] + _check("rehearse-强制ROLLBACK", statements[-1] == "rollback" and "commit" not in statements) + _check("exact恢复-事务局部关闭更新时间触发器", + "set local session_replication_role to replica" in statements) + _check("恢复边界-额外卡软删", any( + sql.startswith("update muse_knowledge_draft") and "not (d.id = any" in sql + for sql in statements + )) + _check("恢复边界-额外向量软删", any( + sql.startswith("update example_knowledge_embedding") and "not (e.id = any" in sql + for sql in statements + )) + for table in ("example_upgrade_alias", "example_upgrade_presence", + "example_upgrade_card_state", "example_upgrade_audit"): + if not any(sql.startswith(f"delete from {table}") for sql in statements): + raise AssertionError(f"恢复前未清理 {table}") + _check("恢复边界-四域先清再插", True) + _check("恢复边界-windows拒绝额外active", any( + "from example_upgrade_window" in sql and "not (w.id = any" in sql + for sql in statements + )) + _check("恢复边界-七域均有upsert", sum("on conflict" in sql for sql in statements) == 7) + + rejected = _RestoreConnection(extra_window=True) + try: + backup.apply_restore_kernel(rejected, manifest, rows) + except backup.BackupError: + _check("恢复边界-实际拒绝额外active窗且写前失败", not any( + sql.startswith(("insert ", "update ", "delete ")) + for sql, _params in rejected.executions + )) + else: + raise AssertionError("额外 active 窗未阻断恢复") + + +def test_restore_execute_confirmation_and_sequence_forward_only(): + """execute 必须双确认；sequence 当前值更大时不得 setval 回退。""" + + manifest = _restore_manifest() + for supplied in (None, "22222222-2222-4222-8222-222222222222"): + try: + backup.validate_execute_confirmation( + True, supplied, manifest["backup_id"], + manifest["confirmationSha"], manifest["confirmationSha"], + ) + except backup.BackupError: + pass + else: + raise AssertionError("execute 缺少匹配 backup_id 仍被允许") + backup.validate_execute_confirmation( + True, manifest["backup_id"], manifest["backup_id"], + manifest["confirmationSha"], manifest["confirmationSha"], + ) + try: + backup.validate_execute_confirmation( + False, None, manifest["backup_id"], "wrong", manifest["confirmationSha"] + ) + except backup.BackupError: + _check("confirmation确认-rehearse也必须精确匹配", True) + else: + raise AssertionError("rehearse confirmationSha 不匹配仍被允许") + _check("execute确认-必须显式flag与匹配backup_id", True) + + conn = _RestoreConnection(sequence_last=100, table_max=50) + backup.advance_sequences_forward(conn) + _check("序列-不回退", not any("setval" in sql for sql, _params in conn.executions)) + + forward = _RestoreConnection(sequence_last=100, table_max=150) + backup.advance_sequences_forward(forward) + _check("序列-仅在需要时向前", sum( + sql.startswith("select setval") for sql, _params in forward.executions + ) == len(backup.SEQUENCE_TABLES)) + + execute_conn = _RestoreConnection() + with mock.patch.object(backup, "verify_backup", return_value=manifest), \ + mock.patch.object(backup, "read_backup_rows", return_value=_restore_rows()), \ + mock.patch.object(backup, "capture_input_snapshot", return_value=manifest["input"]), \ + mock.patch.object(backup, "verify_restored_domains"), \ + mock.patch.object(backup, "advance_sequences_forward"): + backup.restore_backup( + pathlib.Path("/private/tmp/offline-restore"), work_id=8, tenant=1, + dsn="unused", connect=lambda *_args, **_kwargs: execute_conn, execute=True, + backup_id=manifest["backup_id"], code_identity=_code_identity(), + confirmation_sha=manifest["confirmationSha"], lock_context=_no_lock, + ) + execute_statements = [sql for sql, _params in execute_conn.executions] + _check("execute确认-匹配后才COMMIT", execute_statements[-1] == "commit") + + +def test_expected_scope_code_dependencies_and_lock_fail_closed(): + """期望范围、依赖代码与同书锁均进入机械门。""" + + chapters = [ + {"chapter_id": 1, "chapter_order": 1, "block_id": 101, "content_text": "甲"}, + {"chapter_id": 2, "chapter_order": 2, "block_id": 102, "content_text": "乙"}, + ] + windows = [ + {"id": 11, "window_no": 1, "from_chapter": 1, "to_chapter": 1}, + {"id": 12, "window_no": 2, "from_chapter": 2, "to_chapter": 2}, + ] + backup.validate_expected_scope(chapters, windows, expected_chapters=2, expected_windows=2) + _check("expected-章节窗口连续且每章非空", True) + for bad_chapters, bad_windows in ( + (chapters[:1], windows), + ([dict(chapters[0], content_text=""), chapters[1]], windows), + (chapters, [windows[0], dict(windows[1], window_no=3)]), + (chapters, [windows[0], dict(windows[1], from_chapter=3)]), + ): + try: + backup.validate_expected_scope( + bad_chapters, bad_windows, expected_chapters=2, expected_windows=2 + ) + except backup.BackupError: + pass + else: + raise AssertionError("expected 范围异常未拒绝") + _check("expected-数量非空与连续性异常均拒绝", True) + + identity = backup.capture_code_identity("HEAD") + _check("代码身份-补齐实际依赖SHA", { + "parse_llm.py", "embed_drafts.py", "upgrade_work_lock.py" + }.issubset(identity["codeFiles"])) + + business_calls = {"count": 0} + + @contextmanager + def unavailable_lock(*_args, **_kwargs): + raise backup.UpgradeWorkLockUnavailable(1, 8) + yield + + def forbidden_connect(*_args, **_kwargs): + business_calls["count"] += 1 + raise AssertionError("锁失败后仍创建业务连接") + + output = _new_backup_dir() + try: + backup.backup_work( + 8, output, tenant=1, dsn="unused", connect=forbidden_connect, + code_identity=_code_identity(), expected_chapters=594, expected_windows=116, + lock_context=unavailable_lock, + ) + except backup.UpgradeWorkLockUnavailable: + _check("advisory-lock-失败时零业务SQL", business_calls["count"] == 0) + else: + raise AssertionError("同书锁失败未阻断 backup") + + with mock.patch.object(backup, "verify_backup") as offline_verify: + try: + backup.restore_backup( + pathlib.Path("/private/tmp/not-read"), work_id=8, tenant=1, dsn="unused", + connect=forbidden_connect, confirmation_sha="unused", + lock_context=unavailable_lock, + ) + except backup.UpgradeWorkLockUnavailable: + _check("advisory-lock-恢复演练也在离线verify前阻断", not offline_verify.called) + else: + raise AssertionError("同书锁失败未阻断 restore/rehearse") + + +def main(): + """顺序运行全部离线测试，便于直接以 Python 执行。""" + + test_stable_sort_and_json_types() + test_path_guards() + test_backup_transaction_queries_and_round_trip() + test_accidental_corruption_detection() + test_dict_row_and_tuple_rejection() + test_input_identity_and_drift_guard() + test_restore_sql_boundaries_and_rehearse_rollback() + test_restore_execute_confirmation_and_sequence_forward_only() + test_expected_scope_code_dependencies_and_lock_fail_closed() + print("\n全部离线自测通过：49 项（未连库、未发任何 LLM 调用）") + + +if __name__ == "__main__": + main() diff --git a/.claude/skills/parse-book/scripts/test_parse_upgrade_offline.py b/.claude/skills/parse-book/scripts/test_parse_upgrade_offline.py index fe3812b..f5420a6 100644 --- a/.claude/skills/parse-book/scripts/test_parse_upgrade_offline.py +++ b/.claude/skills/parse-book/scripts/test_parse_upgrade_offline.py @@ -16,6 +16,7 @@ import json import pathlib import sys +from contextlib import nullcontext from copy import deepcopy from unittest.mock import patch @@ -628,7 +629,8 @@ def test_run_redo_preflight_rejects_unsafe_targets(): text = "\n".join(str(chapter_texts.get(chapter, "")) for chapter in range(a, b + 1)) return text, chapter_texts - with patch.object(pu.psycopg, "connect", side_effect=fake_connect), \ + with patch.object(pu, "upgrade_work_lock", return_value=nullcontext()), \ + patch.object(pu.psycopg, "connect", side_effect=fake_connect), \ patch.object(pu, "load_entity_contracts", return_value=_fake_contracts()), \ patch.object(pu, "load_window_material", side_effect=fake_material), \ patch.object(pu, "_snapshot_redo_window", side_effect=lambda *args: snapshot_calls.append(args)), \ @@ -714,7 +716,7 @@ def _run_with_failures(*, max_calls, failures): patch.object(pu, "m3_json", side_effect=fake_m3_json), \ patch.object(pu, "_snapshot_redo_window", return_value=snapshot), \ patch.object(pu, "_restore_redo_window", side_effect=fake_restore): - pu.run.callback( + pu._run( work_id=8, max_windows=0, max_calls=max_calls, @@ -850,7 +852,7 @@ def test_run_presence_paths_accept_numeric_strings(): return_value=({"安若雪": (701, "character", "旧摘要")}, {}, {}), ), \ patch.object(pu, "m3_json", side_effect=fake_m3_json): - pu.run.callback( + pu._run( work_id=8, max_windows=0, max_calls=0, diff --git a/.claude/skills/parse-book/scripts/test_reset_upgrade_work_offline.py b/.claude/skills/parse-book/scripts/test_reset_upgrade_work_offline.py new file mode 100644 index 0000000..f96e414 --- /dev/null +++ b/.claude/skills/parse-book/scripts/test_reset_upgrade_work_offline.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python3 +"""升格命令锁接线的纯离线测试：锁失败时禁止任何业务调用。""" + +import contextlib +import pathlib +import sys +import unittest +from unittest.mock import Mock, patch + +from click.testing import CliRunner + + +SCRIPT_DIR = pathlib.Path(__file__).resolve().parent +sys.path.insert(0, str(SCRIPT_DIR)) + +import parse_upgrade as parse # noqa: E402 +import reset_upgrade_work as reset # noqa: E402 +from upgrade_work_lock import UpgradeWorkLockUnavailable # noqa: E402 + + +def _lock_failure(*_args, **_kwargs): + """模拟同书已有命令持锁。""" + + raise UpgradeWorkLockUnavailable(tenant_id=1, work_id=8) + + +class UpgradeCommandLockOfflineTest(unittest.TestCase): + """验证锁位于全部业务 SQL、写入和模型调用之前。""" + + def setUp(self): + self.runner = CliRunner() + + def test_parse_run_lock_failure_skips_business(self): + with patch.object(parse, "upgrade_work_lock", side_effect=_lock_failure) as lock, \ + patch.object(parse, "_run") as business: + result = self.runner.invoke(parse.cli, ["run", "--work-id", "8"]) + + self.assertNotEqual(result.exit_code, 0) + self.assertIn("已有升格命令正在处理", result.output) + lock.assert_called_once_with(parse.DSN, parse.TENANT, 8) + business.assert_not_called() + + def test_parse_windows_lock_failure_skips_database(self): + with patch.object(parse, "upgrade_work_lock", side_effect=_lock_failure) as lock, \ + patch.object(parse.psycopg, "connect") as connect: + result = self.runner.invoke(parse.cli, ["windows", "--work-id", "8"]) + + self.assertNotEqual(result.exit_code, 0) + self.assertIn("已有升格命令正在处理", result.output) + lock.assert_called_once_with(parse.DSN, parse.TENANT, 8) + connect.assert_not_called() + + def test_parse_status_does_not_request_lock(self): + lock = Mock(side_effect=AssertionError("status 不应请求锁")) + with patch.object(parse, "upgrade_work_lock", lock), \ + patch.object(parse.psycopg, "connect", side_effect=RuntimeError("离线中止")): + result = self.runner.invoke(parse.cli, ["status", "--work-id", "8"]) + + self.assertNotEqual(result.exit_code, 0) + lock.assert_not_called() + + def test_reset_preview_lock_failure_skips_database(self): + self._assert_reset_lock_failure_skips_database(execute=False) + + def test_reset_execute_lock_failure_skips_database(self): + self._assert_reset_lock_failure_skips_database(execute=True) + + def _assert_reset_lock_failure_skips_database(self, *, execute): + args = ["--work-id", "8"] + (["--execute"] if execute else []) + with patch.object(reset, "upgrade_work_lock", side_effect=_lock_failure) as lock, \ + patch.object(reset.psycopg, "connect") as connect: + result = self.runner.invoke(reset.main, args) + + self.assertNotEqual(result.exit_code, 0) + self.assertIn("已有升格命令正在处理", result.output) + lock.assert_called_once_with(reset.DSN, reset.TENANT, 8) + connect.assert_not_called() + + def test_lock_success_wraps_parse_run_business(self): + events = [] + + @contextlib.contextmanager + def fake_lock(*_args, **_kwargs): + events.append("lock-enter") + try: + yield + finally: + events.append("lock-exit") + + def fake_business(*_args, **_kwargs): + events.append("business") + + with patch.object(parse, "upgrade_work_lock", fake_lock), \ + patch.object(parse, "_run", side_effect=fake_business): + result = self.runner.invoke(parse.cli, ["run", "--work-id", "8"]) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(events, ["lock-enter", "business", "lock-exit"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/.claude/skills/parse-book/scripts/test_upgrade_work_lock_offline.py b/.claude/skills/parse-book/scripts/test_upgrade_work_lock_offline.py new file mode 100644 index 0000000..90f2004 --- /dev/null +++ b/.claude/skills/parse-book/scripts/test_upgrade_work_lock_offline.py @@ -0,0 +1,133 @@ +#!/usr/bin/env python3 +"""升格作品会话锁的纯离线测试：不连接数据库，不调用模型。""" + +import pathlib +import sys +import unittest + + +SCRIPT_DIR = pathlib.Path(__file__).resolve().parent +sys.path.insert(0, str(SCRIPT_DIR)) + +from upgrade_work_lock import ( # noqa: E402 + UpgradeWorkLockUnavailable, + advisory_lock_keys, + upgrade_work_lock, +) + + +class _Result: + """模拟 psycopg 查询结果。""" + + def __init__(self, value): + self.value = value + + def fetchone(self): + return (self.value,) + + +class _LockServer: + """在内存中模拟 PostgreSQL 会话级 advisory lock。""" + + def __init__(self): + self.held = {} + self.connections = [] + + def connect(self, dsn, *, autocommit): + connection = _LockConnection(self, dsn, autocommit) + self.connections.append(connection) + return connection + + +class _LockConnection: + """每个实例代表一条独立数据库会话，只接受锁相关 SQL。""" + + def __init__(self, server, dsn, autocommit): + self.server = server + self.dsn = dsn + self.autocommit = autocommit + self.owned = set() + self.statements = [] + self.closed = False + + def execute(self, sql, params): + normalized = " ".join(sql.split()) + key = tuple(params) + self.statements.append((normalized, key)) + if normalized == "SELECT pg_try_advisory_lock(%s, %s)": + owner = self.server.held.get(key) + if owner is not None and owner is not self: + return _Result(False) + self.server.held[key] = self + self.owned.add(key) + return _Result(True) + if normalized == "SELECT pg_advisory_unlock(%s, %s)": + owned = key in self.owned + if owned: + self.owned.remove(key) + self.server.held.pop(key, None) + return _Result(owned) + raise AssertionError(f"锁连接禁止执行业务 SQL：{normalized}") + + def close(self): + for key in list(self.owned): + self.server.held.pop(key, None) + self.owned.clear() + self.closed = True + + +class UpgradeWorkLockOfflineTest(unittest.TestCase): + """覆盖锁键稳定性、互斥语义和 finally 释放。""" + + def test_keys_are_stable_signed_int32_pair(self): + first = advisory_lock_keys(tenant_id=1, work_id=8) + second = advisory_lock_keys(tenant_id=1, work_id=8) + + self.assertEqual(first, second) + self.assertEqual(len(first), 2) + self.assertTrue(all(-(2 ** 31) <= value < 2 ** 31 for value in first)) + self.assertNotEqual(first, advisory_lock_keys(tenant_id=1, work_id=9)) + self.assertNotEqual(first, advisory_lock_keys(tenant_id=2, work_id=8)) + + def test_same_work_concurrent_session_is_rejected(self): + server = _LockServer() + + with upgrade_work_lock("postgresql://offline", 1, 8, connect=server.connect): + with self.assertRaises(UpgradeWorkLockUnavailable): + with upgrade_work_lock("postgresql://offline", 1, 8, connect=server.connect): + self.fail("同书第二条会话不得进入受保护区") + + self.assertEqual(server.held, {}) + self.assertTrue(all(connection.closed for connection in server.connections)) + + def test_exception_releases_lock_and_closes_autocommit_session(self): + server = _LockServer() + + with self.assertRaisesRegex(RuntimeError, "业务失败"): + with upgrade_work_lock("postgresql://offline", 1, 8, connect=server.connect): + raise RuntimeError("业务失败") + + self.assertEqual(server.held, {}) + self.assertEqual(len(server.connections), 1) + connection = server.connections[0] + self.assertTrue(connection.autocommit) + self.assertTrue(connection.closed) + self.assertEqual( + [sql for sql, _ in connection.statements], + ["SELECT pg_try_advisory_lock(%s, %s)", "SELECT pg_advisory_unlock(%s, %s)"], + ) + + def test_different_work_does_not_conflict(self): + server = _LockServer() + entered = [] + + with upgrade_work_lock("postgresql://offline", 1, 8, connect=server.connect): + with upgrade_work_lock("postgresql://offline", 1, 9, connect=server.connect): + entered.append(True) + + self.assertEqual(entered, [True]) + self.assertEqual(server.held, {}) + + +if __name__ == "__main__": + unittest.main() diff --git a/.claude/skills/parse-book/scripts/upgrade_work_lock.py b/.claude/skills/parse-book/scripts/upgrade_work_lock.py new file mode 100644 index 0000000..3e1064c --- /dev/null +++ b/.claude/skills/parse-book/scripts/upgrade_work_lock.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python3 +"""升格作品命令共用的 PostgreSQL 会话级 advisory lock。 + +锁连接是独立 autocommit 会话，只执行加锁和解锁 SQL，不参与任何业务事务。调用方可让该 +context manager 跨越 LLM/嵌入调用持续存活；退出受保护区时，无论成功或异常都会显式解锁并 +关闭会话。backup 的 restore/rehearse 可直接导入 ``upgrade_work_lock`` 复用同一互斥边界。 +""" + +import hashlib +import struct +from contextlib import contextmanager + +import psycopg + + +LOCK_NAMESPACE = b"muse:upgrade-work-lock:v1" + + +class UpgradeWorkLockUnavailable(RuntimeError): + """同租户、同作品已有另一条升格命令持锁。""" + + def __init__(self, tenant_id, work_id): + self.tenant_id = tenant_id + self.work_id = work_id + super().__init__( + f"作品 tenant={tenant_id} work={work_id} 已有升格命令正在处理；" + "本次未执行任何业务操作" + ) + + +def advisory_lock_keys(tenant_id, work_id): + """把租户与作品稳定映射为 PostgreSQL 接受的两个有符号 int32 锁键。""" + + identity = f"{int(tenant_id)}:{int(work_id)}".encode("ascii") + digest = hashlib.sha256(LOCK_NAMESPACE + b"\0" + identity).digest() + return struct.unpack(">ii", digest[:8]) + + +@contextmanager +def upgrade_work_lock(dsn, tenant_id, work_id, *, connect=None): + """非阻塞获取同书会话锁，失败立即抛错，finally 中释放并关闭专用连接。 + + ``connect`` 仅供离线测试注入；生产默认使用 ``psycopg.connect``。返回的两个锁键可用于 + 调试记录，调用方不得使用该专用连接执行任何业务 SQL。 + """ + + connector = connect or psycopg.connect + keys = advisory_lock_keys(tenant_id, work_id) + connection = connector(dsn, autocommit=True) + acquired = False + try: + row = connection.execute( + "SELECT pg_try_advisory_lock(%s, %s)", keys + ).fetchone() + acquired = bool(row and row[0] is True) + if not acquired: + raise UpgradeWorkLockUnavailable(tenant_id, work_id) + yield keys + finally: + try: + if acquired: + connection.execute("SELECT pg_advisory_unlock(%s, %s)", keys) + finally: + # 即使显式解锁 SQL 失败，关闭会话也会由 PostgreSQL 释放会话级锁。 + connection.close()