feat(backup): 备份目录契约 + 导入前留档覆盖 ES 层

- scripts/backup_layout.sh：定义 $KB_BACKUP_ROOT/{target-kb-snapshots,rds-logical,es-snapshots,logs} 与 /etc/knowledge-governance/backup.env，任意主机跑同一份（测试云主机 → 未来生产云主机） - scripts/es_kb_export.py：只读 scroll 导出目标库 ES 文档（gzip JSONL，_id/_source，可按 _id 重建） - scripts/target_kb_snapshot.sh：留档改为 PG 行 + ES 文档两层；PULL_ES 默认 0（692 MB / 家用上行 19m40s，留在目标主机，清单记 location=cloud:<host>:<path>）；挂载源与落点统一走 KB_BACKUP_ROOT - scripts/logical-backup.sh：自管整库逻辑备份（pg_dump -Fd -j4 -Z6，按天轮转） - internal/releasesnapshot：门禁要求 PG 与 ES 两层齐备（ES 计数 >0）；云侧留档核声明与哈希、声明本地却缺失则 fail-closed；新增 3 条用例（共 10） - cmd/guardcheck：用真实产物复核门禁判断 - docs：PRODUCTION_BACKUP_RECOVERY.md 重写（契约/RDS 控制台/恢复路径/边界）；P6 同步留档语义
2026-10-09 12:19:15 +08:00
parent 9155bd3170
commit b32755b35f
10 changed files with 742 additions and 56 deletions
@@ -0,0 +1,31 @@
+// 用真实产物验证发布前门禁：Guard 必须放行覆盖该库的完整留档（PG + ES）。
+package main
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"time"
+
+	"knowledge-governance/internal/releasesnapshot"
+)
+
+func main() {
+	if len(os.Args) < 3 {
+		fmt.Println("用法: guardcheck <快照根目录> <知识库ID> [...]")
+		os.Exit(2)
+	}
+	root, ids := os.Args[1], os.Args[2:]
+	guard := releasesnapshot.Guard{Root: root, MaxAge: 24 * time.Hour, Now: time.Now, VerifyLocalFiles: true}
+	for _, id := range ids {
+		err := guard.Check(id)
+		switch {
+		case err == nil:
+			fmt.Printf("PASS  %s → 门禁放行（PG 行 + ES 文档 + 哈希校验通过）\n", id)
+		case errors.Is(err, releasesnapshot.ErrNoFreshSnapshot):
+			fmt.Printf("BLOCK %s → %v\n", id, err)
+		default:
+			fmt.Printf("ERROR %s → %v\n", id, err)
+		}
+	}
+}
@@ -30,7 +30,8 @@ services:
      - ./config/release.sim.json:/run/release-sim/release.sim.json:ro
      - release-bundles:/var/lib/knowledge-governance/release-bundles
      # 发布前快照留档目录（只读）：写目标库之前必须存在覆盖该库的新快照
-      - ./backups/target-kb-snapshots:/var/lib/knowledge-governance/target-kb-snapshots:ro
+      # 备份目录契约：本地默认 ./backups，主机上设 KB_BACKUP_ROOT=/srv/kb-backups（见 scripts/backup_layout.sh）
+      - ${KB_BACKUP_ROOT:-./backups}/target-kb-snapshots:/var/lib/knowledge-governance/target-kb-snapshots:ro
    read_only: false

 volumes:
