test(backup): 留档回填演练脚本 + 实测通过

scripts/restore_drill.py：把留档 PG 行按原文 \copy FROM 回填进目标库，核对每张表行数与清单、
抽样页面正文/标题逐字段比对，再用临时跟踪表 _drill_restored 精确删除并断言残留为 0。

实测（本地测试库 weknora_full，留档 20261009T041913Z-0b217e8a）：13,539 行全部回填成功、
8 张表行数与清单逐一致、冲突 0、抽样页正文一致（388 字符）、残留 0。

同时记录一个格式事实：留档 JSONL 是 psql \copy text 格式（转义），按行 json.loads 会失败
（knowledges 3830/3830、wiki_pages 218/1982），回填/程序化读取须先按 text 格式反转义。

docs/P6_RELEASE_WORKER_ACCEPTANCE.md

@@ -334,6 +334,10 @@ INCLUDE_CHUNKS=0 bash scripts/target_kb_snapshot.sh   # 不导 chunks（快照
   （"可恢复"的最低证明）。只留 PG 行、或 ES 计数为 0 的留档一律拒绝。任一条不满足——
   **包括门禁本身未接线**——都以 `release_target_snapshot_missing`（不可重试）拒绝发布，
   且不触碰目标库一行。页面把该码显示为"缺少目标知识库的发布前快照留档，已拒绝发布"。
+- 回填演练：`python3 scripts/restore_drill.py --snapshot-dir <留档目录> --target-url <本地测试库>`
+  实测把留档 **13,539 行**按原文回填进测试库（8 张表行数与清单逐一致、冲突 0、抽样页面正文与标题
+  逐字段相同），核对后精确删除、残留 0。注意留档 JSONL 是 psql text 格式（`\copy` 转义），
+  回填要用对称的 `\copy FROM`，按行 `json.loads` 会失败。
 - 云内执行：`bash scripts/target_kb_snapshot.sh` 默认在与 DB 同 VPC 的云主机上跑
   （凭据取自已部署的 WeKnora 容器，不经家用带宽），PG 8 张表 **44 秒**；ES 导出走云主机直连
   ES 容器（只读 scroll），两个库共 **131,122 条**文档、gzip 后 **692 MB**；全流程实测

docs/PRODUCTION_BACKUP_RECOVERY.md

@@ -59,13 +59,31 @@ docker run --rm --network host postgres:17-alpine psql "$TARGET_DB_URL" \
 
 | 事故 | 首选 | 耗时 |
 |---|---|---|
-| 某个 KB 的 wiki / 元数据被覆盖 | **留档回填**（快照目录 `README.md` 的 SQL：先删按 `_id`/主键冲突行，再 `\copy` 回） | 秒~分钟 |
+| 某个 KB 的 wiki / 元数据被覆盖 | **留档回填** `scripts/restore_drill.py`（psql 按原文 `\copy FROM` 回填；冲突行跳过并计数） | 秒~分钟 |
 | 检索/向量异常、索引误删 | **ES 快照 restore**（`kb_backup`；或按 `es_documents.jsonl.gz` 用 `_id` 重建） | 分钟级 |
 | OSS 对象被覆盖 / 删除 | **OSS 版本控制**回滚单对象或前缀 | 秒~分钟 |
 | 整库被改坏（要带走/异地一份） | **自管逻辑备份** `pg_restore -j4` 到临时库或新实例 | 分钟~十分钟 |
 | 整实例级错误、需要"回到某个时刻" | **RDS 快照 / PITR** → 新实例 → 按 KB 取行 | 10~30 分钟 |
 | 主机/磁盘整体丢失 | ES 仓 + 留档在**同一块盘**上 → 必须靠 RDS 托管备份 + OSS（见"边界"） | 30 分钟+ |
 
