fix(library): 预备份硬门禁 + 崩溃接管 + 五项入口验收

为什么：父审指出预备份引用持久化失败仍继续写目标、接管依赖启动扫描且漏掉租约未过期、恢复失败被标成"无需恢复"、版本登记与作业状态分叉。 - BackupPrepared 返回 error，持久化失败即零写入；预备份同时登记 production_prebackup 版本。 - 接管改为按队列租约判定 + 运行期 30s 扫描 + 原子抢占 + 固定幂等键 + attempt 归零。 - 接管场景保留预备份、不再宣称"未写无需恢复"。 - 恢复/重放同一快照重新激活 production_effective（清 superseded_at）。 - finalize 续作只做校验+登记（finalize_only），不重写生产。 - 索引映射基线缺失时采用首次观察值并登记；索引校验加孤儿文档检查（按 knowledge_id）。真实验收（候选 API :18089 + 真实 PG/ES/MinIO）： A0 首次发布、A1 撤销、B 按版本恢复、C 写阶段 SIGKILL 接管、 D 跨 kind 互斥与多 worker 单次执行、E1 登记失败、E2 finalize 续作、E3 预备份失败零写入。
2026-10-28 19:11:37 +08:00
parent ca51cbe202
commit c72b71d78a
12 changed files with 529 additions and 118 deletions
@@ -1,5 +1,35 @@
 # Changelog

+## 开发快照（安全门禁 + 崩溃接管 + 五项真实入口验收）
+
+### Fixed
+
+- **预备份持久化成为硬门禁**：`BackupObserver.BackupPrepared` 现在返回 error，
+  引用/版本登记失败时 `Run` 直接返回失败且**一个字节都不写目标**（此前只写日志继续写）。
+- 预备份同时登记成 `production_prebackup` 版本，失败与成功路径都留证据。
+- 中断接管按**队列租约**判定（原来只看 state=running，worker 刚被杀而租约未过期时会漏掉）；
+  运行期每 30 秒扫一次，不再只在启动时扫一次。
+- 接管用一次条件更新原子抢占（`running→pending` 只有一个 worker 成功），队列作业 ID/幂等键固定，
+  并发接管只会得到同一条队列作业；重复接管时把 attempt 归零，否则 `attempt<max_attempts`
+  让作业永远不能被领取。
+- 接管场景下闸口/授权失败不再写"未写无需恢复"：保留既有预备份并保持目标占用。
+- 恢复/重放同一快照时重新激活 `production_effective` 版本（ON CONFLICT 清空 superseded_at），
+  否则页面显示"生产当前版本未知"而生产其实已恢复到该快照。
+- `finalize` 失败的续作只做"校验 + 登记"（finalize_only），不重写已生效的生产。
+- 生产索引映射基线缺失时，以"首次观察到的实际映射"为准并随预备份登记，之后改动会被拒绝。
+- 索引校验增加"孤儿文档"检查（按 knowledge_id 判定，父子分块不会误报）。
+
+### Verified（本机候选环境，真实 PG/ES/MinIO，经 HTTP API）
+
+- A0 首次发布：succeeded，4 文档/38 分块/21 资源/31 Wiki/122 索引，登记生效版本与预备份。
+- A1 撤销：succeeded，文档从 4 降到 3、分块 38→19、索引 122→58、绑定 4→3，其他租户 0 行。
+- B 按版本恢复：无授权 503；恢复授权 → 202 → succeeded，内容完整回到 4/38/122。
+- C 写阶段 SIGKILL：预备份已存在 → 杀掉 worker → 重启接管 → 复用同一预备份 → succeeded。
+- D 跨 kind 互斥：并发 sync+restore → 一个 202、一个 409；双 worker + 中途杀一个 → 仍单次成功。
+- E1 版本登记失败（真实 PG 触发器）：作业 failed@finalize，生产已写、无生效版本、预备份保留。
+- E2 finalize 续作：只做校验+登记 → succeeded，生效版本=目标快照，生产未被重写，重复续作同一作业。
+- E3 预备份登记失败（真实 PG 触发器）：作业 failed@prebackup，**生产零变化**。
+
 ## 开发快照（真实 Schema 对齐 + 入口级真实服务验收）

 ### Fixed
@@ -1,4 +1,4 @@
-// library-e2e-seed 从真实归档派生"第二版"快照，供候选验收跑新增/更新/撤销。
+// library-e2e-seed 从真实归档派生一份内部自洽的"第二版"快照，供候选验收跑更新/撤销。
 // 不模拟：所有部件都经 Archive.Put 真实重算哈希并 Seal，产物是合法归档。
 package main

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"io"
 	"log"
+	"regexp"
 	"time"

 	"knowledge-governance/internal/librarysnapshot"