@@ -288,16 +288,24 @@ bash scripts/target_kb_snapshot.sh <kb_id> [<kb_id>]  # 指定目标库
 INCLUDE_CHUNKS=0 bash scripts/target_kb_snapshot.sh   # 不导 chunks（快照很小，适合每次发布前）
 ```

-输出 `backups/target-kb-snapshots/<UTC>-<kb前8位>/`：KB 维度行级 JSONL（`to_jsonb(row)`，
-含软删行）+ `MANIFEST.json`（行数/字节/sha256/覆盖范围/ES 文档数）+ `export.sql`（原始查询）
-+ `README.md`（恢复步骤）。全过程只有 SELECT + `\copy`，不修改目标库；`backups/` 已在 .gitignore。
+输出 `$KB_BACKUP_ROOT/target-kb-snapshots/<UTC>-<kb前8位>/`（契约目录见
+`docs/PRODUCTION_BACKUP_RECOVERY.md`；云内执行时先在云主机产出，再把小表与 ES 导出拉回本地）：
+
+- KB 维度行级 JSONL（`to_jsonb(row)`，含软删行，8 张表）；
+- `es_documents.jsonl.gz` + `es_documents.jsonl.gz.manifest.json`：该库在 ES 里的全部文档
+  （`_id` + `_source`，只读 scroll 导出，按 `_id` 可重建）；
+- `MANIFEST.json`（行数/字节/sha256/覆盖范围/每库 ES 文档数）+ `export.sql`（原始查询）+ `README.md`（恢复步骤）。
+
+全过程只有 SELECT / `_count` / `_search`，不修改目标库与索引；`backups/` 已在 .gitignore。

 覆盖范围与边界：

- 覆盖：`knowledge_bases`、`knowledges`、`chunks`、`knowledge_tags`、`knowledge_tag_relations`、
-  `resources`、`resource_bindings`、`wiki_pages`、`wiki_folders`（PG 行；不含表结构）。
- 不覆盖：Elasticsearch 文档与向量、OSS 对象本体（`resources.jsonl` 只给路径/大小/哈希清单）、
-  列出的知识库之外的行。整库级恢复用阿里云 RDS 快照/按时间点恢复（PITR），恢复到临时实例后按 KB 取行。
+- 覆盖：`knowledge_bases`、`knowledges`、`chunks`（可选）、`knowledge_tags`、
+  `knowledge_tag_relations`、`resources`、`resource_bindings`、`wiki_pages`、`wiki_folders`（PG 行），
+  **以及 Elasticsearch 里该库的文档**（含向量字段，按 `_id` 可重建索引）。
+- 不覆盖：OSS 对象本体（`resources.jsonl` 只给路径/大小/哈希清单）、ES 索引的 settings/mappings、
+  列出的知识库之外的行、表结构。整库级恢复用阿里云 RDS 快照/按时间点恢复（PITR），
+  自管整库逻辑备份用 `scripts/logical-backup.sh`。

 约定：对**共享真实知识库**发布、撤权、回滚前必须先跑一次快照，并把快照目录与
 `MANIFEST.json` 的 sha256 记进发布单；`previous_production_revision=absent` 时必须写明
@@ -321,13 +329,15 @@ INCLUDE_CHUNKS=0 bash scripts/target_kb_snapshot.sh   # 不导 chunks（快照
  只对**源库 → 目标库**映射的项强制（同 ID 目标的本地演练不受影响），`always` 则一律强制，
  且此时 `root` 缺失直接判定配置非法。
 - `Publish` / `Delete` / `Rollback` 在**任何写目标库的动作之前**检查：存在覆盖目标库、
-  未过期（`max_age_hours`，默认 24h）、且含 `wiki_pages.jsonl` 及其 sha256 的快照；
-  本地文件会再按哈希校验一次（"可恢复"的最低证明）。任一条不满足——
+  未过期（`max_age_hours`，默认 24h）、含 `wiki_pages.jsonl` 及其 sha256、**且含该库的 ES 文档留档
+  （`es_documents.jsonl.gz` + 计数 > 0）**的快照；本地文件会再按哈希逐份校验
+  （"可恢复"的最低证明）。只留 PG 行、或 ES 计数为 0 的留档一律拒绝。任一条不满足——
  **包括门禁本身未接线**——都以 `release_target_snapshot_missing`（不可重试）拒绝发布，
  且不触碰目标库一行。页面把该码显示为"缺少目标知识库的发布前快照留档，已拒绝发布"。
 - 云内执行：`bash scripts/target_kb_snapshot.sh` 默认在与 DB 同 VPC 的云主机上跑
-  （凭据取自已部署的 WeKnora 容器，不经家用带宽），实测 **44 秒**产出 8 张表 + 清单；
-  隧道模式保留为 `EXECUTION=tunnel` 兜底。
+  （凭据取自已部署的 WeKnora 容器，不经家用带宽），PG 8 张表 **44 秒**；ES 导出走云主机直连
+  ES 容器（只读 scroll），两个库共 13 万条文档 gzip 后约 450 MiB；隧道模式保留为
+  `EXECUTION=tunnel` 兜底。`PULL_ES=0` 可把 ES 大文件留在云主机，只回传清单。

 ## Elasticsearch 通道

@@ -1,52 +1,85 @@
 # 生产级备份与恢复（目标环境）

-面向目标环境（云上 WeKnora v0.7.1 + Aliyun RDS + ES + OSS）。**发布链本身不做发布前备份**，
-所以下面五层各司其职，缺一层就有对应的事故类型救不回来。
+面向目标环境（云上 WeKnora v0.7.1 + Aliyun RDS + ES + OSS）。核心原则只有一条：
+**导入/发布之前，先把目标库现场留档，然后才允许写库** —— 这件事不是靠人记得做，而是靠发布前置硬门禁强制。

 ## 现状（2026-10-09 实测）

 | 层 | 状态 | 证据 |
 |---|---|---|
-| PG 行级快照（KB 维度） | ✅ 已建：`scripts/target_kb_snapshot.sh`，云内执行 **44 秒**出 8 张表 + `MANIFEST.json`（行数/字节/sha256）+ `export.sql` + 恢复 README | `backups/target-kb-snapshots/20261009T025017Z-0b217e8a/` |
-| PG 发布前硬门禁 | ✅ 已上线：release-worker 镜像 `sha256:2913e9e9…`，映射目标（源库→目标库）必须有 24h 内、覆盖该库、哈希可校验的快照，否则 `release_target_snapshot_missing`（不可重试） | 容器内挂载 `/var/lib/knowledge-governance/target-kb-snapshots` + `target_snapshot{mapped,24h}` 已核对；单测 7（门禁）+6（发布器） |
-| ES 快照 | ✅ 已建：仓库 `kb_backup`（`path.repo=/usr/share/elasticsearch/backup`，宿主 `/opt/weknora-test/es-backup`）；首个快照 `initial-20261009t024209z` **36 秒 / 4.7 GB / 0 分片失败**；SLM `kb-hourly`（保留 3 天）+ `kb-daily`（保留 7 天）已建并各执行成功一次 | `GET _snapshot/kb_backup`、`_cat/snapshots`、`_slm/policy/*` |
-| OSS 对象 | ✅ 目标桶 `dobot-weknora-test` 已开启**版本控制**（此前未开启）；未配跨区域复制 | `GET /?versioning` = `<VersioningConfiguration/>` → 开启后 `Enabled` |
-| RDS 整库 | ⏳ **待你在控制台确认** | 库里只看到 `archive_mode=on`，`pg_stat_archiver` 计数 0（平台侧管理），PITR/保留期在库里看不到 |
+| 导入前留档（PG 行 + ES 文档） | ✅ 已建：`scripts/target_kb_snapshot.sh` 云内执行，8 张表 + 该库在 ES 的全部文档（只读 scroll，gzip JSONL，按 `_id` 可重建）+ `MANIFEST.json`（行数/字节/sha256/每库 ES 文档数）+ `export.sql` + `README.md` | `$KB_BACKUP_ROOT/target-kb-snapshots/<UTC>-<kb8>/`（云主机产出一份，本地镜像一份） |
+| 发布前硬门禁 | ✅ 已上线（`internal/releasesnapshot`）：写目标库之前必须存在覆盖该库、24h 内、**PG 行与 ES 文档两层都齐备且哈希可校验**的留档，否则 `release_target_snapshot_missing`（不可重试），目标库一行不碰 | 单测 9（门禁）+ 6（发布器）；`cmd/guardcheck` 可用真实产物复核 |
+| 自管整库逻辑备份 | ✅ 已建：`scripts/logical-backup.sh`（`pg_dump -Fd -j4 -Z6` 目录格式，可并行/按表恢复；按天轮转保留 7 份 + 每月一份） | `$KB_BACKUP_ROOT/rds-logical/<UTC>/<库>.dir` + `MANIFEST.json` |
+| ES 快照 | ✅ 已建：仓库 `kb_backup`（容器内 `path.repo=/usr/share/elasticsearch/backup`，宿主 `$KB_BACKUP_ROOT/es-snapshots`）→ 首个快照 `initial-20261009t024209z` **36 秒 / 4.7 GB / 0 分片失败**；SLM `kb-hourly`（留 3 天）+ `kb-daily`（留 7 天）已各成功执行 | `GET _snapshot/kb_backup/_all`、`_slm/policy/*` |
+| OSS 对象 | ✅ 目标桶 `dobot-weknora-test` 已开启**版本控制**（覆盖/删除可按 VersionId 回滚）；未配跨区域复制 | `GET /?versioning` = `Enabled` |
+| RDS 托管备份 | ⏳ **待控制台确认**：库里 `archive_mode=on` 但 `archive_command` 为空、`pg_stat_archiver.archived_count=0`；当前连接 `is_superuser=off`（看不到备节点）→ 系列与日志备份开关只能看控制台 | 见下节 |

-## 要你做的（RDS 控制台，约 5 分钟）
+## 备份目录契约（换主机只改一处）

-1. **备份设置**：数据备份保留期 **≥7 天**（建议 14 天）；**日志备份（WAL）开启** —— 它决定能不能"按时间点恢复"。
-2. 确认"按时间点恢复"的形态：Aliyun RDS PostgreSQL 一般是**恢复成新实例**，再按 KB 取行。
-3. 把策略页的关键值抄到本目录 `RDS_BACKUP_POLICY.md`：保留天数、备份时间窗、最近一次备份时间、**最近可恢复时间点**。
-4. 若实例支持"备份下载/逻辑备份到 OSS"，一并开启：那是整库级、与行级快照互补。
+任何一台扮演"目标库所在主机"的机器（现在的**测试云主机**、未来的**生产云主机**、以及本地演练机）
+都跑同一份 `scripts/backup_layout.sh`，得到同一套落点：
+
+| 目录 | 内容 | 谁写 |
+|---|---|---|
+| `$KB_BACKUP_ROOT/target-kb-snapshots/<UTC>-<kb8>/` | 导入前留档（PG 8 表 + ES 文档 + 清单） | `scripts/target_kb_snapshot.sh` |
+| `$KB_BACKUP_ROOT/rds-logical/<UTC>/` | 整库逻辑备份（`pg_dump -Fd`） | `scripts/logical-backup.sh` |
+| `$KB_BACKUP_ROOT/es-snapshots/` | ES 共享仓库 `kb_backup` 的宿主目录 | ES 快照 / SLM |
+| `$KB_BACKUP_ROOT/logs/` | 上述脚本运行日志 | 各脚本 |
+
+- 默认 `$KB_BACKUP_ROOT=/srv/kb-backups`，由 `/etc/knowledge-governance/backup.env` 落盘（`backup_layout.sh` 生成）；
+  脚本与 compose 都读它：`source /etc/knowledge-governance/backup.env`，compose 用 `${KB_BACKUP_ROOT:-…}` 插值。
+- 已执行实例：测试云主机 `/srv/kb-backups/`（原 `/opt/weknora-test/{kb-snapshots,es-backup}` 已迁移，
+  容器内路径不变，ES 仓库无需重新注册、快照与 SLM 全部保留）。
+- **切到生产云主机时**：① 在生产云主机跑同一份 `backup_layout.sh`；② 把生产 compose 的
+  ES 挂载源与发布前留档挂载源指向 `$KB_BACKUP_ROOT`；③ 门禁配置 `target_snapshot.root` 保持
+  **容器内路径** `/var/lib/knowledge-governance/target-kb-snapshots`（不用改）；④ 在发布管理里授权前先跑一次留档。
+
+## RDS 控制台（约 5 分钟；先看"系列"）
+
+实例：`pgm-wz9ohr9gk919pr1q`（PostgreSQL 17.9，阿里云 RDS）。
+
+1. 实例列表 → 点实例 ID → 左侧「备份恢复」→「备份策略」。
+2. 「数据备份设置」编辑：备份周期 ≥ 每周 2 次（建议周一/周四）、时间窗 03:00-04:00、
+   **保留天数 14 天**（可选 7~730；不可关闭；云盘实例可开秒级备份/提高频率）。
+3. 「日志备份设置」：**打开日志备份**（决定能否按时间点恢复 PITR）、保留 7 天。
+   注意：日志保留必须 ≤ 数据保留；关掉会清空已有日志备份并失去 PITR。
+4. 抄下四个值到 `RDS_BACKUP_POLICY.md`：数据保留天数、日志备份开关+保留天数、最近一次备份完成时间、**最近可恢复时间点**。
+5. 分支：**高可用系列** → 支持"按时间点恢复到新实例"（云盘实例不复制白名单/安全组，需重配）；
+   **基础系列** → 不支持日志备份/PITR，改用：数据保留期拉长 + 稀疏备份（月/年各留一份）+
+   开云盘加密后启用跨地域备份；控制台有「实例备份下载」按钮时才可导出到 OSS。

 只读旁证（最终以控制台为准）：

 ```bash
-docker run --rm --network host postgres:16 psql "$TARGET_DB_URL" \
-  -c "show archive_mode; show wal_level; select * from pg_stat_archiver"
+docker run --rm --network host postgres:17-alpine psql "$TARGET_DB_URL" \
+  -c "show archive_mode; show archive_command; select archived_count, last_archived_time from pg_stat_archiver"
 ```

 ## 恢复路径（按事故类型选，越快越靠前）

 | 事故 | 首选 | 耗时 |
 |---|---|---|
-| 某个 KB 的 wiki / 元数据被覆盖 | **行级快照回填**（快照目录 `README.md` 的 SQL） | 秒~分钟 |
-| 向量库损坏 / 索引误删 | **ES 快照 restore**（仓库 `kb_backup`） | 分钟级 |
+| 某个 KB 的 wiki / 元数据被覆盖 | **留档回填**（快照目录 `README.md` 的 SQL：先删按 `_id`/主键冲突行，再 `\copy` 回） | 秒~分钟 |
+| 检索/向量异常、索引误删 | **ES 快照 restore**（`kb_backup`；或按 `es_documents.jsonl.gz` 用 `_id` 重建） | 分钟级 |
 | OSS 对象被覆盖 / 删除 | **OSS 版本控制**回滚单对象或前缀 | 秒~分钟 |
-| 整库 / 整实例级错误 | **RDS 快照 / PITR** → 新实例 → 按 KB 取行 | 10~30 分钟 |
-| 只丢了整个库的 wiki 页面 | RDS PITR 取 `wiki_pages`/`wiki_folders` 全量后按 KB 回填 | 10~30 分钟 |
+| 整库被改坏（要带走/异地一份） | **自管逻辑备份** `pg_restore -j4` 到临时库或新实例 | 分钟~十分钟 |
+| 整实例级错误、需要"回到某个时刻" | **RDS 快照 / PITR** → 新实例 → 按 KB 取行 | 10~30 分钟 |
+| 主机/磁盘整体丢失 | ES 仓 + 留档在**同一块盘**上 → 必须靠 RDS 托管备份 + OSS（见"边界"） | 30 分钟+ |

-## 每次发布前的标准动作
+## 每次导入/发布前的标准动作

-1. `bash scripts/target_kb_snapshot.sh`（默认云内执行、默认不含 chunks，约 44 秒；`INCLUDE_CHUNKS=1` 才带 chunks）；
-2. **不需要人工登记**：worker 门禁会核验"覆盖目标库 + 24h 内 + `wiki_pages.jsonl` 哈希可校验"；
-3. 门禁报 `release_target_snapshot_missing` → 先跑第 1 步，再在发布管理里授权。
+1. `bash scripts/target_kb_snapshot.sh`（默认云内执行；PG 8 张表 44 秒，ES 导出按库只读 scroll
+   另需数分钟；`INCLUDE_CHUNKS=1` 才带 chunks，`PULL_ES=0` 把 ES 大文件留在云主机只回传清单）；
+2. **不需要人工登记**：worker 门禁自动核验"覆盖目标库 + 24h 内 + PG 与 ES 两层哈希可校验"；
+3. 门禁报 `release_target_snapshot_missing` → 先跑第 1 步，再在发布管理里授权发布；
+4. 想复核门禁判断，可直接跑 `go run ./cmd/guardcheck $KB_BACKUP_ROOT/target-kb-snapshots <kb_id>`。

 ## 边界（别把这些当"已经备份了"）

- 行级快照**不含**：ES 文档与向量、OSS 对象本体（`resources.jsonl` 只记路径/大小/哈希）、表结构与索引。
- ES 快照仓在**同一台云主机**上（`/opt/weknora-test/es-backup`）；要做异地副本，还需把它同步到 OSS
-  （需要一把可写备份桶的 AK，属你授权项）。
- 行级快照的默认档不含 `chunks`：向量在 ES、分块行可由源端重发；要"整库行级可离线重建"才需要 `INCLUDE_CHUNKS=1`
-  （实测本机隧道约 24 分钟、云内 44 秒内）。
+- 留档**不含**：OSS 对象本体（`resources.jsonl` 只记路径/大小/哈希）、ES 索引 settings/mappings、
+  表结构与索引、列出知识库之外的行。整库级要靠 RDS 托管备份 + 自管逻辑备份。
+- **ES 快照仓与留档在同一台云主机、同一块盘**上：整机故障会一起丢。异地副本需把
+  `es-snapshots/` 与 `target-kb-snapshots/` 同步到 OSS（需一把可写备份桶的 AK，属授权项）。
+- 自管逻辑备份**不能替代 PITR**：它只能回到"备份那一刻"，且恢复要靠 `pg_restore` 手工执行。
+- 留档默认不含 `chunks` 行：向量在 ES 里、分块行可由源端重发；要"整库行级可离线重建"才需
+  `INCLUDE_CHUNKS=1`（实测本机隧道约 24 分钟、云内 44 秒内）。
@@ -2,8 +2,9 @@
 //
 // 背景：发布链本身不做发布前备份（DocumentTarget.Prepare 只归档本包、解析
 // previousRevision），而 wiki 通道在共享真实知识库上会覆盖整库页面。留档由
-// scripts/target_kb_snapshot.sh 产生（KB 维度只读行级 JSONL + MANIFEST.json），
-// 本包只负责在写目标库之前证明"这份现场确实存在且是新的"。
+// scripts/target_kb_snapshot.sh 产生（KB 维度只读 PG 行级 JSONL + ES 文档 + MANIFEST.json），
+// 本包只负责在写目标库之前证明"这份现场确实存在、且 PG 与 ES 两层都覆盖了目标库"。
+// 只留 PG 行的留档在恢复后检索不到东西，因此同样拒绝。
 package releasesnapshot

 import (
@@ -39,6 +40,10 @@ type manifest struct {
 	CreatedAt        string         `json:"created_at"`
 	KnowledgeBaseIDs []string       `json:"knowledge_base_ids"`
 	Files            []manifestFile `json:"files"`
+	// ESDocumentsFile/ESDocumentCounts：ES 那一层的留档。导入前留档必须同时覆盖
+	// PG 行与 ES 文档，否则"恢复"只能恢复一半。
+	ESDocumentsFile  string         `json:"es_documents_file"`
+	ESDocumentCounts map[string]int `json:"es_document_counts"`
 }

 // Guard 在快照根目录下寻找覆盖目标库的有效快照。
@@ -53,9 +58,9 @@ type Guard struct {
 	VerifyLocalFiles bool
 }

-// Check 要求 Root 下存在一份覆盖 knowledgeBaseID 的快照：类型正确、在有效期内、
-// 含 wiki_pages.jsonl 且其 sha256 非空（本地文件再按哈希校验一次）。
-// 找到即返回 nil；否则返回包装了 ErrNoFreshSnapshot 的原因。
+// Check 要求 Root 下存在一份覆盖 knowledgeBaseID 的**完整**导入前留档：
+// 类型正确、在有效期内、含 wiki_pages.jsonl、且含该库的 ES 文档留档（计数 > 0）；
+// 本地文件再按 sha256 校验一次。找到即返回 nil，否则返回包装 ErrNoFreshSnapshot 的原因。
 func (guard Guard) Check(knowledgeBaseID string) error {
 	if strings.TrimSpace(knowledgeBaseID) == "" {
 		return fmt.Errorf("%w: 目标知识库为空", ErrNoFreshSnapshot)
@@ -137,14 +142,40 @@ func (guard Guard) inspect(directory, knowledgeBaseID string, now time.Time) (bo
 		return false, fmt.Errorf("%w: 快照缺少 wiki_pages.jsonl 或哈希（%s）",
 			ErrNoFreshSnapshot, filepath.Base(directory))
 	}
+	// ES 必须一起留档：只有 PG 行的快照，恢复后检索不到东西。
+	if parsed.ESDocumentsFile == "" {
+		return false, fmt.Errorf("%w: 快照未包含 ES 文档留档（%s）",
+			ErrNoFreshSnapshot, filepath.Base(directory))
+	}
+	if parsed.ESDocumentCounts[knowledgeBaseID] <= 0 {
+		return false, fmt.Errorf("%w: 快照里目标库 %s 的 ES 文档数为 %d（%s）",
+			ErrNoFreshSnapshot, knowledgeBaseID, parsed.ESDocumentCounts[knowledgeBaseID], filepath.Base(directory))
+	}
+	esFile, esPresent := findFile(parsed.Files, parsed.ESDocumentsFile)
+	if !esPresent || esFile.SHA256 == "" {
+		return false, fmt.Errorf("%w: 快照缺少 ES 文档文件或其哈希（%s）",
+			ErrNoFreshSnapshot, filepath.Base(directory))
+	}
 	if guard.VerifyLocalFiles {
 		if err := verifyLocal(directory, wiki); err != nil {
 			return false, err
 		}
+		// ES 留档可能是"云侧留档"（location 以 cloud: 开头）：家用带宽拉几 GB 不现实，
+		// 此时字节在目标主机上，本地只核清单声明与哈希字符串；声明为本地却读不到 → 拒绝。
+		if !isCloudLocated(esFile.Location) {
+			if err := verifyLocal(directory, esFile); err != nil {
+				return false, err
+			}
+		}
 	}
 	return true, nil
 }

+// isCloudLocated 判断某份留档是否声明为"字节在目标主机上"（而非本地）。
+func isCloudLocated(location string) bool {
+	return strings.HasPrefix(strings.ToLower(strings.TrimSpace(location)), "cloud:")
+}
+
 // parseStamp 接受脚本写出的 20261009T025017Z，也接受 RFC3339。
 func parseStamp(value string) (time.Time, error) {
 	value = strings.TrimSpace(value)
@@ -34,8 +34,22 @@ func writeSnapshot(t *testing.T, root, name, createdAt string, kbIDs []string, w
 			"file": "wiki_pages.jsonl", "rows": 1, "bytes": len(wikiContent), "sha256": sha, "location": "local",
 		})
 	}
+	esContent := "{\"_id\":\"chunk-1\",\"_source\":{}}\n"
+	if err := os.WriteFile(filepath.Join(directory, "es_documents.jsonl.gz"), []byte(esContent), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	esDigest := sha256.Sum256([]byte(esContent))
+	files = append(files, map[string]any{
+		"file": "es_documents.jsonl.gz", "bytes": len(esContent),
+		"sha256": hex.EncodeToString(esDigest[:]), "location": "local",
+	})
+	counts := map[string]int{}
+	for _, kb := range kbIDs {
+		counts[kb] = 1
+	}
 	manifestPayload, err := json.Marshal(map[string]any{
 		"kind": manifestKind, "created_at": createdAt, "knowledge_base_ids": kbIDs, "files": files,
+		"es_documents_file": "es_documents.jsonl.gz", "es_document_counts": counts,
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -121,3 +135,110 @@ func TestGuardSkipsHalfWrittenSnapshotAndPicksNewestValid(t *testing.T) {
 		t.Fatalf("valid snapshot behind a half-written dir must still pass: %v", err)
 	}
 }
+
+// 只有 PG 行的快照不算完整留档：恢复后检索不到东西，必须拒绝。
+func TestGuardRejectsSnapshotWithoutESDocuments(t *testing.T) {
+	root := t.TempDir()
+	directory := filepath.Join(root, "20261009T025017Z-0b217e8a")
+	if err := os.MkdirAll(directory, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	wiki := "{}\n"
+	if err := os.WriteFile(filepath.Join(directory, "wiki_pages.jsonl"), []byte(wiki), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	digest := sha256.Sum256([]byte(wiki))
+	manifestPayload, err := json.Marshal(map[string]any{
+		"kind": manifestKind, "created_at": "20261009T025017Z", "knowledge_base_ids": []string{"kb-a"},
+		"files": []map[string]any{{"file": "wiki_pages.jsonl", "sha256": hex.EncodeToString(digest[:]), "location": "local"}},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(directory, "MANIFEST.json"), manifestPayload, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	guard := Guard{Root: root, Now: fixedNow(), VerifyLocalFiles: true}
+	if err := guard.Check("kb-a"); !errors.Is(err, ErrNoFreshSnapshot) {
+		t.Fatalf("snapshot without ES layer must be refused, got %v", err)
+	}
+}
+
+// ES 层存在但目标库文档数为 0（导出失败/库写错）也必须拒绝。
+func TestGuardRejectsSnapshotWithZeroESDocumentsForTarget(t *testing.T) {
+	root := t.TempDir()
+	writeSnapshot(t, root, "20261009T025017Z-0b217e8a", "20261009T025017Z", []string{"kb-a"}, "{}\n", "")
+	// 把目标库的 ES 计数改成 0
+	path := filepath.Join(root, "20261009T025017Z-0b217e8a", "MANIFEST.json")
+	payload, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var parsed map[string]any
+	if err := json.Unmarshal(payload, &parsed); err != nil {
+		t.Fatal(err)
+	}
+	parsed["es_document_counts"] = map[string]int{"kb-a": 0}
+	rewritten, err := json.Marshal(parsed)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(path, rewritten, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	guard := Guard{Root: root, Now: fixedNow(), VerifyLocalFiles: true}
+	if err := guard.Check("kb-a"); !errors.Is(err, ErrNoFreshSnapshot) {
+		t.Fatalf("zero ES documents for the target must be refused, got %v", err)
+	}
+}
+
+// 云侧留档（家用带宽不拉大文件）：清单声明 + 哈希字符串即放行；声明本地却读不到 → 拒绝。
+func TestGuardAcceptsCloudLocatedESArtifacts(t *testing.T) {
+	root := t.TempDir()
+	directory := filepath.Join(root, "20261009T040000Z-0b217e8a")
+	if err := os.MkdirAll(directory, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	wiki := "{}\n"
+	if err := os.WriteFile(filepath.Join(directory, "wiki_pages.jsonl"), []byte(wiki), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	digest := sha256.Sum256([]byte(wiki))
+	manifestPayload, err := json.Marshal(map[string]any{
+		"kind": manifestKind, "created_at": "20261009T040000Z", "knowledge_base_ids": []string{"kb-a"},
+		"es_documents_file": "es_documents.jsonl.gz", "es_document_counts": map[string]int{"kb-a": 61613},
+		"files": []map[string]any{
+			{"file": "wiki_pages.jsonl", "sha256": hex.EncodeToString(digest[:]), "location": "local"},
+			{"file": "es_documents.jsonl.gz", "sha256": "deadbeef", "bytes": 692023258,
+				"location": "cloud:weknora-test:/srv/kb-backups/target-kb-snapshots/20261009T040000Z-0b217e8a/es_documents.jsonl.gz"},
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(directory, "MANIFEST.json"), manifestPayload, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	guard := Guard{Root: root, Now: fixedNow(), VerifyLocalFiles: true}
+	if err := guard.Check("kb-a"); err != nil {
+		t.Fatalf("cloud-located ES artifact must be accepted, got %v", err)
+	}
+
+	// 同一份清单改成"声明为本地"，文件却不在 → 必须拒绝（fail-closed）
+	var parsed map[string]any
+	if err := json.Unmarshal(manifestPayload, &parsed); err != nil {
+		t.Fatal(err)
+	}
+	files := parsed["files"].([]any)
+	files[1].(map[string]any)["location"] = "local"
+	rewritten, err := json.Marshal(parsed)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(directory, "MANIFEST.json"), rewritten, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if err := guard.Check("kb-a"); !errors.Is(err, ErrNoFreshSnapshot) {
+		t.Fatalf("local-declared but missing ES artifact must be refused, got %v", err)
+	}
+}
@@ -0,0 +1,66 @@
+#!/usr/bin/env bash
+# 备份/快照的**目录契约**：任何一台扮演"目标库所在主机"的机器（现在的测试云主机、
+# 未来的生产云主机、以及本地演练机）都跑这一份，得到同一套落点。
+#
+# 为什么要统一：发布前的留档、整库逻辑备份、ES 快照仓、发布包归档分属不同脚本，
+# 一旦各自选路径，换主机时就要逐个改脚本，且"发布前快照"的门禁会指到错的地方。
+#
+# 目录布局（$KB_BACKUP_ROOT 底下）：
+#   target-kb-snapshots/<UTC>-<kb8>/   发布/导入前的行级留档（PG 8 表 + ES 文档，小、随发布留）
+#   rds-logical/<UTC>/                 整库逻辑备份（pg_dump -Fd 目录格式，按天轮转）
+#   es-snapshots/                      Elasticsearch 共享仓库 kb_backup 的宿主目录
+#   logs/                              上述脚本的运行日志
+#
+# 用法（在目标主机上）：
+#   bash backup_layout.sh                 # 建目录 + 落配置，打印现状
+#   KB_BACKUP_ROOT=/data/kb-backups bash backup_layout.sh
+#   bash backup_layout.sh --print         # 只打印契约，不动磁盘
+#
+# 产出配置：/etc/knowledge-governance/backup.env（KB_BACKUP_ROOT=…），供各脚本 source。
+set -euo pipefail
+
+KB_BACKUP_ROOT="${KB_BACKUP_ROOT:-/srv/kb-backups}"
+CONFIG_FILE="${CONFIG_FILE:-/etc/knowledge-governance/backup.env}"
+SUBDIRS=(target-kb-snapshots rds-logical es-snapshots logs)
+PRINT_ONLY=0
+[ "${1:-}" = "--print" ] && PRINT_ONLY=1
+
+echo "备份根目录: $KB_BACKUP_ROOT"
+echo "配置文件  : $CONFIG_FILE"
+echo "子目录    :"
+total_bytes=0
+for sub in "${SUBDIRS[@]}"; do
+  path="$KB_BACKUP_ROOT/$sub"
+  if [ "$PRINT_ONLY" = "0" ]; then
+    mkdir -p "$path"
+  fi
+  if [ -d "$path" ]; then
+    usage="$(du -sh "$path" 2>/dev/null | cut -f1)"
+    entries="$(find "$path" -mindepth 1 -maxdepth 1 -type d 2>/dev/null | wc -l)"
+    printf '  %-20s %-6s %s 个子项\n' "$sub" "$usage" "$entries"
+  else
+    printf '  %-20s (未创建)\n' "$sub"
+  fi
+done
+
+if [ "$PRINT_ONLY" = "1" ]; then
+  echo "（--print：未改动磁盘）"
+  exit 0
+fi
+
+mkdir -p "$(dirname "$CONFIG_FILE")"
+cat >"$CONFIG_FILE" <<EOF
+# 由 scripts/backup_layout.sh 生成：备份目录契约（换主机时只改这里）
+KB_BACKUP_ROOT=$KB_BACKUP_ROOT
+EOF
+echo "已写入 $CONFIG_FILE"
+echo
+echo "磁盘剩余空间："
+df -h "$KB_BACKUP_ROOT" | tail -1
+echo
+echo "接下来说明："
+echo "  1) 各脚本通过 KB_BACKUP_ROOT 定位落点，可用 'source $CONFIG_FILE' 载入；"
+echo "  2) ES 快照仓：compose 把 \$KB_BACKUP_ROOT/es-snapshots 挂到容器 /usr/share/elasticsearch/backup，"
+echo "     仓库名 kb_backup（容器内路径不变，迁移目录无需重新注册仓库）；"
+echo "  3) 发布前门禁读取的目录 = \$KB_BACKUP_ROOT/target-kb-snapshots，容器内只读挂载；"
+echo "  4) 换到生产云主机时：跑同一份脚本 → 改 /opt/*/compose.yaml 的挂载源 → 门禁配置里的 root 保持容器内路径即可。"
@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+"""目标 WeKnora ES 索引里某个（些）知识库的文档导出（只读，scroll + gzip JSONL）。
+
+用途：发布/导入之前把 ES 那一层也留档到本地，与 PG 行级快照配成"导入前完整现场"。
+只发 _count / _search / _search/scroll，不写 ES 任何数据。
+
+用法：
+  python3 es_kb_export.py --endpoint http://172.18.0.3:9200 --index weknora_test_chunks \\
+      --kb <kb_id> [--kb <kb_id> ...] --output /path/es_documents.jsonl.gz [--limit N] [--batch 2000]
+
+输出：
+  <output>                 gzip JSONL，一行一条 {"_index":..,"_id":..,"_source":{...}}
+  <output>.manifest.json   每个库的文档数、字节数、gz 的 sha256/size、索引与 endpoint
+"""
+import argparse
+import gzip
+import hashlib
+import json
+import sys
+import urllib.error
+import urllib.request
+
+
+def call(endpoint, path, payload=None, method=None, timeout=300):
+    url = endpoint.rstrip("/") + path
+    data = json.dumps(payload).encode() if payload is not None else None
+    request = urllib.request.Request(url, data=data, method=method or ("POST" if data else "GET"),
+                                     headers={"Content-Type": "application/json"})
+    try:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            return json.load(response)
+    except urllib.error.HTTPError as error:
+        body = error.read().decode("utf-8", "replace")[:300]
+        raise SystemExit(f"ES 请求失败 {path}: HTTP {error.code} {body}")
+
+
+def count_for(endpoint, index, knowledge_base_id):
+    result = call(endpoint, f"/{index}/_count",
+                  {"query": {"term": {"knowledge_base_id.keyword": knowledge_base_id}}})
+    return int(result.get("count", 0))
+
+
+def export(endpoint, index, knowledge_base_ids, output, batch, limit):
+    counts = {}
+    written = 0
+    uncompressed_bytes = 0
+    for kb in knowledge_base_ids:
+        counts[kb] = count_for(endpoint, index, kb)
+
+    # scroll：每次 batch 条，_doc 排序，只取 _source
+    with gzip.open(output, "wt", encoding="utf-8", compresslevel=6) as handle:
+        for kb in knowledge_base_ids:
+            if counts[kb] == 0:
+                continue
+            scroll_id = None
+            try:
+                response = call(endpoint, f"/{index}/_search?scroll=5m", {
+                    "size": min(batch, limit) if limit else batch,
+                    "sort": ["_doc"],
+                    "query": {"term": {"knowledge_base_id.keyword": kb}},
+                    "_source": True,
+                })
+                while True:
+                    scroll_id = response.get("_scroll_id") or scroll_id
+                    hits = response.get("hits", {}).get("hits", [])
+                    if not hits:
+                        break
+                    for hit in hits:
+                        line = json.dumps({"_index": hit.get("_index"), "_id": hit.get("_id"),
+                                           "_source": hit.get("_source")}, ensure_ascii=False)
+                        handle.write(line + "\n")
+                        written += 1
+                        uncompressed_bytes += len(line.encode("utf-8"))
+                        if limit and written >= limit:
+                            break
+                    if limit and written >= limit:
+                        break
+                    response = call(endpoint, "/_search/scroll",
+                                    {"scroll": "5m", "scroll_id": scroll_id})
+            finally:
+                if scroll_id:
+                    try:
+                        call(endpoint, "/_search/scroll", {"scroll_id": [scroll_id]}, method="DELETE")
+                    except SystemExit:
+                        pass
+            if limit and written >= limit:
+                break
+
+    digest = hashlib.sha256()
+    with open(output, "rb") as handle:
+        for block in iter(lambda: handle.read(1 << 20), b""):
+            digest.update(block)
+    manifest = {
+        "kind": "weknora-es-kb-export",
+        "endpoint": endpoint,
+        "index": index,
+        "knowledge_base_document_counts": counts,
+        "documents_written": written,
+        "uncompressed_bytes": uncompressed_bytes,
+        "file": output.split("/")[-1],
+        "bytes": __import__("os").path.getsize(output),
+        "sha256": digest.hexdigest(),
+    }
+    with open(output + ".manifest.json", "w", encoding="utf-8") as handle:
+        json.dump(manifest, handle, ensure_ascii=False, indent=2)
+        handle.write("\n")
+    return manifest
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--endpoint", required=True)
+    parser.add_argument("--index", default="weknora_test_chunks")
+    parser.add_argument("--kb", action="append", required=True)
+    parser.add_argument("--output", required=True)
+    parser.add_argument("--batch", type=int, default=2000)
+    parser.add_argument("--limit", type=int, default=0, help="仅用于冒烟测试：最多导出多少条")
+    arguments = parser.parse_args()
+    manifest = export(arguments.endpoint, arguments.index, arguments.kb, arguments.output,
+                      arguments.batch, arguments.limit)
+    print(json.dumps({"documents": manifest["documents_written"],
+                      "counts": manifest["knowledge_base_document_counts"],
+                      "bytes": manifest["bytes"], "sha256": manifest["sha256"][:16] + "…"},
+                     ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,190 @@
+#!/usr/bin/env bash
+# 目标 RDS PostgreSQL 的**自管逻辑备份**：手动触发，文件落在你自己这台云主机上，
+# 不经过 RDS 托管备份服务。
+#
+# 为什么要它：RDS 托管的备份只存在阿里云内部存储，你既看不到也拿不走，只能"恢复成
+# 新实例"；这份是你能随手取用、能在任何地方 pg_restore 的自有副本。两者互补：
+# 托管备份 + 日志备份负责"按时间点恢复"，这份负责"带走 / 异地 / 独立校验"。
+#
+# 用法（在云主机上）：
+#   bash logical-backup.sh                          # 全库逻辑备份 + 轮转
+#   BACKUP_ROOT=/data/rds-backups bash logical-backup.sh
+#   DBS="weknora_test" bash logical-backup.sh       # 只备指定库
+#   KEEP_DAILY=14 KEEP_MONTHS=6 bash logical-backup.sh
+#   DRY_RUN=1 bash logical-backup.sh                # 只看会做什么
+#
+# 环境变量：
+#   BACKUP_ROOT   备份根目录（默认 /opt/weknora-test/rds-backups）
+#   APP_CONTAINER 提供 DB 凭据的容器（默认 weknora-test-v071-app-1）
+#   PG_IMAGE      pg_dump 容器镜像（默认 postgres:17.9-alpine，必须 ≥ 服务端 17.9）
+#   DBS           逗号分隔的库名；默认自动列出非模板库
+#   JOBS          并行度（默认 4）
+#   KEEP_DAILY    保留最近几份每日备份（默认 7）
+#   KEEP_MONTHS   额外保留最近几个自然月各一份（默认 3）
+#   EXCLUDE_DBS   逗号分隔的跳过列表（默认空）
+set -euo pipefail
+
+# 备份目录契约：由 scripts/backup_layout.sh 写到 /etc/knowledge-governance/backup.env。
+# 换主机（测试云主机 → 生产云主机）时只改这一处，脚本本身不动。
+if [ -z "${KB_BACKUP_ROOT:-}" ] && [ -f /etc/knowledge-governance/backup.env ]; then
+  # shellcheck disable=SC1091
+  . /etc/knowledge-governance/backup.env
+fi
+KB_BACKUP_ROOT="${KB_BACKUP_ROOT:-/srv/kb-backups}"
+BACKUP_ROOT="${BACKUP_ROOT:-$KB_BACKUP_ROOT/rds-logical}"
+APP_CONTAINER="${APP_CONTAINER:-weknora-test-v071-app-1}"
+PG_IMAGE="${PG_IMAGE:-postgres:17.9-alpine}"
+JOBS="${JOBS:-4}"
+KEEP_DAILY="${KEEP_DAILY:-7}"
+KEEP_MONTHS="${KEEP_MONTHS:-3}"
+DRY_RUN="${DRY_RUN:-0}"
+DBS="${DBS:-}"
+EXCLUDE_DBS="${EXCLUDE_DBS:-}"
+STAMP="$(date -u +%Y%m%dT%H%M%SZ)"
+
+env_value() {
+  docker inspect "$APP_CONTAINER" --format '{{range .Config.Env}}{{println .}}{{end}}' \
+    | grep -E "^$1=" | head -1 | cut -d= -f2-
+}
+DB_HOST_VALUE="$(env_value DB_HOST)"; DB_PORT_VALUE="$(env_value DB_PORT)"
+DB_USER_VALUE="$(env_value DB_USER)"; DB_PASS_VALUE="$(env_value DB_PASSWORD)"
+if [ -z "${DB_HOST_VALUE}${DB_PORT_VALUE}${DB_USER_VALUE}${DB_PASS_VALUE}" ]; then
+  echo "缺少数据库凭据：容器 $APP_CONTAINER 里没有 DB_* 环境变量" >&2; exit 1
+fi
+PGURL="$(python3 - "$DB_USER_VALUE" "$DB_PASS_VALUE" "$DB_HOST_VALUE" "$DB_PORT_VALUE" "$(env_value DB_NAME)" <<'PY'
+import sys, urllib.parse as u
+user, password, host, port, name = sys.argv[1:6]
+print(f"postgres://{u.quote(user)}:{u.quote(password)}@{host}:{port}/{name}")
+PY
+)"
+
+# 库列表：默认所有非模板库
+if [ -z "$DBS" ]; then
+  DBS="$(docker run --rm -e PGURL="$PGURL" "$PG_IMAGE" sh -c \
+    "psql \"\$PGURL\" -tAc \"select datname from pg_database where datistemplate = false and datallowconn order by datname\"" \
+    | tr -d '\r' | paste -sd, -)"
+fi
+if [ -n "$EXCLUDE_DBS" ]; then
+  DBS="$(python3 -c "
+import sys
+exclude = set(filter(None, sys.argv[2].split(',')))
+print(','.join(d for d in sys.argv[1].split(',') if d and d not in exclude))
+" "$DBS" "$EXCLUDE_DBS")"
+fi
+if [ -z "$DBS" ]; then echo "没有可备份的库" >&2; exit 1; fi
+
+TARGET="$BACKUP_ROOT/$STAMP"
+echo "备份目标: $TARGET"
+echo "库列表  : $DBS"
+echo "镜像    : $PG_IMAGE (并行 $JOBS)"
+if [ "$DRY_RUN" = "1" ]; then echo "DRY_RUN=1，不执行"; exit 0; fi
+
+mkdir -p "$TARGET"
+STARTED_AT="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
+START_SECONDS="$(date +%s)"
+
+# 全局对象（角色/表空间）：RDS 上通常受限，尽力而为，失败不阻塞。
+docker run --rm -e PGURL="$PGURL" "$PG_IMAGE" sh -c \
+  "pg_dumpall --globals-only --no-role-passwords --dbname=\"\$PGURL\"" >"$TARGET/globals.sql" 2>"$TARGET/globals.err" || true
+[ -s "$TARGET/globals.err" ] || rm -f "$TARGET/globals.err"
+
+IFS=',' read -r -a DB_ARRAY <<<"$DBS"
+for db in "${DB_ARRAY[@]}"; do
+  db="$(printf '%s' "$db" | tr -d ' ')"
+  [ -n "$db" ] || continue
+  DB_URL="$(python3 - "$DB_USER_VALUE" "$DB_PASS_VALUE" "$DB_HOST_VALUE" "$DB_PORT_VALUE" "$db" <<'PY'
+import sys, urllib.parse as u
+user, password, host, port, name = sys.argv[1:6]
+print(f"postgres://{u.quote(user)}:{u.quote(password)}@{host}:{port}/{name}")
+PY
+)"
+  echo "  - $db ..."
+  docker run --rm -e DB_URL="$DB_URL" -v "$TARGET:/backup" "$PG_IMAGE" sh -c \
+    "pg_dump -Fd -j $JOBS -Z 6 --no-sync -d \"\$DB_URL\" -f /backup/${db}.dir"
+  # pg_dump -Fd 不接受已有目录；目录名冲突说明本次重复执行，直接报错退出
+done
+
+# dump 之后写清单与说明（先写清单，再算哈希，避免把哈希写进自己的输入）
+DURATION="$(( $(date +%s) - START_SECONDS ))"
+python3 - "$TARGET" "$STAMP" "$STARTED_AT" "$DURATION" "$DBS" "$PG_IMAGE" "$JOBS" <<'PY'
+import hashlib, json, os, pathlib, sys
+target = pathlib.Path(sys.argv[1]); stamp, started, duration = sys.argv[2], sys.argv[3], int(sys.argv[4])
+databases, image, jobs = sys.argv[5], sys.argv[6], int(sys.argv[7])
+files = []
+for path in sorted(target.rglob("*")):
+    if path.is_dir() or path.name in ("MANIFEST.json",):
+        continue
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for block in iter(lambda: handle.read(1 << 20), b""):
+            digest.update(block)
+    files.append({"file": str(path.relative_to(target)), "bytes": path.stat().st_size,
+                  "sha256": digest.hexdigest()})
+manifest = {
+    "kind": "rds-logical-backup",
+    "created_at": stamp,
+    "started_at": started,
+    "duration_seconds": duration,
+    "databases": [name for name in databases.split(",") if name],
+    "format": "pg_dump -Fd (directory, parallel, zlib)",
+    "pg_dump_image": image,
+    "parallel_jobs": jobs,
+    "restore_hint": "pg_restore -d <目标库> -j4 <库名>.dir（先建库；跨实例恢复可能需要 --no-owner/--no-acl）",
+    "total_bytes": sum(item["bytes"] for item in files),
+    "files": files,
+}
+(target / "MANIFEST.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n")
+print(f"duration={duration}s total={manifest['total_bytes']/1048576:.1f}MiB files={len(files)}")
+PY
+
+cat >"$TARGET/README.md" <<'MD'
+# 自管逻辑备份（pg_dump 目录格式）
+
+- 谁做的：`logical-backup.sh`，手动触发，文件在本机磁盘（不经 RDS 托管备份服务）。
+- 每个库一个 `.dir` 目录（`pg_dump -Fd -jN -Z6`）：**目录格式支持并行恢复与按表选择性恢复**。
+- `MANIFEST.json`：库列表、耗时、每个文件的 sha256/size。
+- `globals.sql`：角色/表空间（RDS 上通常受限，尽力而为，可能为空）。
+
+## 怎么恢复
+
+```bash
+# 1) 建目标库（已在别处建好则跳过）
+psql "$TARGET_URL" -c 'CREATE DATABASE restore_check'
+
+# 2) 并行恢复（-j 至少 4；目录格式才能并行）
+docker run --rm -e TARGET_URL="$TARGET_URL" -v "$PWD:/backup" postgres:17.9-alpine \
+  pg_restore -d "$TARGET_URL" -j4 --clean --if-exists /backup/<库名>.dir
+
+# 3) 只验某张表（不动生产）
+docker run --rm -v "$PWD:/backup" postgres:17.9-alpine \
+  pg_restore -l /backup/<库名>.dir | grep -i wiki_pages
+```
+
+注意：
+- **pg_restore 版本必须 ≥ dump 时的 pg_dump**（本次是 17.9）→ 用 `postgres:17.9-alpine`。
+- 跨实例恢复若报角色/权限错误，加 `--no-owner --no-acl`。
+- 这份备份**不能替代按时间点恢复（PITR）**：它只能回到"备份那一刻"。
+MD
+
+# 轮转：保留最近 KEEP_DAILY 份，另在最近 KEEP_MONTHS 个自然月各留一份
+python3 - "$BACKUP_ROOT" "$KEEP_DAILY" "$KEEP_MONTHS" <<'PY'
+import pathlib, re, shutil, sys
+root = pathlib.Path(sys.argv[1]); keep_daily = int(sys.argv[2]); keep_months = int(sys.argv[3])
+stamps = sorted([p for p in root.iterdir() if p.is_dir() and re.fullmatch(r"\d{8}T\d{6}Z", p.name)], reverse=True)
+keep = {p.name for p in stamps[:keep_daily]}
+months = []
+for path in stamps:
+    month = path.name[:6]
+    if month not in months:
+        months.append(month)
+        keep.add(path.name)
+    if len(months) >= keep_months:
+        break
+for path in stamps:
+    if path.name not in keep:
+        shutil.rmtree(path)
+        print("rotated out:", path.name)
+print("kept:", len([p for p in stamps if p.name in keep]))
+PY
+
+echo "done: $TARGET"
@@ -16,6 +16,10 @@
 #   SNAPSHOT_ROOT   输出根目录（默认 backups/target-kb-snapshots）
 #   INCLUDE_CHUNKS  1=连 chunks 一起导（默认 0：秒级完成；向量在 ES 侧，行可从源端重发）
 #   PULL_CHUNKS     1=云内执行时把 chunks 结果也拉回本机（默认 0，留在云主机）