+## 回填演练（"可恢复"的实测证明）
+
+哈希校验只能证明文件没坏，不能证明能恢复。`scripts/restore_drill.py` 把留档**真的插回一个库**再核对：
+
+```bash
+python3 scripts/restore_drill.py --snapshot-dir $KB_BACKUP_ROOT/target-kb-snapshots/<UTC>-<kb8> \
+  --target-url "$LOCAL_TEST_DB_URL"          # 用本地测试库；--keep 可保留插入的行
+```
+
+实测（本地测试库 WKNFull `weknora_full`，留档 `20261009T041913Z-0b217e8a`）：
+
+| 表 | 清单行数 | 回填 | 已插入 | 冲突跳过 | 清理后残留 |
+|---|---|---|---|---|---|
+| knowledge_bases / knowledges / wiki_folders / wiki_pages / resources / resource_bindings | 2 / 3830 / 65 / 1982 / 3830 / 3830 | 同左 | 同左 | 0 | 0 |
+
+合计 **13,539 行**回填成功，抽样页面正文与标题逐字段相同；`knowledge_tags` / `knowledge_tag_relations` 为空表。
+演练在目标库临时建一张 `_drill_restored` 跟踪表记录插入了哪些行，核对后**按行精确删除**并断言残留为 0（需要目标库有建表权限，故演练只用测试库）。
+
 ## 每次导入/发布前的标准动作
 
 1. `bash scripts/target_kb_snapshot.sh`（默认云内执行 + ES 字节留云主机，实测 **2 分 52 秒**；
@@ -86,5 +104,8 @@ docker run --rm --network host postgres:17-alpine psql "$TARGET_DB_URL" \
   且 **同一实例上的生产库（`weknora_prod`）也在其中** —— 备份目录权限按 0700/0600 保管。
 - 留档默认不含 `chunks` 行：向量在 ES 里、分块行可由源端重发；要"整库行级可离线重建"才需
   `INCLUDE_CHUNKS=1`（实测本机隧道约 24 分钟、云内 44 秒内）。
+- 留档 JSONL 是 psql `\copy` **text 格式**（反斜杠与控制字符被转义）：回填必须用对称的
+  `\copy FROM`（由 psql 反转义，`restore_drill.py` 就是这么做的）；按行 `json.loads` 会失败
+  （实测 `knowledges` 3830/3830、`wiki_pages` 218/1982 行报错），程序化读取需先反转义。
 - `es_documents.jsonl.gz` 的 sha256 **每次导出都不同**（scroll 顺序非确定），判据是清单里的
   每库文档数与逐条内容；要跨次比对请比 `es_document_counts` 与 `_id` 集合，不要比文件哈希。

scripts/restore_drill.py