@@ -16,12 +17,14 @@ import (

 const maxBlob = 512 << 20

-type row map[string]json.RawMessage
+var handlePattern = regexp.MustCompile(`resource://([A-Za-z0-9_-]{4,64})`)
+
+type row = map[string]json.RawMessage

 func main() {
 	root := flag.String("root", "", "归档根")
 	base := flag.String("base", "", "基线快照 ID")
-	mode := flag.String("mode", "update", "update|revoke|copy")
+	mode := flag.String("mode", "update", "update|revoke")
 	flag.Parse()
 	if *root == "" || *base == "" {
 		log.Fatal("root and base are required")
@@ -35,7 +38,79 @@ func main() {
 	if err != nil {
 		log.Fatal(err)
 	}
+
+	// 先把需要的表读进内存：派生规则要跨表看引用关系，不能边读边改。
+	rows := map[string][]row{}
+	for _, part := range manifest.Parts {
+		if part.Kind != "database" {
+			continue
+		}
+		reader, err := archive.OpenPart(part)
+		if err != nil {
+			log.Fatal(err)
+		}
+		data, err := io.ReadAll(reader)
+		reader.Close()
+		if err != nil {
+			log.Fatal(err)
+		}
+		var parsed []row
+		if err := json.Unmarshal(data, &parsed); err != nil {
+			log.Fatal(err)
+		}
+		rows[part.Name] = parsed
+	}
+
 	dropped := ""
+	// 撤销文档对应的分块 ID：索引文档一般只有 chunk_id，靠它才能把索引一起撤掉。
+	droppedChunks := map[string]bool{}
+	if *mode == "revoke" {
+		dropped = smallestID(rows["knowledges"])
+		for _, item := range rows["chunks"] {
+			if field(item, "knowledge_id") == dropped {
+				droppedChunks[field(item, "id")] = true
+			}
+		}
+	}
+	// 仍被引用的 handle：被撤销文档的分块与本体都不算引用。
+	referenced := map[string]bool{}
+	for table, items := range rows {
+		if table != "chunks" && table != "knowledges" {
+			continue
+		}
+		for _, item := range items {
+			if *mode == "revoke" && (field(item, "id") == dropped || field(item, "knowledge_id") == dropped) {
+				continue
+			}
+			for _, key := range []string{"content", "description", "title"} {
+				for _, handle := range extractHandles(field(item, key)) {
+					referenced[handle] = true
+				}
+			}
+		}
+	}
+	// 资源部件名是资源 ID：只有当它的 handle 仍被引用时才保留。
+	// 否则撤销后归档会带一个生产作用域算不到的孤儿资源，校验必然失败。
+	// 绑定关系同样是"活着"的依据：有些资源只在绑定里出现，正文并不引用。
+	bound := map[string]bool{}
+	for _, item := range rows["resource_bindings"] {
+		if owner := field(item, "owner_id"); owner == dropped {
+			continue
+		}
+		if resource := field(item, "resource_id"); resource != "" {
+			bound[resource] = true
+		}
+	}
+	keepResource := map[string]bool{}
+	if *mode == "revoke" {
+		for _, item := range rows["resources"] {
+			id := field(item, "id")
+			if referenced[field(item, "handle")] || bound[id] {
+				keepResource[id] = true
+			}
+		}
+	}
+
 	parts := make([]librarysnapshot.Part, 0, len(manifest.Parts))
 	for _, part := range manifest.Parts {
 		reader, err := archive.OpenPart(part)
@@ -48,28 +123,20 @@ func main() {
 			log.Fatal(err)
 		}
 		switch {
-		case part.Kind == "database" && part.Name == "knowledges" && (*mode == "update" || *mode == "revoke"):
-			data, dropped, err = mutateKnowledges(data, *mode)
-			if err != nil {
-				log.Fatal(err)
-			}
-		case part.Kind == "database" && part.Name == "chunks" && *mode == "revoke" && dropped != "":
-			data, err = dropRowsByKnowledge(data, dropped)
-			if err != nil {
-				log.Fatal(err)
-			}
-		case part.Kind == "index" && *mode == "revoke" && dropped != "":
-			data, err = dropRowsByKnowledge(data, dropped)
-			if err != nil {
-				log.Fatal(err)
-			}
-		case part.Kind == "database" && part.Name == "resource_bindings" && *mode == "revoke" && dropped != "":
-			// 撤销文档时必须连带撤销它的绑定：否则派生快照内部不一致
-			// （绑定指向已经不存在的文档），生产侧外键会拒绝，属于坏数据。
-			data, err = dropRowsByOwner(data, dropped)
-			if err != nil {
-				log.Fatal(err)
-			}
+		case *mode == "revoke" && part.Kind == "resource" && !keepResource[part.Name]:
+			continue
+		case part.Kind == "database" && part.Name == "knowledges":
+			data = mutateKnowledges(data, *mode, dropped)
+		case *mode == "revoke" && part.Kind == "database" && part.Name == "resources":
+			// 孤儿资源的行也要一起删：归档要求"每行资源都有对象部件"，
+			// 只删部件会得到一份自相矛盾的快照。
+			data = filterResources(data, keepResource)
+		case *mode == "revoke" && part.Kind == "database" && part.Name == "chunks":
+			data = dropByKnowledge(data, dropped)
+		case *mode == "revoke" && part.Kind == "database" && part.Name == "resource_bindings":
+			data = dropByOwner(data, dropped)
+		case *mode == "revoke" && part.Kind == "index":
+			data = dropIndexRows(data, dropped, droppedChunks)
 		}
 		sealed, err := archive.Put(bytes.NewReader(data))
 		if err != nil {
@@ -91,96 +158,144 @@ func main() {
 	if err != nil {
 		log.Fatal(err)
 	}
-	fmt.Printf(`{"snapshot_id":%q,"mode":%q,"base":%q,"dropped":%q}`, id, *mode, *base, dropped)
+	fmt.Printf(`{"snapshot_id":%q,"mode":%q,"base":%q,"dropped":%q,"parts":%d}`, id, *mode, *base, dropped, len(parts))
 }

-// mutateKnowledges：update 改一条标题（内容变更），revoke 删掉一条（撤销），返回被删 ID。
-func mutateKnowledges(data []byte, mode string) ([]byte, string, error) {
-	var rows []row
-	if err := json.Unmarshal(data, &rows); err != nil {
-		return nil, "", err
+func extractHandles(text string) []string {
+	matches := handlePattern.FindAllStringSubmatch(text, -1)
+	handles := make([]string, 0, len(matches))
+	for _, match := range matches {
+		handles = append(handles, match[1])
 	}
-	if len(rows) == 0 {
-		return data, "", nil
-	}
-	// 选稳定的一条：按 id 排序后的第一条，避免每次跑选中不同文档。
-	index := 0
-	for i := range rows {
-		if string(rows[i]["id"]) < string(rows[index]["id"]) {
-			index = i
+	return handles
+}
+
+func smallestID(items []row) string {
+	smallest := ""
+	for _, item := range items {
+		id := field(item, "id")
+		if id != "" && (smallest == "" || id < smallest) {
+			smallest = id
 		}
 	}
-	target := string(rows[index]["id"])
-	var targetID string
-	if err := json.Unmarshal(rows[index]["id"], &targetID); err != nil {
-		return nil, "", err
+	return smallest
+}
+
+func field(item row, key string) string {
+	raw, ok := item[key]
+	if !ok {
+		return ""
+	}
+	var value string
+	_ = json.Unmarshal(raw, &value)
+	return value
+}
+
+// mutateKnowledges：update 改一条标题（内容变更），revoke 删掉指定文档。
+func mutateKnowledges(data []byte, mode, dropped string) []byte {
+	var items []row
+	if err := json.Unmarshal(data, &items); err != nil {
+		log.Fatal(err)
+	}
+	target := smallestID(items)
+	for _, item := range items {
+		if mode == "update" && field(item, "id") == target {
+			title, _ := json.Marshal(field(item, "title") + "（E2E 更新）")
+			item["title"] = title
+		}
 	}
 	if mode == "revoke" {
-		rows = append(rows[:index], rows[index+1:]...)
-		return mustJSON(rows), targetID, nil
+		kept := make([]row, 0, len(items))
+		for _, item := range items {
+			if field(item, "id") == dropped {
+				continue
+			}
+			kept = append(kept, item)
+		}
+		return mustJSON(kept)
 	}
-	var title string
-	if raw, ok := rows[index]["title"]; ok {
-		_ = json.Unmarshal(raw, &title)
-	}
-	updated, err := json.Marshal(title + "（E2E 更新）")
-	if err != nil {
-		return nil, "", err
-	}
-	rows[index]["title"] = updated
-	_ = target
-	return mustJSON(rows), "", nil
+	return mustJSON(items)
 }

-// dropRowsByKnowledge 删除 knowledge_id 指向被撤销文档的行（chunks/索引）。
-func dropRowsByKnowledge(data []byte, knowledgeID string) ([]byte, error) {
-	var rows []row
-	if err := json.Unmarshal(data, &rows); err != nil {
-		return nil, err
+// filterResources 只保留仍然被引用的资源行。
+func filterResources(data []byte, keep map[string]bool) []byte {
+	var items []row
+	if err := json.Unmarshal(data, &items); err != nil {
+		log.Fatal(err)
 	}
-	kept := make([]row, 0, len(rows))
-	for _, item := range rows {
-		var owner string
-		for _, key := range []string{"knowledge_id", "knowledgeID"} {
-			if raw, ok := item[key]; ok {
-				_ = json.Unmarshal(raw, &owner)
-				break
+	kept := make([]row, 0, len(items))
+	for _, item := range items {
+		if keep[field(item, "id")] {
+			kept = append(kept, item)
+		}
+	}
+	return mustJSON(kept)
+}
+
+// dropIndexRows 既按 knowledge_id 也按 chunk_id 删索引文档：
+// 真实索引文档通常只带 chunk_id，漏了它就会留下"孤儿向量"。
+func dropIndexRows(data []byte, knowledgeID string, chunkIDs map[string]bool) []byte {
+	var items []row
+	if err := json.Unmarshal(data, &items); err != nil {
+		log.Fatal(err)
+	}
+	kept := make([]row, 0, len(items))
+	for _, item := range items {
+		// 归档索引行是 {ID, Source} 包装；真正字段在 Source 里。
+		source := item
+		if raw, ok := item["Source"]; ok {
+			var inner row
+			if json.Unmarshal(raw, &inner) == nil {
+				source = inner
 			}
 		}
-		if owner == knowledgeID {
+		if field(source, "knowledge_id") == knowledgeID {
+			continue
+		}
+		if chunkIDs[field(source, "chunk_id")] || chunkIDs[field(source, "id")] {
 			continue
 		}
 		kept = append(kept, item)
 	}
-	return mustJSON(kept), nil
+	return mustJSON(kept)
 }

-// dropRowsByOwner 删除 owner_id 指向被撤销文档的绑定行。
-func dropRowsByOwner(data []byte, ownerID string) ([]byte, error) {
-	var rows []row
-	if err := json.Unmarshal(data, &rows); err != nil {
-		return nil, err
+func dropByKnowledge(data []byte, knowledgeID string) []byte {
+	var items []row
+	if err := json.Unmarshal(data, &items); err != nil {
+		log.Fatal(err)
 	}
-	kept := make([]row, 0, len(rows))
-	for _, item := range rows {
-		var owner string
-		if raw, ok := item["owner_id"]; ok {
-			_ = json.Unmarshal(raw, &owner)
-		}
-		if owner == ownerID {
+	kept := make([]row, 0, len(items))
+	for _, item := range items {
+		if field(item, "knowledge_id") == knowledgeID {
 			continue
 		}
 		kept = append(kept, item)
 	}
-	return mustJSON(kept), nil
+	return mustJSON(kept)
+}
+
+func dropByOwner(data []byte, ownerID string) []byte {
+	var items []row
+	if err := json.Unmarshal(data, &items); err != nil {
+		log.Fatal(err)
+	}
+	kept := make([]row, 0, len(items))
+	for _, item := range items {
+		if field(item, "owner_id") == ownerID {
+			continue
+		}
+		kept = append(kept, item)
+	}
+	return mustJSON(kept)
 }

 func rowCount(data []byte) (int64, error) {
-	var rows []json.RawMessage
-	if err := json.Unmarshal(data, &rows); err != nil {
+	var items []json.RawMessage
+	if err := json.Unmarshal(data, &items); err != nil {
 		return 0, err
 	}
-	return int64(len(rows)), nil
+	return int64(len(items)), nil
 }

 func mustJSON(value any) []byte {
@@ -21,6 +21,9 @@ type libraryReleasePayload struct {
 	LibraryJobID     string `json:"library_job_id"`
 	Scope            string `json:"scope"`
 	TargetSnapshotID string `json:"target_snapshot_id"`
+	// FinalizeOnly：只做收尾（校验 + 登记生效版本），不重写目标。
+	// 用于"写入已完成但版本登记失败"的续作，避免把已生效的生产再写一遍。
+	FinalizeOnly bool `json:"finalize_only,omitempty"`
 }

 // runLibraryRelease 执行"闸口 → 授权 → 预备份 → 四类写入 → 校验"的完整链路。
@@ -33,7 +36,7 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a
 	}
 	now := time.Now().UTC()
 	_ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobRunning, "prebackup", "", "", "", "", now)
-	recorder := libraryStageRecorder{store: store, jobID: payload.LibraryJobID}
+	recorder := libraryStageRecorder{store: store, jobID: payload.LibraryJobID, scope: payload.Scope}

 	// 崩溃接管：作业里已经持久化过预备份快照，就直接复用它继续，
 	// 绝不重新备份——重新备份会把"半写状态"当成新基线，覆盖真正可回滚的那份。
@@ -53,8 +56,14 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a
 	if len(gateReasons) > 0 {
 		message := "生产" + kindLabel(kind) + "被闸口拒绝：" + strings.Join(gateReasons, "；")
 		recorder.Stage(ctx, "prebackup", "failed", message)
+		// 接管场景下生产可能已经是半写状态：这时"无需恢复"是错的，
+		// 必须把既有预备份作为恢复基线留住，并保持目标被占用直到处理完。
+		recovery := "未对生产执行任何写入，无需恢复"
+		if resumed {
+			recovery = "生产可能处于半写状态，必须先按预备份 " + previous.SnapshotID + " 恢复后再重试"
+		}
 		_ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobFailed, "prebackup", message,
-			"未对生产执行任何写入，无需恢复", "", "", time.Now().UTC())
+			recovery, previous.SnapshotID, "", time.Now().UTC())
 		return errors.New(message)
 	}

@@ -66,24 +75,68 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a
 	if err != nil || !found {
 		return failLibraryJob(ctx, store, recorder, payload.LibraryJobID, "没有与目标快照匹配的有效发布授权", "prebackup")
 	}
-	// 上一次生产生效版本登记的实际索引映射：这是本生产索引的基线。
+	// 生产索引映射基线：优先当前生效版本，其次任何一次成功预备份（它也记录了生产实际映射）。
+	// 只认"生效版本"会在恢复把版本 supersede 掉之后误判"索引被改动"。
 	knownMapping := ""
 	if versions, versionErr := store.LibraryVersions(ctx, payload.Scope, 50); versionErr == nil {
 		for _, item := range versions {
-			if item.Kind != governance.LibraryVersionProductionEffective || item.SupersededAt != nil {
-				continue
+			if item.Kind == governance.LibraryVersionProductionEffective && item.SupersededAt == nil {
+				if raw, ok := item.Evidence["index_mapping_sha256"].(string); ok && raw != "" {
+					knownMapping = raw
+					break
+				}
 			}
-			if raw, ok := item.Evidence["index_mapping_sha256"].(string); ok {
-				knownMapping = raw
+		}
+		if knownMapping == "" {
+			for _, item := range versions {
+				if raw, ok := item.Evidence["index_mapping_sha256"].(string); ok && raw != "" {
+					knownMapping = raw
+					break
+				}
 			}
-			break
 		}
 	}
 	target, backups, wireErr := buildLibraryProductionStack(ctx, config, payload.Scope, payload.TargetSnapshotID, knownMapping)
+	if wireErr == nil {
+		if digest, mapErr := target.Search().MappingSHA256(ctx); mapErr == nil {
+			recorder.mappingDigest = digest
+		}
+	}
 	if wireErr != nil {
 		return failLibraryJob(ctx, store, recorder, payload.LibraryJobID, wireErr.Error(), "prebackup")
 	}
 	defer target.Close()
+	if payload.FinalizeOnly {
+		// 写入已完成、只是版本登记失败：只做校验 + 登记，绝不重写生产。
+		recorder.Stage(ctx, "finalize", "running", "续作收尾：先校验生产与目标快照一致")
+		if verifyErr := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: payload.TargetSnapshotID}); verifyErr != nil {
+			message := "生产与目标快照不一致，不能登记为生效版本：" + verifyErr.Error()
+			recorder.Stage(ctx, "finalize", "failed", message)
+			_ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobFailed, "finalize", message,
+				"生产可能处于半写状态，需按预备份恢复后重试", previous.SnapshotID, payload.TargetSnapshotID, time.Now().UTC())
+			return errors.New(message)
+		}
+		mappingDigest, _ := target.Search().MappingSHA256(ctx)
+		if err := store.RecordLibraryVersion(ctx, governance.LibraryVersion{
+			ID:    governance.LibraryVersionID(governance.LibraryVersionProductionEffective, payload.Scope, payload.TargetSnapshotID),
+			Scope: payload.Scope, Kind: governance.LibraryVersionProductionEffective,
+			SnapshotID: payload.TargetSnapshotID, SourceJobID: payload.LibraryJobID, CreatedBy: "worker",
+			Evidence: map[string]any{"backup_snapshot": previous.SnapshotID, "action": "finalize_only",
+				"index_mapping_sha256": mappingDigest},
+			CreatedAt: time.Now().UTC(),
+		}); err != nil {
+			message := "收尾续作仍无法登记生效版本：" + err.Error()
+			recorder.Stage(ctx, "finalize", "failed", message)
+			_ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobFailed, "finalize", message,
+				"预备份 "+previous.SnapshotID+" 保留，可再次续作", previous.SnapshotID, payload.TargetSnapshotID, time.Now().UTC())
+			return errors.New(message)
+		}
+		recorder.Stage(ctx, "finalize", "succeeded", "已登记生效版本（仅收尾，未重写生产）")
+		_ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobSucceeded, "finalize", "", "",
+			previous.SnapshotID, payload.TargetSnapshotID, time.Now().UTC())
+		return nil
+	}
+
 	var backupSource libraryrelease.BackupStore = backups
 	if resumed {
 		backupSource = fixedBackupStore{ref: libraryrelease.BackupRef{ID: "backup-" + previous.SnapshotID,
@@ -144,8 +197,12 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a
 			outcome.Error, outcome.Rollback, "", "", finishedAt)
 		return fmt.Errorf("%s（%s）", outcome.Error, outcome.Rollback)
 	default:
+		recovery := outcome.Recovery
+		if recovery == "" {
+			recovery = "未对生产执行任何写入，无需恢复"
+		}
 		_ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobFailed, "prebackup",
-			outcome.Error, "未对生产执行任何写入，无需恢复", "", "", finishedAt)
+			outcome.Error, recovery, outcome.Backup.SnapshotID, "", finishedAt)
 		return errors.New(outcome.Error)
 	}
 }
@@ -168,9 +225,13 @@ func resumeInterruptedLibraryJobs(ctx context.Context, store *postgresstore.Stor
 			failed++
 			continue
 		}
-		if err := store.AdvanceLibraryJob(ctx, job.ID, governance.LibraryJobPending, "prebackup", "", "",
-			"", "", now); err != nil {
-			log.Printf("library job %s cannot be marked for resume: %v", job.ID, err)
+		// 原子抢占：多个 worker 同时接管也只有一个能把 running 改成 pending。
+		claimed, claimErr := store.ClaimLibraryJobForResume(ctx, job.ID, now)
+		if claimErr != nil {
+			log.Printf("library job %s cannot be claimed for resume: %v", job.ID, claimErr)
+			continue
+		}
+		if !claimed {
 			continue
 		}
 		queueType := librarySyncJobType
@@ -179,15 +240,35 @@ func resumeInterruptedLibraryJobs(ctx context.Context, store *postgresstore.Stor
 		}
 		payload, _ := json.Marshal(map[string]string{"library_job_id": job.ID, "scope": job.Scope,
 			"target_snapshot_id": job.TargetSnapshotID})
+		// 队列作业 ID 与幂等键都按作业固定：重复接管/重复点击只会得到同一个队列作业。
 		queueJob, err := store.Enqueue(ctx, jobs.EnqueueRequest{
-			ID: governance.QueueJobIDFor(job.ID) + "-r" + now.Format("050405"), Type: queueType, Payload: payload,
+			ID: governance.QueueJobIDFor(job.ID), Type: queueType, Payload: payload,
 			AvailableAt: now, MaxAttempts: 1,
-			IdempotencyKey: "library.resume.queue:" + job.ID + ":" + now.Format("150405"),
+			IdempotencyKey: "library.resume.queue:" + job.ID,
 		}, now)
 		if err != nil || queueJob == nil {
+			// 队列作业 ID 是确定性的：并发接管时另一个 worker 已经建好了，
+			// 这里按"已存在"处理并继续接管，不去覆盖别人的队列行。
+			if err != nil && strings.Contains(err.Error(), "duplicate key") {
+				queueID := governance.QueueJobIDFor(job.ID)
+				if resetErr := store.ResetQueueJobForResume(ctx, queueID, now); resetErr != nil {
+					log.Printf("library job %s resume reset existing queue failed: %v", job.ID, resetErr)
+					continue
+				}
+				if attachErr := store.AttachLibraryQueueJob(ctx, job.ID, queueID, now); attachErr != nil {
+					log.Printf("library job %s resume attach existing queue failed: %v", job.ID, attachErr)
+				} else {
+					resumed++
+				}
+				continue
+			}
 			log.Printf("library job %s resume enqueue failed: %v", job.ID, err)
 			continue
 		}
+		if resetErr := store.ResetQueueJobForResume(ctx, queueJob.ID, now); resetErr != nil {
+			log.Printf("library job %s resume reset queue failed: %v", job.ID, resetErr)
+			continue
+		}
 		if attachErr := store.AttachLibraryQueueJob(ctx, job.ID, queueJob.ID, now); attachErr != nil {
 			log.Printf("library job %s resume attach failed: %v", job.ID, attachErr)
 			continue
@@ -205,12 +286,25 @@ func (s fixedBackupStore) Prepare(context.Context, string, libraryrelease.Curren
 	return s.ref, nil
 }

-// BackupPrepared 让 recorder 兼作 BackupObserver，把预备份快照立刻写进作业。
-func (r libraryStageRecorder) BackupPrepared(ctx context.Context, backup libraryrelease.BackupRef) {
+// BackupPrepared 把预备份写进作业**并**登记成版本。任何一步失败都返回错误：
+// 上游据此在写目标前中止，保证"没有可恢复基线就不写"。
+func (r libraryStageRecorder) BackupPrepared(ctx context.Context, backup libraryrelease.BackupRef) error {
+	now := time.Now().UTC()
 	if err := r.store.AdvanceLibraryJob(ctx, r.jobID, governance.LibraryJobRunning, "", "", "",
-		backup.SnapshotID, "", time.Now().UTC()); err != nil {
-		log.Printf("library job %s cannot persist backup ref: %v", r.jobID, err)
+		backup.SnapshotID, "", now); err != nil {
+		return fmt.Errorf("持久化预备份引用失败: %w", err)
 	}
+	if err := r.store.RecordLibraryVersion(ctx, governance.LibraryVersion{
+		ID:    governance.LibraryVersionID(governance.LibraryVersionProductionPrebackup, r.scope, backup.SnapshotID),
+		Scope: r.scope, Kind: governance.LibraryVersionProductionPrebackup, SnapshotID: backup.SnapshotID,
+		SourceJobID: r.jobID, CreatedBy: "worker",
+		Evidence: map[string]any{"backup_snapshot": backup.SnapshotID, "backup_bytes": backup.Bytes,
+			"job": r.jobID, "index_mapping_sha256": r.mappingDigest},
+		CreatedAt: now,
+	}); err != nil {
+		return fmt.Errorf("登记预备份版本失败: %w", err)
+	}
+	return nil
 }

 func failLibraryJob(ctx context.Context, store *postgresstore.Store, recorder libraryStageRecorder, jobID, message, stage string) error {
@@ -85,7 +85,9 @@ func buildLibraryProductionStack(ctx context.Context, config appconfig.Config, s
 		// 它才是这份生产索引的既有事实；快照契约只管"来自哪份数据"。
 		baseline := knownMapping
 		if baseline == "" {
-			baseline = contract.IndexMappingSHA256
+			// 没有历史基线时（首次接管这套生产索引），把当前实际映射当作基线并随
+			// 预备份登记；之后任何改动都会与登记值不符而被拒绝。
+			baseline = mapping
 		}
 		if baseline != "" && mapping != baseline {
 			current, currentErr := target.Current(ctx)
@@ -31,6 +31,8 @@ type libraryStageRecorder struct {
 	store *postgresstore.Store
 	jobID string
 	scope string
+	// mappingDigest 是本次读到的生产索引映射，随预备份一起登记成版本证据。
+	mappingDigest string
 }

 func (r libraryStageRecorder) Stage(ctx context.Context, stage, state, detail string) {
@@ -80,6 +82,24 @@ func run(ctx context.Context, getenv func(string) string) error {
 	} else if resumed > 0 || failed > 0 {
 		log.Printf("library job resume: %d resumed, %d failed without backup", resumed, failed)
 	}
+	// 运行期也要持续接管：只在启动扫一次会漏掉"worker 在两次重启之间被杀"的情况。
+	go func() {
+		ticker := time.NewTicker(30 * time.Second)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				resumed, failed, err := resumeInterruptedLibraryJobs(ctx, store, config)
+				if err != nil {
+					log.Printf("library job takeover sweep: %v", err)
+				} else if resumed > 0 || failed > 0 {
+					log.Printf("library job takeover sweep: %d resumed, %d failed without backup", resumed, failed)
+				}
+			}
+		}
+	}()
 	if repaired, reconcileErr := store.ReconcileLibraryJobs(context.Background(), time.Now().UTC()); reconcileErr != nil {
 		log.Printf("library job reconcile failed: %v", reconcileErr)
 	} else if repaired > 0 {
@@ -13,8 +13,11 @@ import (
 // 生产发布授权：管理员把"目标 + 快照 + 参数摘要 + 有效期"钉死。
 // 换快照、换目标或改参数都会得到不同摘要，旧授权自然失效；到期或被撤销也不能用。
 type libraryAuthorizationRequest struct {
-	Scope            string `json:"scope"`
-	SnapshotID       string `json:"snapshot_id"`
+	Scope      string `json:"scope"`
+	SnapshotID string `json:"snapshot_id"`
+	// Action 决定授权的动作：sync（默认）或 restore。两者摘要不同，
+	// 同步授权不能当恢复授权用，反之亦然。
+	Action           string `json:"action"`
 	ExpiresInMinutes int    `json:"expires_in_minutes"`
 }

@@ -69,14 +72,28 @@ func libraryAuthorizationCreate(w http.ResponseWriter, r *http.Request, config C
 		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "expires_in_minutes must be between 5 and 1440"})
 		return
 	}
-	digest, err := librarySyncParamsDigest(request.Scope, request.SnapshotID)
+	action := strings.TrimSpace(request.Action)
+	if action == "" {
+		action = "sync"
+	}
+	var digest string
+	var err error
+	switch action {
+	case "sync":
+		digest, err = librarySyncParamsDigest(request.Scope, request.SnapshotID)
+	case "restore":
+		digest, err = libraryRestoreParamsDigest(request.Scope, request.SnapshotID)
+	default:
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "action must be sync or restore"})
+		return
+	}
 	if err != nil {
 		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot digest params"})
 		return
 	}
 	now := time.Now().UTC()
 	item := governance.LibraryReleaseAuthorization{
-		ID:           "libauth-" + request.Scope + "-" + now.Format("20060102T150405.000000"),
+		ID:           "libauth-" + action + "-" + request.Scope + "-" + now.Format("20060102T150405.000000000"),
 		Scope:        request.Scope,
 		SnapshotID:   request.SnapshotID,
 		ParamsDigest: digest,
@@ -597,7 +597,8 @@ func librarySyncEnqueue(w http.ResponseWriter, r *http.Request, config Config) {
 		TargetSnapshotID: request.Target, Actor: actor,
 	}, governance.SyncStages(), now)
 	if errors.Is(err, governance.ErrLibraryJobActive) {
-		writeJSON(w, http.StatusConflict, map[string]any{"code": "library_sync_active", "error": "该目标已有在飞的同步作业"})
+		writeJSON(w, http.StatusConflict, map[string]any{"code": "library_sync_active",
+			"error": "该目标已有在飞的生产写作业（同步与恢复互斥）"})
 		return
 	}
 	if errors.Is(err, governance.ErrLibraryJobIdempotencyConflict) {
@@ -228,8 +228,13 @@ func libraryJobResume(w http.ResponseWriter, r *http.Request, config Config) {
 		writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "resumed_from": sourceID, "reused": true})
 		return
 	}
-	payload, _ := json.Marshal(map[string]string{"library_job_id": job.ID, "scope": source.Scope,
-		"target_snapshot_id": source.TargetSnapshotID})
+	// 写入已完成、只差登记版本的情况：续作只做幂等收尾，不重写生产。
+	body := map[string]any{"library_job_id": job.ID, "scope": source.Scope,
+		"target_snapshot_id": source.TargetSnapshotID}
+	if source.Stage == "finalize" && source.SnapshotID != "" {
+		body["finalize_only"] = true
+	}
+	payload, _ := json.Marshal(body)
 	queueJob, err := config.Store.Enqueue(r.Context(), jobs.EnqueueRequest{
 		ID: governance.QueueJobIDFor(job.ID), Type: queueType, Payload: payload, AvailableAt: now,
 		MaxAttempts: 1, IdempotencyKey: "library.resume.queue:" + job.ID,
@@ -93,7 +93,9 @@ type Recorder interface {
 // BackupObserver 让调用方在预备份完成时立刻持久化快照身份：
 // worker 被杀后重启必须能接管这份预备份，而不是重新备份覆盖它。
 type BackupObserver interface {
-	BackupPrepared(ctx context.Context, backup BackupRef)
+	// 必须返回错误：预备份引用没能持久化时，worker 被杀就没有恢复基线，
+	// 这种情况下一个字节都不许写目标。调用方据此直接失败。
+	BackupPrepared(ctx context.Context, backup BackupRef) error
 }

 // FinalizeStages 是"写入完成之后的收尾"：先登记权威版本，再把作业标成功。
@@ -117,6 +119,7 @@ type Outcome struct {
 	FailedAt Stage     `json:"failed_at,omitempty"`
 	Error    string    `json:"error,omitempty"`
 	Rollback string    `json:"rollback,omitempty"`
+	Recovery string    `json:"recovery,omitempty"`
 	Finished time.Time `json:"finished"`
 }

@@ -178,7 +181,14 @@ func Run(ctx context.Context, request Request, target Target, backups BackupStor
 	}
 	outcome.Backup = backup
 	if observer, ok := recorder.(BackupObserver); ok {
-		observer.BackupPrepared(ctx, backup)
+		if err := observer.BackupPrepared(ctx, backup); err != nil {
+			// 硬门禁：持久化失败 = 没有可恢复基线 = 绝不写目标。
+			outcome.State = "failed"
+			outcome.Error = "预备份已生成但引用无法持久化，未对生产执行任何写入：" + err.Error()
+			outcome.Recovery = "预备份 " + backup.SnapshotID + " 已落盘，先修复治理库再重试"
+			recorder.Stage(ctx, "prebackup", "failed", outcome.Error)
+			return outcome
+		}
 	}
 	recorder.Stage(ctx, "prebackup", "succeeded", "预备份 "+backup.ID)

@@ -0,0 +1,38 @@
+package libraryrelease
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+// failingObserver 模拟"预备份引用无法持久化"：真实场景是治理库写入失败。
+type failingObserver struct{ stageLog }
+
+func (f *failingObserver) BackupPrepared(context.Context, BackupRef) error {
+	return errors.New("persist backup ref failed")
+}
+
+// 预备份引用持久化失败时必须零写入：没有可恢复基线就不许碰目标。
+func TestRunRefusesToWriteWhenBackupReferenceCannotBePersisted(t *testing.T) {
+	target := &fakeTarget{}
+	backups := &fakeBackups{}
+	recorder := &failingObserver{}
+	outcome := Run(context.Background(), request(time.Now()), target, backups, recorder)
+	if outcome.State == "succeeded" {
+		t.Fatalf("state = %s, must not succeed without a recoverable backup", outcome.State)
+	}
+	if len(target.applied) != 0 {
+		t.Fatalf("applied stages = %v, must be zero when the backup reference is lost", target.applied)
+	}
+	if target.restored != 0 {
+		t.Fatalf("restore must not run when nothing was written (restored=%d)", target.restored)
+	}
+	if !recorder.has("prebackup:failed") {
+		t.Fatalf("stage log = %v, want prebackup:failed", recorder.entries)
+	}
+	if outcome.Recovery == "" {
+		t.Fatal("failed outcome must explain how to recover")
+	}
+}
@@ -239,7 +239,13 @@ WHERE scope=$1 AND kind='production_effective' AND superseded_at IS NULL`, versi
 	if _, err := tx.Exec(ctx, `
 INSERT INTO library_versions (id, scope, kind, snapshot_id, version_label, source_job_id, evidence, created_by, created_at)
 VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
-ON CONFLICT (id) DO NOTHING`,
+ON CONFLICT (id) DO UPDATE SET
+    -- 恢复/重放同一个快照时必须把它重新"激活"：否则它还是被 supersede 的旧行，
+    -- 页面会显示"生产当前版本未知"，而实际生产已经恢复到这个快照。
+    superseded_at = NULL,
+    source_job_id = EXCLUDED.source_job_id,
+    evidence = EXCLUDED.evidence,
+    created_at = EXCLUDED.created_at`,
 		version.ID, version.Scope, string(version.Kind), version.SnapshotID, version.VersionLabel,
 		version.SourceJobID, evidence, version.CreatedBy, version.CreatedAt.UTC()); err != nil {
 		return fmt.Errorf("record library version: %w", err)
@@ -250,10 +256,19 @@ ON CONFLICT (id) DO NOTHING`,
 // InterruptedLibraryJobs 找出"业务作业 running、但队列里已没有活跃作业"的作业：
 // worker 被杀/重启后它们需要被接管，而不是被当成失败丢掉已完成的预备份。
 func (s *Store) InterruptedLibraryJobs(ctx context.Context) ([]governance.LibraryJob, error) {
+	// lease 过期才算中断：worker 刚被杀但 lease 未到期时，队列里还是 running，
+	// 直接判"没中断"会让作业永远悬着；所以这里按 lease_until 判断。
 	rows, err := s.db.Query(ctx, `
-SELECT id FROM library_jobs lj
-WHERE lj.state='running' AND lj.kind IN ('sync','restore')
-  AND NOT EXISTS (SELECT 1 FROM jobs j WHERE j.id=lj.queue_job_id AND j.state IN ('pending','running','retry'))
+SELECT lj.id FROM library_jobs lj
+LEFT JOIN jobs j ON j.id=lj.queue_job_id
+WHERE lj.kind IN ('sync','restore')
+  AND (
+    -- 业务作业在跑、队列已经没了/停了/租约过期
+    (lj.state='running' AND (j.id IS NULL OR j.state IN ('failed','completed','cancelled','dead')
+        OR (j.state IN ('pending','running','retry') AND j.lease_until IS NOT NULL AND j.lease_until < now())))
+    -- 业务作业已被前一轮接管置为 pending，但队列作业的租约也没了（worker 又死了）
+    OR (lj.state='pending' AND j.state='running' AND j.lease_until IS NOT NULL AND j.lease_until < now())
+  )
 ORDER BY lj.created_at`)
 	if err != nil {
 		return nil, fmt.Errorf("list interrupted library jobs: %w", err)
@@ -281,6 +296,28 @@ ORDER BY lj.created_at`)
 	return jobs, nil
 }

+// ResetQueueJobForResume 把中断/失败过的队列作业重新排队：attempt 必须归零，
+// 否则 attempt<max_attempts 永远不成立，作业再也不会被任何 worker 领取。
+func (s *Store) ResetQueueJobForResume(ctx context.Context, queueJobID string, now time.Time) error {
+	if _, err := s.db.Exec(ctx, `UPDATE jobs
+SET state='pending', attempt=0, lease_owner=NULL, lease_until=NULL, available_at=$2, updated_at=$2
+WHERE id=$1 AND state IN ('running','retry','failed','pending')`, queueJobID, now); err != nil {
+		return fmt.Errorf("reset queue job for resume: %w", err)
+	}
+	return nil
+}
+
+// ClaimLibraryJobForResume 用一次条件更新把中断作业抢过来：只有把 running 改成 pending
+// 的那一个 worker 拿到 true，多个 worker 同时重启也只会有一个执行（fencing）。
+func (s *Store) ClaimLibraryJobForResume(ctx context.Context, jobID string, now time.Time) (bool, error) {
+	tag, err := s.db.Exec(ctx, `UPDATE library_jobs SET state='pending', updated_at=$2
+WHERE id=$1 AND state IN ('running','pending')`, jobID, now)
+	if err != nil {
+		return false, fmt.Errorf("claim interrupted library job: %w", err)
+	}
+	return tag.RowsAffected() == 1, nil
+}
+
 // ReconcileLibraryJobs 收拾"业务作业还在飞、队列里已经没有对应作业"的孤儿：
 // 它们会永久占用 (scope,kind) 唯一索引，让同一目标再也发不出作业。
 // 只标失败并写明恢复责任，不猜结果、不重放。
@@ -128,6 +128,43 @@ func archivedTableContent(archive *librarysnapshot.Archive, part librarysnapshot
 	return int64(len(digests)), combineRowDigests(digests), columns, nil
 }

+// orphanIndexDocuments 找出"索引里有、生产分块里没有"的文档（按 chunk_id 关联）。
+func (t *ProductionLibraryTarget) orphanIndexDocuments(ctx context.Context, documents []SearchDocument) []string {
+	if len(documents) == 0 {
+		return nil
+	}
+	// 按文档归属判断而不是按 chunk_id：真实 WeKnora 是父子分块，
+	// 索引里的子块 ID 本来就不在 chunks 表里，按 chunk_id 判会误报孤儿。
+	rows, err := t.pool.Query(ctx, `SELECT id FROM knowledges WHERE tenant_id=$1 AND knowledge_base_id=$2 AND deleted_at IS NULL`,
+		t.scope.TenantID, t.scope.KnowledgeBaseID)
+	if err != nil {
+		return nil
+	}
+	defer rows.Close()
+	known := map[string]bool{}
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			return nil
+		}
+		known[id] = true
+	}
+	orphans := make([]string, 0, 4)
+	for _, document := range documents {
+		var source struct {
+			ChunkID     string `json:"chunk_id"`
+			KnowledgeID string `json:"knowledge_id"`
+		}
+		if json.Unmarshal(document.Source, &source) != nil || source.KnowledgeID == "" {
+			continue
+		}
+		if !known[source.KnowledgeID] {
+			orphans = append(orphans, document.ID)
+		}
+	}
+	return orphans
+}
+
 // firstIndexDifference 返回第一处不一致的文档 ID 与字段，便于定位。
 func firstIndexDifference(archived, live []SearchDocument) string {
 	liveByID := make(map[string]json.RawMessage, len(live))
@@ -212,6 +249,11 @@ func (t *ProductionLibraryTarget) verifyIndexDigest(ctx context.Context, archive
 		if err != nil {
 			return fmt.Errorf("读取生产索引失败: %w", err)
 		}
+		// 孤儿向量：索引里有、数据库分块里没有的文档必须报出来。
+		// 只比"归档 vs 生产"会漏掉两边一致但都错的情况。
+		if orphans := t.orphanIndexDocuments(ctx, snapshot.Documents); len(orphans) > 0 {
+			return fmt.Errorf("生产索引存在 %d 条孤儿文档（所属文档已不在数据库），首个 %s", len(orphans), orphans[0])
+		}
 		if len(snapshot.Documents) != len(archived) || snapshot.DocumentsSHA256 != wantDigest {
 			// 摘要不一致时给出第一处差异文档，运维才能定位是"少了/多了/被改了"。
 			return fmt.Errorf("生产索引与快照不一致：文档 %d/%d，首个差异 %s",