+#   PULL_ES         1=把 ES 文档导出（gzip JSONL）也拉回本机。默认 0：ES 导出有数百 MB
+#                   （实测两库 692 MB），家用上行只有 ~0.6 MB/s（约 19 分钟），留在目标主机更实际；
+#                   门禁按 location=cloud:… 核清单声明与哈希，同机部署时直接读本地文件
+#   ES_BATCH        ES scroll 每次取多少条（默认 2000）
 #   TUNNEL_HOST/TUNNEL_PORT  隧道模式的目标 PG 地址（默认 172.17.0.1:25432）
 #   SSH_CONFIG     SSH 配置路径（默认 ~/.ssh/config.weknora-cloud-test）
 #   SSH_HOST       云主机别名（默认 weknora-test）
@@ -27,6 +31,7 @@
 # 覆盖范围（PG 行级）：knowledge_bases / knowledges / knowledge_tags /
 # knowledge_tag_relations / wiki_pages / wiki_folders / resources /
 # resource_bindings（+ chunks，可选）。含软删行，不含表结构。
+# 另外：ES 文档按库导出 es_documents.jsonl.gz（含 _id/_source，可按 _id 重建）。
 #
 # 不覆盖：Elasticsearch 文档与向量（见 ES 仓库 kb_backup 与 SLM 策略 kb-hourly/kb-daily）、
 # OSS 对象本体（只在 resources.jsonl 里记路径/大小/哈希做清单）、表结构与索引。