@@ -0,0 +1,208 @@
+#!/usr/bin/env python3
+"""留档回填演练：把发布前留档（PG 行级 JSONL）真的插进一个目标库，核对行数与内容，再清理。
+
+为什么需要它：哈希校验只能证明"文件没坏"，不能证明"能恢复"。这个演练完整走一遍
+"psql 按原文回填 → 行数与清单一致 → 抽样逐字段比对 → 精确删除插入的行（残留必须为 0）"。
+
+格式注意：留档 JSONL 是 psql `\\copy` **text 格式**输出（反斜杠与控制字符被转义），
+所以回填必须用对称的 `\\copy FROM`（由 psql 反转义）；按行 `json.loads` 会失败，
+程序化读取需先反转义（见 unescape_text_format）。
+
+用法：
+  python3 restore_drill.py --snapshot-dir <快照目录> --target-url <目标库URL> [--keep] [--slug <页面slug>]
+
+目标库选择：演练用本地测试库（如 WKNFull 的 weknora_full），不要用生产库。
+演练会在目标库创建并用完即删一张跟踪表 `_drill_restored`（记录插入了哪些行，保证清理精确、无残留）。
+"""
+import argparse
+import json
+import pathlib
+import subprocess
+import sys
+
+# 依赖顺序：主表 → 子表；resources 在 resource_bindings 之前
+TABLES = [
+    "knowledge_bases",
+    "knowledges",
+    "knowledge_tags",
+    "wiki_folders",
+    "wiki_pages",
+    "resources",
+    "resource_bindings",
+    "knowledge_tag_relations",
+]
+TRACKING_TABLE = "_drill_restored"
+
+
+def psql(image, snapshot_dir, url, sql):
+    command = [
+        "docker", "run", "--rm", "-i", "--network", "host",
+        "-e", f"PGURL={url}",
+        "-v", f"{snapshot_dir}:/snap:ro",
+        image, "sh", "-c", 'psql "$PGURL" -v ON_ERROR_STOP=1 -f -',
+    ]
+    process = subprocess.run(command, input=sql, text=True, capture_output=True)
+    if process.returncode != 0:
+        sys.stderr.write(process.stdout[-4000:] + "\n" + process.stderr[-4000:])
+        raise SystemExit(f"回填失败（psql 退出码 {process.returncode}）")
+    return process.stdout
+
+
+def build_insert(table):
+    """回填一张表：staging(jsonb 原文) → 目标表（冲突跳过）→ 记录插入了哪些行。"""
+    return f"""WITH ins AS (
+  INSERT INTO {table}
+  SELECT (jsonb_populate_record(NULL::{table}, s.row)).*
+  FROM stage_{table} s
+  ON CONFLICT DO NOTHING
+  RETURNING to_jsonb({table}) AS row_json
+), rec AS (
+  INSERT INTO {TRACKING_TABLE} (table_name, row)
+  SELECT '{table}', ins.row_json FROM ins
+  RETURNING 1
+)
+SELECT 'drill:{table}' AS marker,
+       (SELECT count(*) FROM stage_{table}) AS staged,
+       (SELECT count(*) FROM rec) AS inserted;"""
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--snapshot-dir", required=True)
+    parser.add_argument("--target-url", required=True)
+    parser.add_argument("--pg-image", default="postgres:17-alpine")
+    parser.add_argument("--slug", default="", help="抽样比对的 wiki 页面 slug（默认自动挑一个）")
+    parser.add_argument("--keep", action="store_true", help="保留插入的行（默认核对后精确删除）")
+    arguments = parser.parse_args()
+
+    snapshot = pathlib.Path(arguments.snapshot_dir).resolve()
+    manifest = json.loads((snapshot / "MANIFEST.json").read_text())
+    expected = {item["file"].removesuffix(".jsonl"): item.get("rows")
+                for item in manifest["files"] if item["file"].endswith(".jsonl")}
+    print(f"留档: {snapshot.name} | created_at={manifest['created_at']} "
+          f"| 库: {', '.join(manifest['knowledge_base_ids'])}")
+
+    slug = arguments.slug
+    if not slug:
+        with (snapshot / "wiki_pages.jsonl").open("rb") as handle:
+            for raw in handle:
+                try:
+                    row = json.loads(unescape_text_format(raw))
+                except Exception:
+                    continue
+                if row.get("content"):
+                    slug = row["slug"]
+                    break
+        print(f"抽样页面: {slug or '（留档里没有带正文的页面）'}")
+
+    statements = [
+        "\\set ON_ERROR_STOP on",
+        f"DROP TABLE IF EXISTS {TRACKING_TABLE};",
+        f"CREATE TABLE {TRACKING_TABLE} (table_name text NOT NULL, row jsonb NOT NULL);",
+        "BEGIN;",
+    ]
+    for table in TABLES:
+        if not (snapshot / f"{table}.jsonl").exists():
+            continue
+        statements += [
+            f"CREATE TEMP TABLE stage_{table} (row jsonb);",
+            f"\\copy stage_{table} (row) FROM '/snap/{table}.jsonl'",
+            build_insert(table),
+        ]
+    if slug:
+        statements.append(f"""SELECT 'drill:sample' AS marker,
+  ((s.r).title = t.title) AS title_equal,
+  ((s.r).content = t.content) AS content_equal,
+  length((s.r).content) AS chars
+FROM (SELECT jsonb_populate_record(NULL::wiki_pages, row) AS r FROM stage_wiki_pages
+      WHERE (jsonb_populate_record(NULL::wiki_pages, row)).slug = '{slug}') s
+JOIN wiki_pages t ON t.id = (s.r).id;""")
+    statements.append("COMMIT;")
+
+    if not arguments.keep:
+        statements.append("BEGIN;")
+        for table in reversed(TABLES):
+            if not (snapshot / f"{table}.jsonl").exists():
+                continue
+            statements.append(
+                f"DELETE FROM {table} t USING {TRACKING_TABLE} d "
+                f"WHERE d.table_name = '{table}' AND to_jsonb(t) = d.row;")
+            statements.append(
+                f"SELECT 'drill:cleanup:{table}' AS marker, count(*) AS left "
+                f"FROM {table} t JOIN {TRACKING_TABLE} d ON d.table_name = '{table}' AND to_jsonb(t) = d.row;")
+        statements.append("COMMIT;")
+        statements.append(f"DROP TABLE {TRACKING_TABLE};")
+
+    output = psql(arguments.pg_image, snapshot, arguments.target_url, "\n".join(statements) + "\n")
+
+    staged, inserted, left, sample = {}, {}, {}, {}
+    for line in output.splitlines():
+        parts = [part.strip() for part in line.split("|")]
+        if not parts or not parts[0].startswith("drill:"):
+            continue
+        marker = parts[0].split(":", 1)[1]
+        if marker.startswith("cleanup:"):
+            if len(parts) != 2:
+                raise SystemExit(f"清理结果列数异常（{len(parts)}）: {line}")
+            left[marker.split(":", 1)[1]] = int(parts[1])
+        elif marker == "sample":
+            if len(parts) != 4:
+                raise SystemExit(f"抽样结果列数异常（{len(parts)}）: {line}")
+            sample = {"title_equal": parts[1] == "t", "content_equal": parts[2] == "t", "chars": parts[3]}
+        else:
+            if len(parts) != 3:
+                raise SystemExit(f"回填结果列数异常（{len(parts)}）: {line}")
+            staged[marker] = int(parts[1])
+            inserted[marker] = int(parts[2])
+
+    print(f"\n{'表':<26}{'清单行数':>10}{'回填尝试':>10}{'已插入':>8}{'冲突跳过':>10}{'清理后残留':>12}")
+    bad = []
+    for table in TABLES:
+        if table not in staged:
+            continue
+        want, got, added = expected.get(table), staged[table], inserted.get(table, 0)
+        residue = left.get(table)
+        if want != got:
+            bad.append(f"{table} 行数与清单不符（清单 {want} / 回填 {got}）")
+        if arguments.keep:
+            residue_text = "（保留）"
+        elif residue is None:
+            residue_text = "未验证"
+            bad.append(f"{table} 清理结果未取到，无法证明无残留")
+        else:
+            residue_text = str(residue)
+            if residue:
+                bad.append(f"{table} 清理后仍有 {residue} 行")
+        print(f"{table:<26}{str(want):>10}{got:>10}{added:>8}{got - added:>10}{residue_text:>12}")
+    if not sample:
+        bad.append("抽样页面没有完成比对（结果未取到）")
+    else:
+        print(f"\n抽样比对: 标题{'一致' if sample['title_equal'] else '不一致'} / "
+              f"正文{'一致' if sample['content_equal'] else '不一致'}（{sample['chars']} 字符）")
+        if not sample["content_equal"] or not sample["title_equal"]:
+            bad.append("抽样页面的正文/标题与留档不一致")
+    if bad:
+        raise SystemExit("回填演练未通过: " + "; ".join(bad))
+    print(f"结论: {sum(staged.values())} 行按原文回填成功（与清单一致、抽样逐字段相同）；"
+          + ("插入的行保留在目标库中。" if arguments.keep else "核对后已精确删除，残留为 0。"))
+
+
+def unescape_text_format(raw: bytes) -> bytes:
+    """psql text 格式反转义（\\\\→\\，\\n→换行，\\t→制表符…）。"""
+    out = bytearray()
+    index = 0
+    while index < len(raw):
+        char = raw[index]
+        if char == 0x5C and index + 1 < len(raw):
+            following = raw[index + 1]
+            out.append({0x62: 0x08, 0x66: 0x0C, 0x6E: 0x0A, 0x72: 0x0D,
+                        0x74: 0x09, 0x76: 0x0B}.get(following, following))
+            index += 2
+        else:
+            out.append(char)
+            index += 1
+    return bytes(out)
+
+
+if __name__ == "__main__":
+    main()

scripts/target_kb_snapshot.sh

@@ -308,6 +308,10 @@ cat >"$OUT/README.md" <<'MD'
 > 若 `MANIFEST.json` 里某文件的 `location` 是 `cloud:<主机>:<路径>`，说明字节留在目标主机上
 > （家用带宽拉不动），本地只有清单与哈希；同机执行时按该路径直接读。
 >
+> JSONL 是 psql `\copy` **text 格式**（反斜杠/控制字符被转义）：回填用对称的 `\copy FROM`
+> 即可原文还原（`scripts/restore_drill.py` 走的就是这条路，并核对行数与抽样内容）；
+> 用程序按行 `json.loads` 会失败，需先反转义。
+>
 > `es_documents.jsonl.gz` 的 sha256 每次导出都不同（scroll 顺序非确定）；判断"两份留档是不是
 > 同一份现场"请看 `es_document_counts` 与逐条的 `_id` 集合，而不是文件哈希。