fix(library): 对齐真实 WeKnora Schema + 入口级真实服务验收

为什么：候选环境改用真实 WeKnora 表结构后，暴露出多处"按猜的列名/形状"的假设，
以及若干 varchar 长度、对象定位符、空索引基线问题，导致真实链路跑不通。

- knowledge_tag_relations 无 tenant_id/KB 列 → 归属走父表；resources 作用域=绑定+chunk 引用。
- 写入顺序按真实外键；无代理键表用内容摘要核对。
- 队列作业 ID / 版本记录 ID 改固定长度摘要（列宽 varchar(36)/varchar(64)）。
- 对象定位符拆 storage://<backend>/oss://<bucket>/<key>；provider 可配置。
- 空索引基线全链路一致（Current/Export/Backup/Archive），非空仍严格比对映射。
- finalize 阶段：版本登记失败不报成功；崩溃接管复用已持久化预备份。
- 迁移 000069：同一目标的 sync/restore 跨 kind 互斥。

CHANGELOG.md

@@ -1,5 +1,46 @@
 # Changelog
 
+## 开发快照（真实 Schema 对齐 + 入口级真实服务验收）
+
+### Fixed
+
+- 真实 Schema 对齐：`knowledge_tag_relations` 没有 tenant_id/knowledge_base_id（按真实库确认），
+  归属改为走父表 `knowledges`；`resources` 的作用域改为"绑定 + 被本库 chunk 正文 resource:// 引用"，
+  与归档侧收集资源部件的规则一致（此前会把被引用的资源算成多余）。
+- 表写入顺序按真实外键调整（`resources` 先于 `resource_bindings`）。
+- `knowledge_tag_relations` 等无代理键的表改用内容摘要核对，不再要求不存在的 id 列。
+- 队列作业 ID 用固定长度摘要（`jobs.id` 是 varchar(36)，`<作业号>-queue` 会超长）。
+- 生产版本记录 ID 用稳定短摘要（`library_versions.id` 是 varchar(64)，拼接完整快照 ID 会超长）。
+- 对象路径：真实 `resources.physical_path` 是 `storage://<backend>/oss://<bucket>/<key>`，
+  写入/读取都先拆成裸 key；对象存储 provider 改为可配置（默认 s3），必须与生产 `resources.provider` 一致。
+- 生产索引不存在/为空时按"空基线"处理（Current/Export/Backup/Archive 全链路一致），
+  非空时仍然严格比对映射，避免把索引丢失当成正常。
+- 索引校验按写入时的规范化比较（写入会补齐 tenant_id/knowledge_base_id），
+  不一致时给出首个差异文档与字段。
+
+### Added
+
+- 收尾阶段 `finalize`：版本登记失败不再报成功，作业标失败并保留预备份，修好后可续作。
+- 崩溃接管：worker 启动时接管"running 但队列已无任务"的写作业，复用已持久化的预备份，
+  绝不重新备份覆盖；没有预备份的判失败。
+- 跨 kind 互斥：迁移 000069 把唯一索引从 `(scope,kind)` 改成 `scope`（sync/restore 共用）。
+- 按版本恢复 `POST /library/restores`（管理员 + 该快照恢复授权）与续作 `POST /library/jobs/{id}/resume`
+  （管理员 + 闸口 + 授权，幂等键按源作业固定）。
+- 首次发布放行：生产可读且为空视为合法空基线；预备份不再作为启动前门禁（它由作业自身生成）。
+
+### Verified（本机候选环境，真实 PG/ES/MinIO，全部经 HTTP API）
+
+- 环境：候选 API :18089 + 候选 worker + 独立治理库 `kg_candidate_1028165117` +
+  真实 WeKnora 表结构的生产库 `kg_prod_1028165117` + 独立 ES 索引 + 独立 MinIO 桶。
+- 首次发布：生产为空 → 预备份 → 四类写入 → 校验 → 登记 `production_effective` 成功；
+  真实落库 4 文档 / 38 分块 / 21 资源 / 31 Wiki 页 / 122 索引文档。
+- 同请求重复：返回同一作业且 `reused=true`，不重复写生产。
+- 未授权快照：503 并写明"没有与目标快照匹配的有效发布授权"，不建作业。
+- 更新版本：第二份快照同步成功（真实改标题）。
+- 失败自动恢复：索引内容不一致、资源绑定行数不一致、对象上传失败等场景均被拒绝并回滚，
+  回滚后生产回到上一版本（多次真实复现）。
+- 收尾失败不报成功：版本登记因 ID 超长失败时作业标 failed、生产写入保留并记录恢复责任。
+
 ## 开发快照（按版本恢复 + 作业续作）
 
 ### Added

cmd/api/main.go

@@ -196,16 +196,20 @@ func run(ctx context.Context, getenv func(string) string) error {
 		ProductionReleaseAuthorizerRef:      config.ProductionReleaseAuthorizerRef,
 		ProductionReleaseAuthorizationToken: releaseAuthorizationToken,
 		// 数据源发现（系统设置 → 数据源配置）：用与 worker 同一份 dws 状态做只读列举。
-		DingtalkSources:            dingtalkSourceLister{},
-		DingtalkDocuments:          dingtalkSourceLister{},
-		BackupSchedule:             postgresstore.NewStore(pool),
-		VersionHistory:             versionHistoryStore,
-		VersionHistoryRepo:         versionHistoryRepo,
-		LibraryArchiveRoot:         config.LibraryArchiveRoot,
-		LibraryAgentSocket:         config.LibraryAgentSocket,
-		LibraryCapacityBudgetBytes: config.LibraryCapacityBudgetBytes,
-		LibraryRetentionVersions:   config.LibraryRetentionVersions,
-		LibraryRetentionDays:       config.LibraryRetentionDays,
+		DingtalkSources:              dingtalkSourceLister{},
+		DingtalkDocuments:            dingtalkSourceLister{},
+		BackupSchedule:               postgresstore.NewStore(pool),
+		VersionHistory:               versionHistoryStore,
+		VersionHistoryRepo:           versionHistoryRepo,
+		LibraryArchiveRoot:           config.LibraryArchiveRoot,
+		LibraryAgentSocket:           config.LibraryAgentSocket,
+		LibraryCapacityBudgetBytes:   config.LibraryCapacityBudgetBytes,
+		LibraryProductionDatabaseURL: config.LibraryProductionDatabaseURL,
+		LibraryProductionInternalKB:  config.LibraryProductionInternalKB,
+		LibraryProductionExternalKB:  config.LibraryProductionExternalKB,
+		LibraryProductionTenantID:    config.LibraryProductionTenantID,
+		LibraryRetentionVersions:     config.LibraryRetentionVersions,
+		LibraryRetentionDays:         config.LibraryRetentionDays,
 		BackupHost: backuphost.New(backuphost.Config{
 			SSHConfig: config.BackupHostSSHConfig,
 			Target:    config.BackupHostTarget,

cmd/library-e2e-seed/main.go

@@ -0,0 +1,192 @@
+// library-e2e-seed 从真实归档派生"第二版"快照，供候选验收跑新增/更新/撤销。
+// 不模拟：所有部件都经 Archive.Put 真实重算哈希并 Seal，产物是合法归档。
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"log"
+	"time"
+
+	"knowledge-governance/internal/librarysnapshot"
+)
+
+const maxBlob = 512 << 20
+
+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")
+	flag.Parse()
+	if *root == "" || *base == "" {
+		log.Fatal("root and base are required")
+	}
+	archive, err := librarysnapshot.Open(*root, maxBlob)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer archive.Close()
+	manifest, err := archive.Load(*base)
+	if err != nil {
+		log.Fatal(err)
+	}
+	dropped := ""
+	parts := make([]librarysnapshot.Part, 0, len(manifest.Parts))
+	for _, part := range manifest.Parts {
+		reader, err := archive.OpenPart(part)
+		if err != nil {
+			log.Fatal(err)
+		}
+		data, err := io.ReadAll(reader)
+		reader.Close()
+		if err != nil {
+			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)
+			}
+		}
+		sealed, err := archive.Put(bytes.NewReader(data))
+		if err != nil {
+			log.Fatal(err)
+		}
+		sealed.Kind, sealed.Name, sealed.Count = part.Kind, part.Name, part.Count
+		if part.Kind == "database" || part.Kind == "index" {
+			count, countErr := rowCount(data)
+			if countErr != nil {
+				log.Fatal(countErr)
+			}
+			sealed.Count = count
+		}
+		parts = append(parts, sealed)
+	}
+	manifest.Parts = parts
+	manifest.CreatedAt = time.Now().UTC()
+	id, err := archive.Seal(manifest)
+	if err != nil {
+		log.Fatal(err)
+	}
+	fmt.Printf(`{"snapshot_id":%q,"mode":%q,"base":%q,"dropped":%q}`, id, *mode, *base, dropped)
+}
+
+// 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
+	}
+	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
+		}
+	}
+	target := string(rows[index]["id"])
+	var targetID string
+	if err := json.Unmarshal(rows[index]["id"], &targetID); err != nil {
+		return nil, "", err
+	}
+	if mode == "revoke" {
+		rows = append(rows[:index], rows[index+1:]...)
+		return mustJSON(rows), targetID, nil
+	}
+	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
+}
+
+// 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
+	}
+	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
+			}
+		}
+		if owner == knowledgeID {
+			continue
+		}
+		kept = append(kept, item)
+	}
+	return mustJSON(kept), nil
+}
+
+// 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
+	}
+	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 {
+			continue
+		}
+		kept = append(kept, item)
+	}
+	return mustJSON(kept), nil
+}
+
+func rowCount(data []byte) (int64, error) {
+	var rows []json.RawMessage
+	if err := json.Unmarshal(data, &rows); err != nil {
+		return 0, err
+	}
+	return int64(len(rows)), nil
+}
+
+func mustJSON(value any) []byte {
+	data, err := json.Marshal(value)
+	if err != nil {
+		panic(err)
+	}
+	return data
+}

cmd/worker/library_handlers.go

@@ -35,6 +35,14 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a
 	_ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobRunning, "prebackup", "", "", "", "", now)
 	recorder := libraryStageRecorder{store: store, jobID: payload.LibraryJobID}
 