@@ -34,6 +39,7 @@
 set -euo pipefail

 WORKDIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 cd "$WORKDIR"

 EXECUTION="${EXECUTION:-cloud}"
@@ -42,10 +48,20 @@ TUNNEL_HOST="${TUNNEL_HOST:-172.17.0.1}"
 TUNNEL_PORT="${TUNNEL_PORT:-25432}"
 INCLUDE_CHUNKS="${INCLUDE_CHUNKS:-0}"
 PULL_CHUNKS="${PULL_CHUNKS:-0}"
+# ES 文档也留档（只读 scroll 导出）；PULL_ES=0 时只在云主机留一份
+PULL_ES="${PULL_ES:-0}"
+ES_BATCH="${ES_BATCH:-2000}"
 SSH_CONFIG="${SSH_CONFIG:-$HOME/.ssh/config.weknora-cloud-test}"
 SSH_HOST="${SSH_HOST:-weknora-test}"
 APP_CONTAINER="${APP_CONTAINER:-weknora-test-v071-app-1}"
-REMOTE_ROOT="${REMOTE_ROOT:-/opt/weknora-test/kb-snapshots}"
+# 备份目录契约：由 scripts/backup_layout.sh 写到 /etc/knowledge-governance/backup.env。
+# 换主机（测试云主机 → 生产云主机）时只改这一处，脚本本身不动。
+if [ -z "${KB_BACKUP_ROOT:-}" ] && [ -f /etc/knowledge-governance/backup.env ]; then
+  # shellcheck disable=SC1091
+  . /etc/knowledge-governance/backup.env
+fi
+KB_BACKUP_ROOT="${KB_BACKUP_ROOT:-/srv/kb-backups}"
+REMOTE_ROOT="${REMOTE_ROOT:-$KB_BACKUP_ROOT/target-kb-snapshots}"
 WITH_ES="${WITH_ES:-1}"
 NOTE="${NOTE:-}"

@@ -100,13 +116,18 @@ SQL
 fi

 REMOTE_DIR=""
+CLOUD_REF=""
 if [ "$EXECUTION" = "cloud" ]; then
  # ---------- 云内执行：与 DB 同 VPC，内网直连，不受家用带宽限制 ----------
  REMOTE_DIR="$REMOTE_ROOT/$STAMP-${PREFIX}"
+  CLOUD_REF="${SSH_HOST}:$REMOTE_DIR"
  ssh -F "$SSH_CONFIG" "$SSH_HOST" "mkdir -p '$REMOTE_DIR'"
  scp -q -F "$SSH_CONFIG" "$OUT/export.sql" "$SSH_HOST:$REMOTE_DIR/export.sql"
+  scp -q -F "$SSH_CONFIG" "$SCRIPT_DIR/es_kb_export.py" "$SSH_HOST:$REMOTE_DIR/es_kb_export.py"
  ssh -F "$SSH_CONFIG" "$SSH_HOST" \
    APP_CONTAINER="$APP_CONTAINER" REMOTE_DIR="$REMOTE_DIR" INCLUDE_CHUNKS="$INCLUDE_CHUNKS" \
+    ES_INDEX="${ES_INDEX:-weknora_test_chunks}" ES_BATCH="$ES_BATCH" KB_LIST="$(IFS=,; echo "${KB_IDS[*]}")" \
+    KB_BACKUP_ROOT="$KB_BACKUP_ROOT" \
    'bash -s' <<'REMOTE'
 set -euo pipefail
 env_value() { docker inspect "$APP_CONTAINER" --format '{{range .Config.Env}}{{println .}}{{end}}' | grep -E "^$1=" | head -1 | cut -d= -f2-; }