+	// 崩溃接管：作业里已经持久化过预备份快照，就直接复用它继续，
+	// 绝不重新备份——重新备份会把"半写状态"当成新基线，覆盖真正可回滚的那份。
+	previous, err := store.LibraryJob(ctx, payload.LibraryJobID)
+	resumed := err == nil && previous.SnapshotID != ""
+	if resumed {
+		recorder.Stage(ctx, "prebackup", "skipped", "接管失败前已创建的预备份 "+previous.SnapshotID)
+	}
+
 	gateReasons := make([]string, 0, 3)
 	if !config.ProductionReleaseRequestsEnabled {
 		gateReasons = append(gateReasons, "生产写入闸口关闭")
@@ -58,34 +66,72 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a
 	if err != nil || !found {
 		return failLibraryJob(ctx, store, recorder, payload.LibraryJobID, "没有与目标快照匹配的有效发布授权", "prebackup")
 	}
-	target, backups, wireErr := buildLibraryProductionStack(ctx, config, payload.Scope)
+	// 上一次生产生效版本登记的实际索引映射：这是本生产索引的基线。
+	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 raw, ok := item.Evidence["index_mapping_sha256"].(string); ok {
+				knownMapping = raw
+			}
+			break
+		}
+	}
+	target, backups, wireErr := buildLibraryProductionStack(ctx, config, payload.Scope, payload.TargetSnapshotID, knownMapping)
 	if wireErr != nil {
 		return failLibraryJob(ctx, store, recorder, payload.LibraryJobID, wireErr.Error(), "prebackup")
 	}
 	defer target.Close()
+	var backupSource libraryrelease.BackupStore = backups
+	if resumed {
+		backupSource = fixedBackupStore{ref: libraryrelease.BackupRef{ID: "backup-" + previous.SnapshotID,
+			SnapshotID: previous.SnapshotID}}
+	}
 
 	request := libraryrelease.Request{Target: payload.Scope, SnapshotID: payload.TargetSnapshotID,
 		ParamsDigest: digest, Now: now,
 		Grant: libraryrelease.Grant{Target: payload.Scope, SnapshotID: grant.SnapshotID,
 			ParamsDigest: grant.ParamsDigest, ExpiresAt: grant.ExpiresAt, AuthorizedBy: grant.AuthorizedBy}}
-	outcome := libraryrelease.Run(ctx, request, target, backups, recorder)
+	outcome := libraryrelease.Run(ctx, request, target, backupSource, recorder)
 	finishedAt := time.Now().UTC()
 	switch outcome.State {
 	case "succeeded":
-		// 成功后登记"生产生效版本"：这是生产当前的权威依据，只有真的验证通过才写。
+		mappingDigest := ""
+		if digest, mapErr := target.Search().MappingSHA256(ctx); mapErr == nil {
+			mappingDigest = digest
+		}
+		// 收尾阶段：先登记权威版本，登记成功才允许把作业标成功。
+		// 登记失败必须让作业失败（可续作），否则"版本记录"和"作业状态"会分叉。
+		recorder.Stage(ctx, "finalize", "running", "登记生产生效版本")
 		versionErr := store.RecordLibraryVersion(ctx, governance.LibraryVersion{
-			ID: "libver-prod-" + payload.Scope + "-" + outcome.Backup.SnapshotID, Scope: payload.Scope,
-			Kind: governance.LibraryVersionProductionEffective, SnapshotID: payload.TargetSnapshotID,
+			ID:    governance.LibraryVersionID(governance.LibraryVersionProductionEffective, payload.Scope, payload.TargetSnapshotID),
+			Scope: payload.Scope,
+			Kind:  governance.LibraryVersionProductionEffective, SnapshotID: payload.TargetSnapshotID,
 			SourceJobID: payload.LibraryJobID, CreatedBy: grant.AuthorizedBy,
 			Evidence: map[string]any{"backup_snapshot": outcome.Backup.SnapshotID,
-				"backup_bytes": outcome.Backup.Bytes, "action": string(kind)},
+				"backup_bytes": outcome.Backup.Bytes, "action": string(kind),
+				"index_mapping_sha256": mappingDigest},
 			CreatedAt: finishedAt,
 		})
 		if versionErr != nil {
-			log.Printf("library %s recorded but version registration failed: %v", kind, versionErr)
+			message := "生产写入已完成，但生效版本登记失败：" + versionErr.Error()
+			recorder.Stage(ctx, "finalize", "failed", message)
+			// 作业不能报成功：预备份仍在，修好数据库后可用 resume 续作收尾。
+			if advanceErr := store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobFailed, "finalize",
+				message, "预备份 "+outcome.Backup.SnapshotID+" 保留，数据库恢复后可用续作重试登记",
+				outcome.Backup.SnapshotID, payload.TargetSnapshotID, finishedAt); advanceErr != nil {
+				log.Printf("library %s finalize failure could not be recorded: %v", kind, advanceErr)
+			}
+			return errors.New(message)
+		}
+		recorder.Stage(ctx, "finalize", "succeeded", "已登记生效版本")
+		if advanceErr := store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobSucceeded, "finalize", "", "",
+			outcome.Backup.SnapshotID, payload.TargetSnapshotID, finishedAt); advanceErr != nil {
+			// 状态落不下去时不要把队列作业报成功：让上层可见并等待接管。
+			return fmt.Errorf("作业状态写入失败: %w", advanceErr)
 		}
-		_ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobSucceeded, "verify", "", "",
-			outcome.Backup.SnapshotID, payload.TargetSnapshotID, finishedAt)
 		log.Printf("library %s succeeded job=%s scope=%s snapshot=%s backup=%s", kind,
 			payload.LibraryJobID, payload.Scope, payload.TargetSnapshotID, outcome.Backup.SnapshotID)
 		return nil
@@ -104,6 +150,69 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a
 	}
 }
 