@@ -138,6 +159,28 @@ for path in sorted(out.glob("*.jsonl*")):
 (out / "REMOTE_MANIFEST.json").write_text(json.dumps({"host_files": files}, indent=2) + "\n")
 print("remote files:", " ".join(f"{f['file']}={f['bytes']}" for f in files))
 PY
+# ES 文档导出（只读 scroll）：云主机可直连 ES 容器
+if [ -f "$REMOTE_DIR/es_kb_export.py" ]; then
+  ES_IP=$(docker inspect weknora-test-v071-elasticsearch-1 --format "{{json .NetworkSettings.Networks}}" \
+    | python3 -c "import json,sys; d=json.load(sys.stdin); print([v['IPAddress'] for v in d.values() if v.get('IPAddress')][0])")
+  ES_KB_ARGS=""
+  for kb in $(printf '%s' "$KB_LIST" | tr ',' ' '); do ES_KB_ARGS="$ES_KB_ARGS --kb $kb"; done
+  python3 "$REMOTE_DIR/es_kb_export.py" --endpoint "http://$ES_IP:9200" --index "$ES_INDEX" \
+    $ES_KB_ARGS --output "$REMOTE_DIR/es_documents.jsonl.gz" --batch "$ES_BATCH"
+  rm -f "$REMOTE_DIR/es_kb_export.py"
+  python3 - "$REMOTE_DIR" <<'PY'
+import hashlib, json, pathlib, sys
+out = pathlib.Path(sys.argv[1]); files = []
+for path in sorted(out.glob("*.jsonl*")):
+    digest = hashlib.sha256(); size = 0
+    with path.open("rb") as handle:
+        for block in iter(lambda: handle.read(1 << 20), b""):
+            digest.update(block); size += len(block)
+    files.append({"file": path.name, "bytes": size, "sha256": digest.hexdigest()})
+(out / "REMOTE_MANIFEST.json").write_text(json.dumps({"host_files": files}, indent=2) + "\n")
+print("remote files:", " ".join(f"{f['file']}={f['bytes']}" for f in files))
+PY
+fi
 exit 0
 REMOTE
  # 小表拉回本机；chunks 默认留在云主机（PULL_CHUNKS=1 才拉）
@@ -146,6 +189,11 @@ REMOTE
  if [ "$PULL_CHUNKS" = "1" ]; then
    scp -q -F "$SSH_CONFIG" "$SSH_HOST:$REMOTE_DIR/chunks.jsonl.gz" "$OUT/" 2>/dev/null || true
  fi
+  # ES 文档：默认拉回本机；PULL_ES=0 时只拉清单（大文件留在云主机）
+  scp -q -F "$SSH_CONFIG" "$SSH_HOST:$REMOTE_DIR/es_documents.jsonl.gz.manifest.json" "$OUT/" 2>/dev/null || true
+  if [ "$PULL_ES" = "1" ]; then
+    scp -q -F "$SSH_CONFIG" "$SSH_HOST:$REMOTE_DIR/es_documents.jsonl.gz" "$OUT/" 2>/dev/null || true
+  fi
 else
  # ---------- 隧道模式（兜底）：本机经 SSH 隧道直连 ----------
  SECRET_FILE="secrets/release-sim/target_database_url"
@@ -159,6 +207,10 @@ PY
  docker run --rm --network host -e PGURL="$PGURL" \
    -v "$OUT:/out" -v "$OUT/export.sql:/scripts/export.sql:ro" postgres:16 \
    sh -c 'psql "$PGURL" -q -f /scripts/export.sql' >/dev/null