+// resumeInterruptedLibraryJobs 在 worker 启动时接管中断的写作业：
+// 已经持久化预备份的重新排队（同一个业务作业 id，阶段与备份都接着用），
+// 连预备份都没完成的才判失败——那种情况没有可回滚的基线，不能盲写。
+func resumeInterruptedLibraryJobs(ctx context.Context, store *postgresstore.Store, config appconfig.Config) (int, int, error) {
+	interrupted, err := store.InterruptedLibraryJobs(ctx)
+	if err != nil {
+		return 0, 0, err
+	}
+	resumed, failed := 0, 0
+	for _, job := range interrupted {
+		now := time.Now().UTC()
+		if job.SnapshotID == "" {
+			message := "worker 中断时还没有完成预备份，不能盲写目标"
+			_ = store.AdvanceLibraryJob(ctx, job.ID, governance.LibraryJobFailed, "prebackup", message,
+				"未对生产执行写入", "", "", now)
+			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)
+			continue
+		}
+		queueType := librarySyncJobType
+		if job.Kind == governance.LibraryJobRestore {
+			queueType = libraryRestoreJobType
+		}
+		payload, _ := json.Marshal(map[string]string{"library_job_id": job.ID, "scope": job.Scope,
+			"target_snapshot_id": job.TargetSnapshotID})
+		queueJob, err := store.Enqueue(ctx, jobs.EnqueueRequest{
+			ID: governance.QueueJobIDFor(job.ID) + "-r" + now.Format("050405"), Type: queueType, Payload: payload,
+			AvailableAt: now, MaxAttempts: 1,
+			IdempotencyKey: "library.resume.queue:" + job.ID + ":" + now.Format("150405"),
+		}, now)
+		if err != nil || queueJob == nil {
+			log.Printf("library job %s resume enqueue failed: %v", job.ID, err)
+			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
+		}
+		resumed++
+		log.Printf("library job %s resumed after worker restart (backup %s)", job.ID, job.SnapshotID)
+	}
+	return resumed, failed, nil
+}
+
+// fixedBackupStore 在崩溃接管时返回已经存在的预备份，不再新建。
+type fixedBackupStore struct{ ref libraryrelease.BackupRef }
+
+func (s fixedBackupStore) Prepare(context.Context, string, libraryrelease.CurrentState) (libraryrelease.BackupRef, error) {
+	return s.ref, nil
+}
+
+// BackupPrepared 让 recorder 兼作 BackupObserver，把预备份快照立刻写进作业。
+func (r libraryStageRecorder) BackupPrepared(ctx context.Context, backup libraryrelease.BackupRef) {
+	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)
+	}
+}
+
 func failLibraryJob(ctx context.Context, store *postgresstore.Store, recorder libraryStageRecorder, jobID, message, stage string) error {
 	recorder.Stage(ctx, stage, "failed", message)
 	_ = store.AdvanceLibraryJob(ctx, jobID, governance.LibraryJobFailed, stage, message,

cmd/worker/library_wiring.go

@@ -14,7 +14,7 @@ import (
 // buildLibraryProductionStack 用部署注入的配置装配生产整库目标与独立预备份存储。
 // 任何一项缺失都返回明确原因，调用方据此把作业失败并把缺项写进阶段，
 // 不用"未配置"的占位对象假装能跑。
-func buildLibraryProductionStack(ctx context.Context, config appconfig.Config, scope string) (
+func buildLibraryProductionStack(ctx context.Context, config appconfig.Config, scope, snapshotID, knownMapping string) (
 	*v071.ProductionLibraryTarget, *v071.ProductionBackupStore, error) {
 	missing := make([]string, 0, 4)
 	if config.LibraryProductionDatabaseURL == "" {
@@ -42,7 +42,7 @@ func buildLibraryProductionStack(ctx context.Context, config appconfig.Config, s
 	}
 	secure := strings.HasPrefix(config.LibraryProductionS3Endpoint, "https://")
 	objects, err := v071.NewS3ObjectStore(v071.S3ObjectStoreConfig{
-		Provider: "s3", Endpoint: config.LibraryProductionS3Endpoint,
+		Provider: config.LibraryProductionS3Provider, Endpoint: config.LibraryProductionS3Endpoint,
 		Region: config.LibraryProductionS3Region, Bucket: config.LibraryProductionS3Bucket,
 		AccessKeyID: config.LibraryProductionS3AccessKey, SecretAccessKey: config.LibraryProductionS3SecretKey,
 		Secure: secure, ForcePathStyle: true, MaxObjectBytes: 256 * 1024 * 1024,
@@ -60,21 +60,52 @@ func buildLibraryProductionStack(ctx context.Context, config appconfig.Config, s
 	if err != nil {
 		return nil, nil, err
 	}
-	// 契约以生产**当前**索引映射为准：这样预备份的兼容性与恢复目标是同一份事实。
-	mapping, err := target.Search().MappingSHA256(ctx)
+	// 契约以"要同步的这份快照"为准：生产必须向它看齐，而不是反过来。
+	// 生产为空时索引可能根本不存在，这时不能拿生产映射当契约（会永远启动不了）。
+	archive, err := librarysnapshot.Open(config.LibraryArchiveRoot, 256*1024*1024)
 	if err != nil {
 		target.Close()
-		return nil, nil, fmt.Errorf("读取生产索引映射失败: %w", err)
+		return nil, nil, fmt.Errorf("归档不可用，无法读取快照契约: %w", err)
 	}
-	contract := librarysnapshot.Compatibility{
-		WeKnoraVersion: "v0.7.1", SchemaVersion: 74, RetrieverEngine: "elasticsearch_v8",
-		EmbeddingModel: config.LibraryEmbeddingModel, EmbeddingDimension: config.LibraryEmbeddingDimension,
-		IndexMappingSHA256: mapping,
+	manifest, err := archive.Load(snapshotID)
+	archive.Close()
+	if err != nil {
+		target.Close()
+		return nil, nil, fmt.Errorf("读取目标快照失败: %w", err)
+	}
+	contract := manifest.Compatibility
+	if contract.EmbeddingDimension == 0 {
+		contract.EmbeddingModel = config.LibraryEmbeddingModel
+		contract.EmbeddingDimension = config.LibraryEmbeddingDimension
+	}
+	// 生产索引存在时必须与快照契约一致；不存在或为空基线时跳过比对
+	// （空索引的映射没有可比性，非空之后仍然严格比对）。
+	if mapping, mapErr := target.Search().MappingSHA256(ctx); mapErr == nil {
+		// 映射基线：优先用"上一次生产生效版本登记的实际映射"，
+		// 它才是这份生产索引的既有事实；快照契约只管"来自哪份数据"。
+		baseline := knownMapping
+		if baseline == "" {
+			baseline = contract.IndexMappingSHA256
+		}
+		if baseline != "" && mapping != baseline {
+			current, currentErr := target.Current(ctx)
+			emptyBaseline := currentErr == nil && current.Counts["index"] == 0 && current.Counts["chunks"] == 0
+			if !emptyBaseline {
+				target.Close()
+				return nil, nil, errors.New("生产索引映射与上次生效版本不一致（索引被改动过），先对齐索引映射")
+			}
+		}
+	}
+	// 预备份描述的是"生产现在的索引"，不是"快照的来源索引"：
+	// 生产映射基线优先用上次生效版本登记的实际值。
+	backupContract := contract
+	if knownMapping != "" {
+		backupContract.IndexMappingSHA256 = knownMapping
 	}
 	backups, err := v071.NewProductionBackupStore(v071.ProductionBackupConfig{
 		Pool: target.Pool(), Search: target.Search(), Objects: target.Objects(),
 		ArchiveRoot: config.LibraryArchiveRoot, Scope: scope, Identity: target.Identity(),
-		Compatibility: contract, MaxDocuments: 200000, MaxObjectByte: 256 * 1024 * 1024,
+		Compatibility: backupContract, MaxDocuments: 200000, MaxObjectByte: 256 * 1024 * 1024,
 	})
 	if err != nil {
 		target.Close()

cmd/worker/main.go

@@ -73,8 +73,13 @@ func run(ctx context.Context, getenv func(string) string) error {
 		return err
 	}
 	store := postgresstore.NewStore(pool)
-	// 启动先收拾孤儿业务作业：它们会永久占住 (scope,kind) 唯一索引，
-	// 不清掉的话页面会一直显示"在飞"，同一目标再也发不出新作业。
+	// 启动先接管被中断的写作业（复用已持久化的预备份），再把剩下的孤儿判失败：
+	// 顺序反了会把可接管的作业先标失败，丢掉已完成的备份。
+	if resumed, failed, resumeErr := resumeInterruptedLibraryJobs(context.Background(), store, config); resumeErr != nil {
+		log.Printf("library job resume failed: %v", resumeErr)
+	} else if resumed > 0 || failed > 0 {
+		log.Printf("library job resume: %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 {
@@ -91,6 +96,15 @@ func run(ctx context.Context, getenv func(string) string) error {
 	if mirrorErr != nil {
 		log.Print("QA local mirror configuration invalid; handler was not registered")
 	}
+	// 生产同步/恢复的执行器不依赖主机侧封存 agent：没有 agent socket 也要能跑。
+	// 这里在全部注册之后无条件覆盖一次，确保 library.sync/library.restore 一定在队列类型里。
+	handlers[librarySyncJobType] = func(ctx context.Context, job jobs.Job) error {
+		return runLibraryRelease(ctx, store, config, job, governance.LibraryJobSync)
+	}
+	handlers[libraryRestoreJobType] = func(ctx context.Context, job jobs.Job) error {
+		return runLibraryRelease(ctx, store, config, job, governance.LibraryJobRestore)
+	}
+
 	worker, err := jobs.NewWorker(jobs.WorkerConfig{
 		Queue:         store,
 		WorkerID:      config.WorkerID,
@@ -192,8 +206,9 @@ func workerHandlers(store *postgresstore.Store, config appconfig.Config) (map[st
 			}
 			// 封存产物成版本记录：只登记"测试封存"，绝不写成生产生效版本。
 			versionErr := store.RecordLibraryVersion(ctx, governance.LibraryVersion{
-				ID: "libver-" + payload.Scope + "-" + result.SnapshotID, Scope: payload.Scope,
-				Kind: governance.LibraryVersionTestSealed, SnapshotID: result.SnapshotID, SourceJobID: libraryJobID,
+				ID:    governance.LibraryVersionID(governance.LibraryVersionTestSealed, payload.Scope, result.SnapshotID),
+				Scope: payload.Scope,
+				Kind:  governance.LibraryVersionTestSealed, SnapshotID: result.SnapshotID, SourceJobID: libraryJobID,
 				Evidence: map[string]any{"capture_job": payload.CaptureJob, "replay": result.Replay,
 					"requested_by": payload.RequestedBy, "agent_result": json.RawMessage(result.Payload)},
 				CreatedBy: payload.RequestedBy, CreatedAt: finishedAt,

internal/appconfig/config.go

@@ -79,6 +79,9 @@ type Config struct {
 	LibraryProductionTenantID    uint64
 	LibraryProductionInternalKB  string
 	LibraryProductionExternalKB  string
+	// LibraryProductionS3Provider 必须与生产库 resources.provider 的取值一致：
+	// 对象路径按 provider 路由，写错会在上传时被拒绝（而不是写错桶）。
+	LibraryProductionS3Provider  string
 	LibraryProductionS3Endpoint  string
 	LibraryProductionS3Region    string
 	LibraryProductionS3Bucket    string
@@ -173,6 +176,7 @@ func Load(getenv func(string) string) (Config, error) {
 		LibraryProductionTenantID:    uint64(tenantID),
 		LibraryProductionInternalKB:  strings.TrimSpace(getenv("LIBRARY_PRODUCTION_KB_INTERNAL")),
 		LibraryProductionExternalKB:  strings.TrimSpace(getenv("LIBRARY_PRODUCTION_KB_EXTERNAL")),
+		LibraryProductionS3Provider:  valueOrDefault(strings.TrimSpace(getenv("LIBRARY_PRODUCTION_S3_PROVIDER")), "s3"),
 		LibraryProductionS3Endpoint:  strings.TrimSpace(getenv("LIBRARY_PRODUCTION_S3_ENDPOINT")),
 		LibraryProductionS3Region:    valueOrDefault(strings.TrimSpace(getenv("LIBRARY_PRODUCTION_S3_REGION")), "us-east-1"),
 		LibraryProductionS3Bucket:    strings.TrimSpace(getenv("LIBRARY_PRODUCTION_S3_BUCKET")),

internal/governance/library_flow.go

@@ -97,6 +97,21 @@ type LibraryVersion struct {
 
 // LibraryRequestDigest 绑定一次请求的确切含义：同键不同体必须被拒绝，而不是
 // 复用旧作业。只覆盖会影响执行结果的字段；actor 不参与摘要。
+// LibraryVersionID 生成版本记录 ID：library_versions.id 是 varchar(64)，
+// 用完整快照 ID 拼会超长（真实踩过），所以用稳定短摘要——同样的输入永远得到同一个 ID，
+// 收尾重试因此是幂等的。
+func LibraryVersionID(kind LibraryVersionKind, scope, snapshotID string) string {
+	sum := sha256.Sum256([]byte(string(kind) + "|" + scope + "|" + snapshotID))
+	return "libver-" + string(kind)[:0] + hex.EncodeToString(sum[:])[:16]
+}
+
+// QueueJobIDFor 生成队列作业 ID：jobs.id 是 varchar(36)，
+// 直接用业务作业 ID 加后缀会超长（真实踩过），所以用固定长度摘要。
+func QueueJobIDFor(libraryJobID string) string {
+	sum := sha256.Sum256([]byte(libraryJobID))
+	return "q" + hex.EncodeToString(sum[:])[:32]
+}
+
 func LibraryRequestDigest(scope string, kind LibraryJobKind, payload map[string]any) (string, error) {
 	canonical := map[string]any{"scope": scope, "kind": string(kind), "payload": payload}
 	raw, err := json.Marshal(canonical)
@@ -110,7 +125,8 @@ func LibraryRequestDigest(scope string, kind LibraryJobKind, payload map[string]
 // SyncStages 是生产同步作业的固定阶段顺序：先备份并核验恢复条件，再写生产，
 // 最后验证；失败时走恢复而不是把作业标成功。
 func SyncStages() []string {
-	return []string{"prebackup", "database", "resources", "index", "wiki", "verify", "rollback"}
+	// finalize 是收尾：登记生产生效版本。登记失败不得报成功，所以它是独立阶段。
+	return []string{"prebackup", "database", "resources", "index", "wiki", "verify", "finalize", "rollback"}
 }
 
 // SealStages 是封存作业的阶段：导出是原子操作，没有分项进度。

internal/httpapi/library_flow.go

@@ -5,10 +5,13 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"log"
 	"net/http"
 	"strings"
 	"time"
 
+	"github.com/jackc/pgx/v5/pgxpool"
+
 	"knowledge-governance/internal/governance"
 	"knowledge-governance/internal/jobs"
 	"knowledge-governance/internal/librarydiff"
@@ -42,6 +45,35 @@ type libraryKnownView struct {
 	State   string              `json:"state"` // known | unknown
 	Version *libraryVersionView `json:"version,omitempty"`
 	Reason  string              `json:"reason,omitempty"`
+	// EmptyBaseline 表示"没有生效版本记录，但生产目标可读且确实是空的"：
+	// 这是首次发布的合法起点，和"生产非空却没有权威版本"必须区分。
+	EmptyBaseline bool `json:"empty_baseline,omitempty"`
+}
+
+// productionScopeEmpty 只读探测生产目标：真正连库数一遍，不用缓存或推断。
+func productionScopeEmpty(ctx context.Context, config Config, scope string) (bool, error) {
+	if strings.TrimSpace(config.LibraryProductionDatabaseURL) == "" {
+		return false, errors.New("生产整库目标未配置")
+	}
+	kb := config.LibraryProductionInternalKB
+	if scope == "external" {
+		kb = config.LibraryProductionExternalKB
+	}
+	if kb == "" {
+		return false, errors.New("生产知识库未配置")
+	}
+	pool, err := pgxpool.New(ctx, config.LibraryProductionDatabaseURL)
+	if err != nil {
+		return false, err
+	}
+	defer pool.Close()
+	var count int64
+	err = pool.QueryRow(ctx, `SELECT count(*) FROM knowledges WHERE tenant_id=$1 AND knowledge_base_id=$2 AND deleted_at IS NULL`,
+		config.LibraryProductionTenantID, kb).Scan(&count)
+	if err != nil {
+		return false, err
+	}
+	return count == 0, nil
 }
 
 type libraryDiffView struct {
@@ -302,6 +334,13 @@ func libraryVersionViewOf(item governance.LibraryVersion) libraryVersionView {
 
 // librarySyncGate 只依据真实存在的证据判断能不能写生产：闸口、生产端点、
 // 生产权威版本、目标快照的预备份。任何一项缺失都给出具体原因，不默认放行。
+func errString(err error) string {
+	if err == nil {
+		return ""
+	}
+	return err.Error()
+}
+
 func librarySyncGate(config Config, production libraryKnownView, prebackup libraryKnownView, target string) libraryGateView {
 	return librarySyncGateWithGrant(config, production, prebackup, target, false)
 }
@@ -319,17 +358,23 @@ func librarySyncGateWithGrant(config Config, production libraryKnownView, prebac
 	if !config.ProductionBackupRestoreVerified {
 		gate.Reasons = append(gate.Reasons, "没有已核验的生产恢复演练证据（PRODUCTION_BACKUP_RESTORE_VERIFIED=false）")
 	}
-	if strings.TrimSpace(config.ProductionWeKnoraBaseURL) == "" {
-		gate.Reasons = append(gate.Reasons, "生产端点未配置，无法核验生产当前版本")
+	// 生产端点以"整库目标是否配置"为准：整库同步直接写 PG/ES/对象，
+	// 不依赖 WeKnora 的 HTTP 端，所以不能拿它是否配置当门禁。
+	if strings.TrimSpace(config.LibraryProductionDatabaseURL) == "" {
+		gate.Reasons = append(gate.Reasons, "生产整库目标未配置（LIBRARY_PRODUCTION_DATABASE_URL），无法核验生产当前状态")
 	}
-	if production.State != "known" {
+	// 首次发布：生产还没有"生效版本"记录，但只要生产目标可读且是空的，
+	// 这就是一个合法的"空基线"，不应该被永久挡住。非空但无权威版本仍然阻断。
+	if production.State != "known" && !production.EmptyBaseline {
 		gate.Reasons = append(gate.Reasons, "生产当前权威版本未知，不能启动写入")
 	}
 	if target == "" {
 		gate.Reasons = append(gate.Reasons, "还没有可同步的测试封存版本")
 	}
-	if prebackup.State != "known" {
-		gate.Reasons = append(gate.Reasons, "本次目标快照还没有生产预备份")
+	// 预备份是作业内第一步生成的：启动前要求它已存在会造成死循环。
+	// 这里只在作业已经被判定失败或回滚失败时提示，不作为启动前的硬门禁。
+	if prebackup.State == "failed" {
+		gate.Reasons = append(gate.Reasons, "上一次预备份失败，先处理后才能重试")
 	}
 	if config.LibraryCapacityBudgetBytes <= 0 {
 		gate.Reasons = append(gate.Reasons, "容量预算未设置（LIBRARY_CAPACITY_BUDGET_BYTES），无法证明装得下")
@@ -507,6 +552,10 @@ func librarySyncEnqueue(w http.ResponseWriter, r *http.Request, config Config) {
 	}
 	overview.Production = libraryKnownView{State: "unknown"}
 	overview.Prebackup = libraryKnownView{State: "unknown"}
+	if empty, probeErr := productionScopeEmpty(r.Context(), config, request.Scope); probeErr == nil && empty {
+		overview.Production = libraryKnownView{State: "unknown", EmptyBaseline: true,
+			Reason: "生产目标可读且为空：本次是首次发布，写入前仍会生成完整预备份"}
+	}
 	for index := range versions {
 		item := versions[index]
 		if item.Kind == governance.LibraryVersionProductionEffective && item.SupersededAt == nil {
@@ -559,6 +608,21 @@ func librarySyncEnqueue(w http.ResponseWriter, r *http.Request, config Config) {
 		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot create library sync job"})
 		return
 	}
+	// 上一次同参数请求失败/被取消后，幂等键不能再把新请求塞回那个死作业里：
+	// 换一个"重试代次"键重新建作业，旧作业保留为审计记录。
+	if reused && (job.State == governance.LibraryJobFailed || job.State == governance.LibraryJobCancelled) {
+		job, reused, err = store.CreateLibraryJob(r.Context(), governance.LibraryJob{
+			ID: fmt.Sprintf("libsync-%s-%d", request.Scope, now.UnixNano()), Scope: request.Scope,
+			Kind:           governance.LibraryJobSync,
+			IdempotencyKey: "library.sync:" + request.Scope + ":" + request.Target + ":retry:" + now.Format("20060102T150405.000000000"),
+			RequestDigest:  digest, TargetSnapshotID: request.Target, Actor: actor,
+		}, governance.SyncStages(), now)
+		if err != nil || reused {
+			writeJSON(w, http.StatusConflict, map[string]any{"code": "library_sync_retry_conflict",
+				"error": "无法为重试创建新作业", "detail": errString(err)})
+			return
+		}
+	}
 	if reused {
 		writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "reused": true})
 		return
@@ -572,10 +636,11 @@ func librarySyncEnqueue(w http.ResponseWriter, r *http.Request, config Config) {
 		return
 	}
 	queueJob, err := config.Store.Enqueue(r.Context(), jobs.EnqueueRequest{
-		ID: job.ID + "-queue", Type: librarySyncJobType, Payload: payload, AvailableAt: now,
+		ID: governance.QueueJobIDFor(job.ID), Type: librarySyncJobType, Payload: payload, AvailableAt: now,
 		MaxAttempts: 1, IdempotencyKey: "library.sync.queue:" + job.ID,
 	}, now)
 	if err != nil || queueJob == nil {
+		log.Printf("library sync enqueue failed job=%s: %v", job.ID, err)
 		_ = store.CancelLibraryJob(r.Context(), job.ID, "队列投递失败，未开始同步", now)
 		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot enqueue sync job"})
 		return

internal/httpapi/library_flow_test.go

@@ -128,7 +128,7 @@ func TestLibrarySyncGateRequiresEveryPieceOfEvidence(t *testing.T) {
 		{
 			name: "flags on but restore evidence missing",
 			config: Config{ProductionReleaseRequestsEnabled: true,
-				ProductionWeKnoraBaseURL: "http://prod.invalid"},
+				LibraryProductionDatabaseURL: "postgres://prod.invalid/db", LibraryProductionInternalKB: "kb-int"},
 			production: known,
 			prebackup:  known,
 			target:     "snap-1",
@@ -138,7 +138,7 @@ func TestLibrarySyncGateRequiresEveryPieceOfEvidence(t *testing.T) {
 		{
 			name: "capacity budget missing blocks even when everything else is present",
 			config: Config{ProductionReleaseRequestsEnabled: true, ProductionBackupRestoreVerified: true,
-				ProductionWeKnoraBaseURL: "http://prod.invalid"},
+				LibraryProductionDatabaseURL: "postgres://prod.invalid/db", LibraryProductionInternalKB: "kb-int"},
 			production: known,
 			prebackup:  known,
 			target:     "snap-1",
@@ -148,7 +148,7 @@ func TestLibrarySyncGateRequiresEveryPieceOfEvidence(t *testing.T) {
 		{
 			name: "everything present",
 			config: Config{ProductionReleaseRequestsEnabled: true, ProductionBackupRestoreVerified: true,
-				ProductionWeKnoraBaseURL: "http://prod.invalid", LibraryCapacityBudgetBytes: 1 << 40},
+				LibraryProductionDatabaseURL: "postgres://prod.invalid/db", LibraryProductionInternalKB: "kb-int", LibraryCapacityBudgetBytes: 1 << 40},
 			production: known,
 			prebackup:  known,
 			target:     "snap-1",
@@ -198,26 +198,62 @@ func TestLibrarySyncEnqueueRefusesWithoutCreatingJob(t *testing.T) {
 	}
 }
 
-// 即使闸口打开，缺少与目标快照绑定的预备份也必须拒绝。
+// 预备份是作业内第一步生成的，不能拿"启动前还没有预备份"当门禁：
+// 否则预备份只能由作业生成、作业又要求预备份，永远启动不了。
+// 但"上一次预备份失败"必须阻断重试。
+func TestLibrarySyncEnqueueDoesNotRequirePrebackupBeforeStart(t *testing.T) {
+	store := &fakeLibraryStore{}
+	config := Config{Store: store, ProductionReleaseRequestsEnabled: true, ProductionBackupRestoreVerified: true,
+		LibraryProductionDatabaseURL: "postgres://prod.invalid/db", LibraryProductionInternalKB: "kb-int",
+		LibraryCapacityBudgetBytes: 1 << 40}
+	gate := librarySyncGateWithGrant(config, libraryKnownView{State: "known"}, libraryKnownView{State: "unknown"}, "snap-1", true)
+	if !gate.SyncAllowed {
+		t.Fatalf("missing prebackup must not block the start: %v", gate.Reasons)
+	}
+	failed := librarySyncGateWithGrant(config, libraryKnownView{State: "known"},
+		libraryKnownView{State: "failed"}, "snap-1", true)
+	if failed.SyncAllowed || !strings.Contains(strings.Join(failed.Reasons, "|"), "预备份失败") {
+		t.Fatalf("failed prebackup must block: %v", failed.Reasons)
+	}
+}
+
+// 首次发布：生产没有生效版本，但目标为空 = 合法空基线。
+func TestLibrarySyncGateAllowsFirstPublishOnEmptyBaseline(t *testing.T) {
+	config := Config{ProductionReleaseRequestsEnabled: true, ProductionBackupRestoreVerified: true,
+		LibraryProductionDatabaseURL: "postgres://prod.invalid/db", LibraryProductionInternalKB: "kb-int",
+		LibraryCapacityBudgetBytes: 1 << 40}
+	empty := libraryKnownView{State: "unknown", EmptyBaseline: true}
+	if gate := librarySyncGateWithGrant(config, empty, libraryKnownView{State: "unknown"}, "snap-1", true); !gate.SyncAllowed {
+		t.Fatalf("empty production must be a legal first-publish baseline: %v", gate.Reasons)
+	}
+	// 非空但没有权威版本：仍然阻断。
+	if gate := librarySyncGateWithGrant(config, libraryKnownView{State: "unknown"}, libraryKnownView{State: "unknown"}, "snap-1", true); gate.SyncAllowed {
+		t.Fatal("unknown non-empty production must block")
+	}
+}
+
+// 上一次预备份失败时必须阻断重试（预备份本身是作业内生成的，不能当启动前门禁）。
 func TestLibrarySyncEnqueueRequiresPrebackupForTarget(t *testing.T) {
 	store := &fakeLibraryStore{versions: []governance.LibraryVersion{{
 		ID: "ver-prod-1", Scope: "internal", Kind: governance.LibraryVersionProductionEffective,
 		SnapshotID: "prod-snap", VersionLabel: "prod-2026.10", CreatedAt: time.Now().UTC(),
+	}, {
+		ID: "ver-backup-1", Scope: "internal", Kind: governance.LibraryVersionProductionPrebackup,
+		SnapshotID: "snap-1", CreatedAt: time.Now().UTC(),
 	}}}
 	config := Config{Store: store, ProductionReleaseRequestsEnabled: true, ProductionBackupRestoreVerified: true,
-		ProductionWeKnoraBaseURL: "http://prod.invalid"}
+		LibraryProductionDatabaseURL: "postgres://prod.invalid/db", LibraryProductionInternalKB: "kb-int",
+		LibraryCapacityBudgetBytes: 1 << 40}
 	request := httptest.NewRequest(http.MethodPost, "/api/v1/library/syncs",
 		strings.NewReader(`{"scope":"internal","target_snapshot_id":"snap-1"}`))
 	recorder := httptest.NewRecorder()
 	librarySyncEnqueue(recorder, request, config)
+	// 没有授权仍然 503，但原因里不能再出现"还没有生产预备份"这种死循环门禁。
 	if recorder.Code != http.StatusServiceUnavailable {
 		t.Fatalf("status = %d, want 503", recorder.Code)
 	}
-	if !strings.Contains(recorder.Body.String(), "预备份") {
-		t.Fatalf("refusal must name the missing prebackup: %s", recorder.Body.String())
-	}
-	if store.createN != 0 {
-		t.Fatalf("created %d jobs without a prebackup", store.createN)
+	if strings.Contains(recorder.Body.String(), "还没有生产预备份") {
+		t.Fatalf("start gate must not require a prebackup that only the job can create: %s", recorder.Body.String())
 	}
 }
 
@@ -236,7 +272,7 @@ func TestLibrarySyncEnqueueCreatesJobOnlyWithGrant(t *testing.T) {
 	}}
 	store.enqueueEchoRequestedID = true
 	config := Config{Store: store, ProductionReleaseRequestsEnabled: true, ProductionBackupRestoreVerified: true,
-		ProductionWeKnoraBaseURL: "http://prod.invalid", LibraryCapacityBudgetBytes: 1 << 40}
+		LibraryProductionDatabaseURL: "postgres://prod.invalid/db", LibraryProductionInternalKB: "kb-int", LibraryCapacityBudgetBytes: 1 << 40}
 	body := `{"scope":"internal","target_snapshot_id":"` + target + `"}`
 
 	// 先没有授权

internal/httpapi/library_restore.go

@@ -100,6 +100,17 @@ func libraryRestoreEnqueue(w http.ResponseWriter, r *http.Request, config Config
 		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot create restore job"})
 		return
 	}
+	if reused && (job.State == governance.LibraryJobFailed || job.State == governance.LibraryJobCancelled) {
+		job, reused, err = store.CreateLibraryJob(r.Context(), governance.LibraryJob{
+			ID: jobID + "-retry", Scope: request.Scope, Kind: governance.LibraryJobRestore,
+			IdempotencyKey: "library.restore:" + request.Scope + ":" + request.SnapshotID + ":retry:" + now.Format("20060102T150405.000000000"),
+			RequestDigest:  digest, TargetSnapshotID: request.SnapshotID, Actor: identity.Username,
+		}, governance.SyncStages(), now)
+		if err != nil || reused {
+			writeJSON(w, http.StatusConflict, map[string]any{"code": "library_restore_retry_conflict", "error": "无法为重试创建新作业"})
+			return
+		}
+	}
 	if reused {
 		writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "reused": true})
 		return
@@ -112,7 +123,7 @@ func libraryRestoreEnqueue(w http.ResponseWriter, r *http.Request, config Config
 		return
 	}
 	queueJob, err := config.Store.Enqueue(r.Context(), jobs.EnqueueRequest{
-		ID: job.ID + "-queue", Type: libraryRestoreJobType, Payload: payload, AvailableAt: now,
+		ID: governance.QueueJobIDFor(job.ID), Type: libraryRestoreJobType, Payload: payload, AvailableAt: now,
 		MaxAttempts: 1, IdempotencyKey: "library.restore.queue:" + job.ID,
 	}, now)
 	if err != nil || queueJob == nil {
@@ -130,8 +141,9 @@ func libraryRestoreEnqueue(w http.ResponseWriter, r *http.Request, config Config
 // 崩溃续作：失败作业可以按原参数重新排队，生成新作业（旧作业保留为审计记录）。
 func libraryJobResume(w http.ResponseWriter, r *http.Request, config Config) {
 	identity, ok := sessionFrom(r.Context())
-	if !ok || roleRank(identity.Role) < roleRank(roleEngineer) {
-		writeJSON(w, http.StatusForbidden, map[string]any{"error": "续作需要知识工程师及以上权限"})
+	// 续作会重新写生产，权限与恢复一致：只有管理员能触发。
+	if !ok || roleRank(identity.Role) < roleRank(roleAdmin) {
+		writeJSON(w, http.StatusForbidden, map[string]any{"error": "续作需要管理员权限"})
 		return
 	}
 	store, ok := libraryStore(config)
@@ -149,8 +161,41 @@ func libraryJobResume(w http.ResponseWriter, r *http.Request, config Config) {
 		writeJSON(w, http.StatusConflict, map[string]any{"error": "只有失败或已取消的作业可以续作"})
 		return
 	}
+	// 与同步/恢复一致：闸口、恢复演练证据与授权都要在 API 侧 fail-closed，
+	// 不能让前台接受一个 worker 必然拒绝的请求。
+	reasons := make([]string, 0, 3)
+	if !config.ProductionReleaseRequestsEnabled {
+		reasons = append(reasons, "生产写入闸口关闭")
+	}
+	if !config.ProductionBackupRestoreVerified {
+		reasons = append(reasons, "没有已核验的生产恢复演练证据")
+	}
+	var digest string
+	switch source.Kind {
+	case governance.LibraryJobSync:
+		digest, _ = librarySyncParamsDigest(source.Scope, source.TargetSnapshotID)
+	case governance.LibraryJobRestore:
+		digest, _ = libraryRestoreParamsDigest(source.Scope, source.TargetSnapshotID)
+	}
+	authorizationOK := false
+	if storeWithAuth, ok := config.Store.(interface {
+		ActiveLibraryAuthorization(ctx context.Context, scope, snapshotID, digest string, now time.Time) (governance.LibraryReleaseAuthorization, bool, error)
+	}); ok && digest != "" {
+		_, found, authErr := storeWithAuth.ActiveLibraryAuthorization(r.Context(), source.Scope,
+			source.TargetSnapshotID, digest, time.Now().UTC())
+		authorizationOK = authErr == nil && found
+	}
+	if !authorizationOK {
+		reasons = append(reasons, "没有针对该快照的有效授权")
+	}
+	if len(reasons) > 0 {
+		writeJSON(w, http.StatusServiceUnavailable, map[string]any{"code": "library_resume_blocked", "reasons": reasons})
+		return
+	}
+
 	now := time.Now().UTC()
-	jobID := sourceID + "-r" + now.Format("150405")
+	// 幂等键按源作业固定：重复点击返回同一个续作作业，而不是每次新建。
+	jobID := sourceID + "-resume"
 	var kind governance.LibraryJobKind
 	queueType := ""
 	switch source.Kind {
@@ -162,24 +207,31 @@ func libraryJobResume(w http.ResponseWriter, r *http.Request, config Config) {
 		writeJSON(w, http.StatusConflict, map[string]any{"error": "封存作业由封存入口重发，不走续作"})
 		return
 	}
-	digest := source.RequestDigest
-	job, _, err := store.CreateLibraryJob(r.Context(), governance.LibraryJob{
+	job, reused, err := store.CreateLibraryJob(r.Context(), governance.LibraryJob{
 		ID: jobID, Scope: source.Scope, Kind: kind,
-		IdempotencyKey: "library.resume:" + sourceID + ":" + now.Format("150405"),
+		IdempotencyKey: "library.resume:" + sourceID,
 		RequestDigest:  digest, TargetSnapshotID: source.TargetSnapshotID, Actor: identity.Username,
 	}, governance.SyncStages(), now)
 	if errors.Is(err, governance.ErrLibraryJobActive) {
 		writeJSON(w, http.StatusConflict, map[string]any{"error": "该目标已有在飞作业"})
 		return
 	}
+	if errors.Is(err, governance.ErrLibraryJobIdempotencyConflict) {
+		writeJSON(w, http.StatusConflict, map[string]any{"error": "同一续作请求参数不一致"})
+		return
+	}
 	if err != nil {
 		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot create resumed job"})
 		return
 	}
+	if reused {
+		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})
 	queueJob, err := config.Store.Enqueue(r.Context(), jobs.EnqueueRequest{
-		ID: job.ID + "-queue", Type: queueType, Payload: payload, AvailableAt: now,
+		ID: governance.QueueJobIDFor(job.ID), Type: queueType, Payload: payload, AvailableAt: now,
 		MaxAttempts: 1, IdempotencyKey: "library.resume.queue:" + job.ID,
 	}, now)
 	if err != nil || queueJob == nil {
@@ -187,6 +239,11 @@ func libraryJobResume(w http.ResponseWriter, r *http.Request, config Config) {
 		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot enqueue resumed job"})
 		return
 	}
-	_ = store.AttachLibraryQueueJob(r.Context(), job.ID, queueJob.ID, now)
-	writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "resumed_from": sourceID})
+	// 附不上队列号就等于作业没人跑：必须显式失败并取消，不能静默返回 202。
+	if attachErr := store.AttachLibraryQueueJob(r.Context(), job.ID, queueJob.ID, now); attachErr != nil {
+		_ = store.CancelLibraryJob(r.Context(), job.ID, "队列归属写入失败", now)
+		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot attach resumed job"})
+		return
+	}
+	writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "resumed_from": sourceID, "reused": false})
 }

internal/httpapi/library_restore_test.go

@@ -78,7 +78,7 @@ func TestLibraryJobResumeRejectsNonFailedJobs(t *testing.T) {
 	request := httptest.NewRequest(http.MethodPost, "/api/v1/library/jobs/job-1/resume", nil)
 	request.SetPathValue("id", "job-1")
 	request = request.WithContext(context.WithValue(request.Context(), sessionKey{},
-		sessionIdentity{Username: "eng", Role: roleEngineer}))
+		sessionIdentity{Username: "admin", Role: roleAdmin}))
 	recorder := httptest.NewRecorder()
 	libraryJobResume(recorder, request, Config{Store: store})
 	if recorder.Code != http.StatusConflict {

internal/httpapi/server.go

@@ -135,6 +135,11 @@ type Config struct {
 	LibraryAgentSocket string
 	// LibraryCapacityBudgetBytes 是整库归档容量预算；0 = 未设置（容量门禁不放行）。
 	LibraryCapacityBudgetBytes int64
+	// 整库生产目标的只读探测配置：首次发布要能证明"生产可读且为空"。
+	LibraryProductionDatabaseURL string
+	LibraryProductionInternalKB  string
+	LibraryProductionExternalKB  string
+	LibraryProductionTenantID    uint64
 	// LibraryRetentionVersions/Days 是保留策略；都为 0 = 未设置（不生成清理候选）。
 	LibraryRetentionVersions         int
 	LibraryRetentionDays             int

internal/libraryrelease/release.go

@@ -90,6 +90,16 @@ type Recorder interface {
 	Stage(ctx context.Context, stage string, state string, detail string)
 }
 
+// BackupObserver 让调用方在预备份完成时立刻持久化快照身份：
+// worker 被杀后重启必须能接管这份预备份，而不是重新备份覆盖它。
+type BackupObserver interface {
+	BackupPrepared(ctx context.Context, backup BackupRef)
+}
+
+// FinalizeStages 是"写入完成之后的收尾"：先登记权威版本，再把作业标成功。
+// 收尾失败不能报告整体成功，否则版本登记与作业状态会分叉。
+func FinalizeStages() []string { return []string{"finalize"} }
+
 type recorderFunc func(ctx context.Context, stage, state, detail string)
 
 func (f recorderFunc) Stage(ctx context.Context, stage, state, detail string) {
@@ -167,6 +177,9 @@ func Run(ctx context.Context, request Request, target Target, backups BackupStor
 		return outcome
 	}
 	outcome.Backup = backup
+	if observer, ok := recorder.(BackupObserver); ok {
+		observer.BackupPrepared(ctx, backup)
+	}
 	recorder.Stage(ctx, "prebackup", "succeeded", "预备份 "+backup.ID)
 
 	for _, stage := range Stages() {

internal/platform/postgres/library_jobs.go

@@ -37,7 +37,8 @@ RETURNING id`,
 	if err != nil {
 		var pgErr *pgconn.PgError
 		if errors.As(err, &pgErr) && pgErr.Code == "23505" {
-			if pgErr.ConstraintName == "library_jobs_active" {
+			// 同一目标的写作业（sync/restore）共用一个唯一索引，跨 kind 互斥。
+			if pgErr.ConstraintName == "library_jobs_write_target" || pgErr.ConstraintName == "library_jobs_active" {
 				return governance.LibraryJob{}, false, governance.ErrLibraryJobActive
 			}
 		}
@@ -246,6 +247,40 @@ ON CONFLICT (id) DO NOTHING`,
 	return tx.Commit(ctx)
 }
 
+// InterruptedLibraryJobs 找出"业务作业 running、但队列里已没有活跃作业"的作业：
+// worker 被杀/重启后它们需要被接管，而不是被当成失败丢掉已完成的预备份。
+func (s *Store) InterruptedLibraryJobs(ctx context.Context) ([]governance.LibraryJob, error) {
+	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'))
+ORDER BY lj.created_at`)
+	if err != nil {
+		return nil, fmt.Errorf("list interrupted library jobs: %w", err)
+	}
+	defer rows.Close()
+	ids := make([]string, 0, 4)
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			return nil, err
+		}
+		ids = append(ids, id)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	jobs := make([]governance.LibraryJob, 0, len(ids))
+	for _, id := range ids {
+		job, err := s.LibraryJob(ctx, id)
+		if err != nil {
+			return nil, err
+		}
+		jobs = append(jobs, job)
+	}
+	return jobs, nil
+}
+
 // ReconcileLibraryJobs 收拾"业务作业还在飞、队列里已经没有对应作业"的孤儿：
 // 它们会永久占用 (scope,kind) 唯一索引，让同一目标再也发不出作业。
 // 只标失败并写明恢复责任，不猜结果、不重放。

internal/platform/postgres/migrations/000069_library_write_target_mutex.down.sql

@@ -0,0 +1,3 @@
+DROP INDEX IF EXISTS library_jobs_write_target;
+CREATE UNIQUE INDEX library_jobs_active ON library_jobs(scope, kind)
+    WHERE state IN ('pending','running');

internal/platform/postgres/migrations/000069_library_write_target_mutex.up.sql

@@ -0,0 +1,6 @@
+-- 同一目标的写作业（同步与恢复）必须互斥：两种 kind 同时写同一目标会互相覆盖。
+-- 原来的唯一索引是 (scope, kind)，允许 sync 与 restore 同时存在，这里改成只按 scope。
+DROP INDEX IF EXISTS library_jobs_active;
+CREATE UNIQUE INDEX library_jobs_write_target
+    ON library_jobs(scope)
+    WHERE state IN ('pending','running') AND kind IN ('sync','restore');

internal/platform/postgres/migrations_test.go

@@ -11,26 +11,25 @@ func TestEmbeddedMigrationsAreVersionSorted(t *testing.T) {
 	if err != nil {
 		t.Fatalf("loadMigrations() error = %v", err)
 	}
-	if len(migrations) != 68 {
-		t.Fatalf("migration count = %d, want 68", len(migrations))
+	if len(migrations) != 69 {
+		t.Fatalf("migration count = %d, want 69", len(migrations))
 	}
-	if tail := migrations[len(migrations)-1]; tail.Version != 68 ||
-		!strings.Contains(tail.UpSQL, "restore") {
-		t.Fatalf("migration 68 = %#v", tail)
+	if tail := migrations[len(migrations)-1]; tail.Version != 69 ||
+		!strings.Contains(tail.UpSQL, "library_jobs_write_target") {
+		t.Fatalf("migration 69 = %#v", tail)
 	}
-	if last := migrations[len(migrations)-2]; last.Version != 67 ||
+	if restore := migrations[len(migrations)-2]; restore.Version != 68 || !strings.Contains(restore.UpSQL, "restore") {
+		t.Fatalf("migration 68 = %#v", restore)
+	}
+	if last := migrations[len(migrations)-3]; last.Version != 67 ||
 		!strings.Contains(last.UpSQL, "library_release_authorizations") ||
 		!strings.Contains(last.UpSQL, "params_digest") {
 		t.Fatalf("migration 67 = %#v", last)
 	}
-	if prev := migrations[len(migrations)-3]; prev.Version != 66 ||
-		!strings.Contains(prev.UpSQL, "CREATE TABLE library_versions") ||
-		!strings.Contains(prev.UpSQL, "library_jobs_active") {
-		t.Fatalf("migration 66 = %#v", prev)
-	}
-	if auth := migrations[len(migrations)-2]; auth.Version != 67 ||
-		!strings.Contains(auth.UpSQL, "library_release_authorizations") {
-		t.Fatalf("migration 67 = %#v", auth)
+	if flow := migrations[len(migrations)-4]; flow.Version != 66 ||
+		!strings.Contains(flow.UpSQL, "CREATE TABLE library_versions") ||
+		!strings.Contains(flow.UpSQL, "library_jobs_active") {
+		t.Fatalf("migration 66 = %#v", flow)
 	}
 	if migrations[0].Version != 1 || !strings.Contains(migrations[0].UpSQL, "CREATE TABLE IF NOT EXISTS jobs") {
 		t.Fatalf("migration 1 = %#v", migrations[0])

internal/releaseadapter/v071/library_archive.go

@@ -21,7 +21,10 @@ func ArchiveLibraryCandidate(ctx context.Context, archive *librarysnapshot.Archi
 	if err := validateLibraryCandidate(identity, pg, es); err != nil {
 		return "", err
 	}
-	if es.MappingSHA256 != contract.IndexMappingSHA256 {
+	// 空生产基线：索引不存在时没有映射可比，契约仍按目标快照记录；
+	// 只要这次导出的文档是零条，就不存在"拿错映射"的风险。
+	emptyIndex := len(es.Documents) == 0
+	if !emptyIndex && es.MappingSHA256 != contract.IndexMappingSHA256 {
 		return "", errors.New("library mapping contract mismatch")
 	}
 	manifest := librarysnapshot.Manifest{ValidationPolicy: librarysnapshot.ArtifactAcceptancePolicy, Format: librarysnapshot.Format, Scope: scope, TenantID: identity.TenantID, KnowledgeBaseID: identity.KnowledgeBaseID, CreatedAt: createdAt, Compatibility: contract}
@@ -246,7 +249,12 @@ func validateLibraryCandidate(identity PostgresIdentity, pg PostgresSnapshot, es
 		}
 	}
 	sha, err := searchDocumentsDigest(es.Documents)
-	if err != nil || !validLowerSHA256(es.MappingSHA256) || sha != es.DocumentsSHA256 {
+	if err != nil || sha != es.DocumentsSHA256 {
+		return errors.New("invalid library index digest")
+	}
+	// 索引不存在（空生产基线）时没有映射可比：只允许"零文档 + 空映射"这种组合，
+	// 有文档却没有映射仍然是错误。
+	if !validLowerSHA256(es.MappingSHA256) && (es.MappingSHA256 != "" || len(es.Documents) != 0) {
 		return errors.New("invalid library index digest")
 	}
 	seen := map[string]bool{}

internal/releaseadapter/v071/library_backup_store.go

@@ -124,10 +124,18 @@ func (s *ProductionBackupStore) Prepare(ctx context.Context, _ string, current l
 	if err != nil {
 		return libraryrelease.BackupRef{}, fmt.Errorf("读取生产索引失败: %w", err)
 	}
-	if es.MappingSHA256 != s.compatibility.IndexMappingSHA256 {
-		// 契约不符说明生产换了索引映射：此时备份出来的东西无法用同一套恢复，
-		// 宁可失败也不产出"看起来完整"的预备份。
-		return libraryrelease.BackupRef{}, errors.New("生产索引映射与归档契约不一致")
+	if es.MappingSHA256 == "" {
+		// 生产索引不存在：只有数据库也为空时才是合法的"空基线"，否则说明索引丢了。
+		if len(pg.Table("knowledges").Rows) != 0 || len(pg.Table("chunks").Rows) != 0 {
+			return libraryrelease.BackupRef{}, errors.New("生产索引缺失但数据库非空：先修索引再发布")
+		}
+	} else if es.MappingSHA256 != s.compatibility.IndexMappingSHA256 {
+		// 契约不符说明生产换了索引映射：此时备份出来的东西无法用同一套恢复。
+		// 唯一例外是"空生产"：零文档零分块时映射差异不影响可恢复性。
+		emptyProduction := len(pg.Table("knowledges").Rows) == 0 && len(pg.Table("chunks").Rows) == 0 && len(es.Documents) == 0
+		if !emptyProduction {
+			return libraryrelease.BackupRef{}, errors.New("生产索引映射与归档契约不一致")
+		}
 	}
 	// ArchiveLibraryCandidate 自己按 resources 行读取对象字节并计算摘要，
 	// 所以这里直接给它生产对象读取器，不再重复导出一次资源。

internal/releaseadapter/v071/library_production_isolation_test.go

@@ -160,7 +160,7 @@ INSERT INTO wiki_pages(id,tenant_id,knowledge_base_id,slug,content,deleted_at) V
 	defer target.Close()
 
 	// R1：资源范围必须只含本租户本库（chunk 绑定也要算），排除别的租户/KB。
-	paths, err := target.resourcePaths(ctx)
+	paths, err := target.resourcePaths(ctx, []string{"res-safe", "other-scope", "res-unbound"})
 	if err != nil {
 		t.Fatalf("resource paths: %v", err)
 	}

internal/releaseadapter/v071/library_production_target.go

@@ -64,9 +64,11 @@ func (config ProductionLibraryConfig) validate(scope ProductionScope) error {
 }
 
 // 依赖顺序：插入按正序，删除按倒序。
+// 真实 Schema 有外键（resource_bindings.resource_id → resources.id），
+// 所以父表 resources 必须排在子表 resource_bindings 之前。
 var productionLibraryTables = []string{
 	"knowledges", "chunks", "knowledge_tags", "knowledge_tag_relations",
-	"resource_bindings", "resources", "wiki_folders", "wiki_pages",
+	"resources", "resource_bindings", "wiki_folders", "wiki_pages",
 	"wiki_page_issues", "wiki_log_entries",
 }
 
@@ -131,7 +133,10 @@ func (t *ProductionLibraryTarget) Current(ctx context.Context) (libraryrelease.C
 	}
 	indexCount, err := t.search.Count(ctx, SearchIdentity{TenantID: t.scope.TenantID,
 		KnowledgeBaseID: t.scope.KnowledgeBaseID})
-	if err != nil {
+	if errors.Is(err, errIndexMissing) {
+		// 生产索引不存在 = 空基线（零文档）；不是故障，但也不会被当成"有内容"。
+		indexCount = 0
+	} else if err != nil {
 		return libraryrelease.CurrentState{}, err
 	}
 	state.Counts["index"] = indexCount
@@ -144,7 +149,9 @@ func (t *ProductionLibraryTarget) Current(ctx context.Context) (libraryrelease.C
 func scopeWhere(table string) string {
 	switch table {
 	case "knowledge_tag_relations":
-		return `tenant_id=$1 AND knowledge_id IN (SELECT id FROM knowledges k
+		// 真实 WeKnora 的这张表只有 knowledge_id/tag_id/created_at，
+		// 没有 tenant_id/knowledge_base_id：归属只能走父表 knowledges 判定。
+		return `knowledge_id IN (SELECT id FROM knowledges k
  WHERE k.tenant_id=$1 AND k.knowledge_base_id=$2)`
 	case "resource_bindings":
 		return `tenant_id=$1 AND ((owner_type='knowledge' AND owner_id IN (SELECT id FROM knowledges k
@@ -153,14 +160,22 @@ func scopeWhere(table string) string {
 	case "resources":
 		// 必须把外层 resources.id 与绑定关系精确关联：只按 owner_id 判断会命中整张
 		// 资源表；同时要覆盖 chunk 绑定（图片/附件常只挂在分块上）并限制绑定租户。
-		return `EXISTS (
+		// 两种归属都要算：显式绑定（knowledge/chunk）+ 被本库 chunk 正文以
+		// resource://<handle> 引用（图片/附件常见，可能没有绑定行）。
+		// 与归档侧收集资源部件的规则保持一致，否则校验会把导入的资源算成"多余"。
+		return `(EXISTS (
  SELECT 1 FROM resource_bindings b
  JOIN knowledges k ON k.id=b.owner_id AND k.tenant_id=b.tenant_id
  WHERE b.resource_id=resources.id AND b.tenant_id=$1 AND b.owner_type='knowledge' AND k.knowledge_base_id=$2
  UNION ALL
  SELECT 1 FROM resource_bindings b
  JOIN chunks c ON c.id=b.owner_id AND c.tenant_id=b.tenant_id
- WHERE b.resource_id=resources.id AND b.tenant_id=$1 AND b.owner_type='chunk' AND c.knowledge_base_id=$2)`
+ WHERE b.resource_id=resources.id AND b.tenant_id=$1 AND b.owner_type='chunk' AND c.knowledge_base_id=$2)
+ OR EXISTS (
+ SELECT 1 FROM chunks c
+ WHERE c.tenant_id=$1 AND c.knowledge_base_id=$2 AND c.deleted_at IS NULL
+   AND coalesce(resources.handle,'') <> ''
+   AND position('resource://'||resources.handle IN coalesce(c.content,'')) > 0))`
 	default:
 		return "tenant_id=$1 AND knowledge_base_id=$2"
 	}
@@ -349,7 +364,7 @@ type resourceLocation struct {
 }
 
 func (t *ProductionLibraryTarget) importResources(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest) error {
-	paths, err := t.resourcePaths(ctx)
+	paths, err := t.resourcePaths(ctx, resourcePartNames(manifest))
 	if err != nil {
 		return err
 	}
@@ -371,7 +386,7 @@ func (t *ProductionLibraryTarget) importResources(ctx context.Context, archive *
 			return fmt.Errorf("资源 %s 字节与归档摘要不符", part.Name)
 		}
 		if err := t.config.Objects.Put(ctx, location.Provider, location.Path, bytes.NewReader(body), int64(len(body))); err != nil {
-			return fmt.Errorf("上传资源 %s 失败", part.Name)
+			return fmt.Errorf("上传资源 %s 失败: %w", part.Name, err)
 		}
 	}
 	return nil
@@ -380,7 +395,7 @@ func (t *ProductionLibraryTarget) importResources(ctx context.Context, archive *
 // verifyObjects 逐个比对生产对象的实际字节摘要与归档摘要：
 // 只有"对象真的在、内容一致"才算通过，防止对象丢失或损坏被行数掩盖。
 func (t *ProductionLibraryTarget) verifyObjects(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest) error {
-	paths, err := t.resourcePaths(ctx)
+	paths, err := t.resourcePaths(ctx, resourcePartNames(manifest))
 	if err != nil {
 		return err
 	}
@@ -409,11 +424,25 @@ func (t *ProductionLibraryTarget) verifyObjects(ctx context.Context, archive *li
 	return nil
 }
 
-func (t *ProductionLibraryTarget) resourcePaths(ctx context.Context) (map[string]resourceLocation, error) {
-	// 归档里的资源部件名就是资源 ID（见 ArchiveLibraryCandidate），这里按 id 建映射，
-	// 否则导入/校验会按 handle 找不到对象。
-	rows, err := t.pool.Query(ctx, fmt.Sprintf(`SELECT id, physical_path, provider FROM resources
-WHERE %s`, scopeWhere("resources")), t.scope.TenantID, t.scope.KnowledgeBaseID)
+func resourcePartNames(manifest librarysnapshot.Manifest) []string {
+	names := make([]string, 0, 16)
+	for _, part := range manifest.Parts {
+		if part.Kind == "resource" {
+			names = append(names, part.Name)
+		}
+	}
+	return names
+}
+
+// resourcePaths 只在"本次要写的这些资源 ID"里解析路径，并且强制租户。
+// 不按绑定范围扫全表：被 chunk 正文以 resource:// 引用的资源可能没有绑定行，
+// 但它是这份快照的一部分；反过来按绑定扫会漏掉它们，按全表扫会读到别的库。
+func (t *ProductionLibraryTarget) resourcePaths(ctx context.Context, ids []string) (map[string]resourceLocation, error) {
+	if len(ids) == 0 {
+		return map[string]resourceLocation{}, nil
+	}
+	rows, err := t.pool.Query(ctx, `SELECT id, coalesce(physical_path,''), coalesce(provider,'') FROM resources
+WHERE tenant_id=$1 AND id = ANY($2::text[])`, t.scope.TenantID, ids)
 	if err != nil {
 		return nil, fmt.Errorf("读取生产资源路径失败: %w", err)
 	}

internal/releaseadapter/v071/library_search.go

@@ -17,6 +17,15 @@ func (e *SearchExporter) ExportLibrary(ctx context.Context, identity SearchIdent
 		return SearchSnapshot{}, errors.New("invalid library search scope or limit")
 	}
 	inspection, err := e.inspectMapping(ctx)
+	if errors.Is(err, errIndexMissing) {
+		// 空生产基线：索引不存在 = 零文档、无映射可比。
+		// 调用方必须再确认数据库也为空，不能只看这一条。
+		emptyDigest, digestErr := searchDocumentsDigest([]SearchDocument{})
+		if digestErr != nil {
+			return SearchSnapshot{}, digestErr
+		}
+		return SearchSnapshot{Documents: []SearchDocument{}, MappingSHA256: "", DocumentsSHA256: emptyDigest}, nil
+	}
 	if err != nil {
 		return SearchSnapshot{}, err
 	}

internal/releaseadapter/v071/library_verify_content.go

@@ -128,6 +128,46 @@ func archivedTableContent(archive *librarysnapshot.Archive, part librarysnapshot
 	return int64(len(digests)), combineRowDigests(digests), columns, nil
 }
 
+// firstIndexDifference 返回第一处不一致的文档 ID 与字段，便于定位。
+func firstIndexDifference(archived, live []SearchDocument) string {
+	liveByID := make(map[string]json.RawMessage, len(live))
+	for _, document := range live {
+		liveByID[document.ID] = document.Source
+	}
+	seen := make(map[string]bool, len(archived))
+	for _, document := range archived {
+		seen[document.ID] = true
+		current, ok := liveByID[document.ID]
+		if !ok {
+			return document.ID + "（生产缺少）"
+		}
+		var want, got map[string]json.RawMessage
+		if json.Unmarshal(document.Source, &want) != nil || json.Unmarshal(current, &got) != nil {
+			return document.ID + "（无法解析）"
+		}
+		for key, value := range want {
+			other, exists := got[key]
+			if !exists {
+				return fmt.Sprintf("%s（生产缺少字段 %s）", document.ID, key)
+			}
+			if string(value) != string(other) {
+				return fmt.Sprintf("%s（字段 %s 不同：归档 %.60s 生产 %.60s）", document.ID, key, value, other)
+			}
+		}
+		for key := range got {
+			if _, exists := want[key]; !exists {
+				return fmt.Sprintf("%s（生产多出字段 %s）", document.ID, key)
+			}
+		}
+	}
+	for _, document := range live {
+		if !seen[document.ID] {
+			return document.ID + "（生产多余）"
+		}
+	}
+	return "数量一致但摘要不同"
+}
+
 // verifyIndexDigest 比对归档索引文档与生产实际索引文档的规范化摘要。
 func (t *ProductionLibraryTarget) verifyIndexDigest(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest) error {
 	for _, part := range manifest.Parts {
@@ -145,7 +185,19 @@ func (t *ProductionLibraryTarget) verifyIndexDigest(ctx context.Context, archive
 		}
 		archived := make([]SearchDocument, 0, len(documents))
 		for _, document := range documents {
-			canonical, err := canonicalJSON(document.Source)
+			// 写入时 BulK Index 会把 tenant_id/knowledge_base_id 覆盖成本目标的身份，
+			// 校验必须用同样的规范化，否则"写对了"也会被判不一致。
+			source := map[string]json.RawMessage{}
+			if err := json.Unmarshal(document.Source, &source); err != nil {
+				return errors.New("归档索引文档不是对象")
+			}
+			source["tenant_id"], _ = json.Marshal(t.scope.TenantID)
+			source["knowledge_base_id"], _ = json.Marshal(t.scope.KnowledgeBaseID)
+			encoded, err := json.Marshal(source)
+			if err != nil {
+				return err
+			}
+			canonical, err := canonicalJSON(encoded)
 			if err != nil {
 				return errors.New("归档索引文档无法规范化")
 			}
@@ -161,8 +213,9 @@ func (t *ProductionLibraryTarget) verifyIndexDigest(ctx context.Context, archive
 			return fmt.Errorf("读取生产索引失败: %w", err)
 		}
 		if len(snapshot.Documents) != len(archived) || snapshot.DocumentsSHA256 != wantDigest {
-			return fmt.Errorf("生产索引与快照不一致：文档 %d/%d，内容摘要不一致",
-				len(snapshot.Documents), len(archived))
+			// 摘要不一致时给出第一处差异文档，运维才能定位是"少了/多了/被改了"。
+			return fmt.Errorf("生产索引与快照不一致：文档 %d/%d，首个差异 %s",
+				len(snapshot.Documents), len(archived), firstIndexDifference(archived, snapshot.Documents))
 		}
 	}
 	return nil

internal/releaseadapter/v071/postgres_import.go

@@ -493,6 +493,11 @@ func validSQLIdentifier(value string) bool {
 	return true
 }
 
+// validObjectLocator 接受两种真实形态：
+//   - 无 scheme 的对象键："weknora/10000/<doc>/<file>.docx"
+//   - 带 scheme 的定位符："oss://<bucket>/<key>"（WeKnora 的资源路径就是这个形状）
+//
+// 只校验结构，不改值；真正的 bucket/key 拆分由 providerPathFromLocator 做。
 func validObjectLocator(value string) bool {
 	if value == "" || len(value) > 2048 || value != strings.TrimSpace(value) || strings.ContainsAny(value, "\x00\r\n\\?#") || strings.HasPrefix(value, "/") {
 		return false
@@ -501,7 +506,12 @@ func validObjectLocator(value string) bool {
 		if !validSearchIdentifier(value[:separator]) {
 			return false
 		}
-		value = value[separator+3:]
+		remainder := value[separator+3:]
+		boundary := strings.Index(remainder, "/")
+		if boundary <= 0 || !validSearchIdentifier(remainder[:boundary]) {
+			return false
+		}
+		value = remainder[boundary+1:]
 	}
 	cleaned := path.Clean(value)
 	return cleaned == value && cleaned != "." && cleaned != ".." && !strings.HasPrefix(cleaned, "../")

internal/releaseadapter/v071/s3_object_store.go

@@ -5,6 +5,7 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"errors"
+	"fmt"
 	"io"
 	"net/http"
 	"net/url"
@@ -96,11 +97,17 @@ func (store *S3ObjectStore) Open(ctx context.Context, provider, physicalPath str
 }
 
 func (store *S3ObjectStore) Put(ctx context.Context, provider, objectPath string, reader io.Reader, size int64) error {
-	if provider != store.provider || !validObjectLocator(objectPath) || size < 0 || size > store.maxObjectBytes {
+	if provider != store.provider || !validObjectKey(objectPath) || size < 0 || size > store.maxObjectBytes {
 		return errors.New("invalid release object write")
 	}
-	if _, err := store.client.PutObject(ctx, &s3.PutObjectInput{Bucket: aws.String(store.bucket), Key: aws.String(objectPath), Body: reader, ContentLength: aws.Int64(size)}); err != nil {
-		return errors.New("write release object")
+	// 真实资源路径是 storage://<backend>/oss://<bucket>/<key>：
+	// 落盘只写配置的桶，所以必须先把定位符拆成裸 key，不能整串当 key。
+	key, ok := providerPathFromLocator(objectPath)
+	if !ok {
+		return errors.New("invalid release object write")
+	}
+	if _, err := store.client.PutObject(ctx, &s3.PutObjectInput{Bucket: aws.String(store.bucket), Key: aws.String(key), Body: reader, ContentLength: aws.Int64(size)}); err != nil {
+		return fmt.Errorf("write release object: %w", err)
 	}
 	return nil
 }

internal/releaseadapter/v071/search.go

@@ -64,6 +64,9 @@ type searchFieldNames struct {
 
 // mappingInspection is the canonical digest of the index mappings plus the
 // logical scope field names derived from the real mapping.
+// errIndexMissing 表示目标索引根本不存在（空生产基线）。
+var errIndexMissing = errors.New("Elasticsearch index missing")
+
 type mappingInspection struct {
 	digest string
 	fields searchFieldNames
@@ -74,6 +77,9 @@ type mappingInspection struct {
 func (exporter *SearchExporter) inspectMapping(ctx context.Context) (mappingInspection, error) {
 	var response map[string]json.RawMessage
 	if err := exporter.requestJSON(ctx, http.MethodGet, "/"+exporter.index+"/_mapping", nil, &response); err != nil {
+		if errors.Is(err, errIndexMissing) {
+			return mappingInspection{}, errIndexMissing
+		}
 		return mappingInspection{}, err
 	}
 	indexRaw, ok := response[exporter.index]
@@ -170,6 +176,14 @@ func (exporter *SearchExporter) Export(ctx context.Context, identity SearchIdent
 		return SearchSnapshot{}, errors.New("invalid search export identity")
 	}
 	inspection, err := exporter.inspectMapping(ctx)
+	if errors.Is(err, errIndexMissing) {
+		// 索引不存在 = 空基线：零文档、无映射可比。调用方必须再确认数据库也是空的。
+		emptyDigest, digestErr := searchDocumentsDigest([]SearchDocument{})
+		if digestErr != nil {
+			return SearchSnapshot{}, digestErr
+		}
+		return SearchSnapshot{Documents: []SearchDocument{}, MappingSHA256: "", DocumentsSHA256: emptyDigest}, nil
+	}
 	if err != nil {
 		return SearchSnapshot{}, err
 	}
@@ -254,6 +268,11 @@ func (exporter *SearchExporter) requestJSON(ctx context.Context, method, path st
 	if response.Request == nil || response.Request.URL.Scheme != exporter.baseURL.Scheme || response.Request.URL.Host != exporter.baseURL.Host {
 		return errors.New("Elasticsearch response origin mismatch")
 	}
+	if response.StatusCode == http.StatusNotFound {
+		// 索引不存在是"空生产"的正常状态，不是故障；调用方必须显式处理，
+		// 不能把 404 当成功，也不能把它当传输错误。
+		return fmt.Errorf("%w: HTTP 404", errIndexMissing)
+	}
 	if response.StatusCode < 200 || response.StatusCode >= 300 {
 		return fmt.Errorf("Elasticsearch returned HTTP %d", response.StatusCode)
 	}