+  ES_KB_ARGS=""
+  for kb in "${KB_IDS[@]}"; do ES_KB_ARGS="$ES_KB_ARGS --kb $kb"; done
+  python3 "$SCRIPT_DIR/es_kb_export.py" --endpoint "http://${TUNNEL_HOST}:29200" --index "${ES_INDEX:-weknora_test_chunks}" \
+    $ES_KB_ARGS --output "$OUT/es_documents.jsonl.gz" --batch "$ES_BATCH"
 fi

 # ---------- ES 文档数（只读，失败不阻塞） ----------
@@ -174,11 +226,11 @@ if [ "$WITH_ES" = "1" ]; then
 fi

 # ---------- 清单 + 说明 ----------
-python3 - "$OUT" "$STAMP" "$EXECUTION" "$INCLUDE_CHUNKS" "$ES_SUMMARY" "$NOTE" "$REMOTE_DIR" "${KB_IDS[@]}" <<'PY'
+python3 - "$OUT" "$STAMP" "$EXECUTION" "$INCLUDE_CHUNKS" "$ES_SUMMARY" "$NOTE" "$CLOUD_REF" "${KB_IDS[@]}" <<'PY'
 import hashlib, json, pathlib, sys
 out = pathlib.Path(sys.argv[1]); stamp, execution = sys.argv[2], sys.argv[3]
 include_chunks = sys.argv[4] == "1"; es_summary = sys.argv[5].strip(); note = sys.argv[6]
-remote_dir = sys.argv[7]; kb_ids = sys.argv[8:]
+remote_ref = sys.argv[7]; kb_ids = sys.argv[8:]
 files = []
 for path in sorted(out.glob("*.jsonl")):
    digest = hashlib.sha256()
@@ -194,11 +246,24 @@ for path in sorted(out.glob("chunks.jsonl.gz")):
            digest.update(block)
    files.append({"file": path.name, "bytes": path.stat().st_size,
                  "sha256": digest.hexdigest(), "location": "local"})
+# ES 文档留档（本地有就按本地哈希；只在云主机则在 remote 段补）
+es_sidecar = out / "es_documents.jsonl.gz.manifest.json"
+es_counts = {}
+if es_sidecar.exists():
+    es_sidecar_data = json.loads(es_sidecar.read_text())
+    es_counts = {str(k): v for k, v in es_sidecar_data.get("knowledge_base_document_counts", {}).items()}
+    for path in sorted(out.glob("es_documents.jsonl.gz")):
+        digest = hashlib.sha256()
+        with path.open("rb") as handle:
+            for block in iter(lambda: handle.read(1 << 20), b""):
+                digest.update(block)
+        files.append({"file": path.name, "bytes": path.stat().st_size,
+                      "sha256": digest.hexdigest(), "location": "local"})
 remote_manifest = out / "REMOTE_MANIFEST.json"
 if remote_manifest.exists():
    for item in json.loads(remote_manifest.read_text()).get("host_files", []):
        if not any(existing["file"] == item["file"] for existing in files):
-            item["location"] = f"cloud:{remote_dir}"
+            item["location"] = f"cloud:{remote_ref}"
            files.append(item)
 manifest = {
    "kind": "weknora-target-kb-snapshot",
@@ -207,14 +272,16 @@ manifest = {
    "knowledge_base_ids": kb_ids,
    "profile": "kb-rows" + ("" if include_chunks else " (chunks excluded: 向量在 ES 侧，行可从源端重发)"),
    "covers": ["knowledge_bases", "knowledges", "knowledge_tags", "knowledge_tag_relations",
-               "wiki_pages", "wiki_folders", "resource_bindings", "resources"]
+               "wiki_pages", "wiki_folders", "resource_bindings", "resources",
+               "elasticsearch_documents"]
              + (["chunks"] if include_chunks else []),
-    "does_not_cover": ["Elasticsearch documents and vectors (see ES repository kb_backup, SLM kb-hourly/kb-daily)",
-                       "OSS object bodies (listed by reference in resources.jsonl)",
+    "does_not_cover": ["OSS object bodies (listed by reference in resources.jsonl; versioning protects overwrite/delete)",
+                       "Elasticsearch index settings/mappings (use ES repository kb_backup snapshots)",
                       "table structure / indexes", "rows outside the listed knowledge bases",
                       "whole-database recovery (use Aliyun RDS snapshot / PITR)"],
-    "es_document_counts": es_summary or None,
-    "remote_directory": remote_dir or None,
+    "es_document_counts": es_counts or (es_summary or None),
+    "es_documents_file": "es_documents.jsonl.gz" if (es_counts or (out / "es_documents.jsonl.gz.manifest.json").exists()) else None,
+    "remote_directory": remote_ref or None,
    "note": note or None,
    "files": files,
 }
@@ -229,9 +296,17 @@ cat >"$OUT/README.md" <<'MD'
 # 目标知识库快照（只读行级）

 ## 它是什么
-发布 / 撤权 / 回滚**之前**的目标库现场：KB 维度的 PG 行（含软删行），每张表一个
-JSONL（`to_jsonb(row)` 一行一条），附 `MANIFEST.json`（行数 / 字节 / sha256 / 覆盖范围 /
-位置）与 `export.sql`（导出用的原始查询）。生成过程只有 SELECT + `\copy`，不修改目标库。
+发布 / 撤权 / 回滚**之前**的目标库现场，两层一起留：
+
+1. **PG 行**：KB 维度的行（含软删行），每张表一个 JSONL（`to_jsonb(row)` 一行一条）；
+2. **ES 文档**：`es_documents.jsonl.gz`（一行一条 `{"_index","_id","_source"}`，含向量字段）
+   + `es_documents.jsonl.gz.manifest.json`（每库文档数 / 字节 / sha256）。
+
+附 `MANIFEST.json`（行数 / 字节 / sha256 / 覆盖范围 / 位置）与 `export.sql`（导出用的原始查询）。
+生成过程只有 SELECT / `\copy` / `_count` / `_search`，不修改目标库与索引。
+
+> 若 `MANIFEST.json` 里某文件的 `location` 是 `cloud:<主机>:<路径>`，说明字节留在目标主机上
+> （家用带宽拉不动），本地只有清单与哈希；同机执行时按该路径直接读。

 ## 它不是什么
 - 不含 Elasticsearch 文档与向量（见 ES 仓库 `kb_backup` 与 SLM 策略 `kb-hourly`/`kb-daily`）、