Merge pull request 'fix(library): 跨源同步按目标范围改写并校验残留' (#13 ) from fix/library-cross-scope-sync-residue into main

fix(library): 跨范围同步按目标范围改写并校验残留
真实事故：源 KB(10000/0c84a098) 同步到目标 KB(10001/e64e82b5) 时，归档行被原样插入目标库，落在来源租户/知识库；资源阶段随即报"生产缺少资源 ... 的存储路径" （资源查询按目标租户过滤，看不见刚写进去的行）。回滚只按目标范围回放，于是页面报"已恢复到预备份"，目标却多出 4 文件/38 分块/21 资源/4 绑定。改： - 写入与校验共用 retargetRow：tenant_id / knowledge_base_id / 资源定位符(后端、 provider、桶、租户前缀) / storage_backend_id / provider / embedding_model_id 统一映射到目标范围；行里的来源身份对不上就失败，宁可不写。 - Verify 增加"本次写集合仍以来源身份残留"判定；Restore 增加整库指纹比对（预备份时刻 vs 恢复后）与孤儿资源清理，回滚后用 Verify 复核，恢复工具退出 0 不再等于恢复完成。 - 新增 cmd/libraryresidue：按归档身份精确、可备份、可复核地清理已知的范围外残留。验证：make verify 全绿；真实 PG+ES 用例复现"源目标 KB 不同 + 目标非空 + 资源阶段失败"，断言回滚后无来源范围残留、其它租户行未动、恢复后内容=预备份；正向用例断言行落在目标范围且资源定位符指向目标存储。
2026-10-29 22:13:49 +08:00 · 2026-10-29 22:11:59 +08:00
11 changed files with 1588 additions and 70 deletions
@@ -0,0 +1,115 @@
+// command libraryresidue 处理"整库同步写到目标范围外"留下的残留行：
+//
+//	读指定快照（来源范围）→ 在目标里找同内容、仍带来源身份的行
+//	→ 先备份成 0600 文件并记录摘要 → 精确按归档身份删除 → 复核来源身份已空、
+//	目标内容与预备份一致、整库指纹里没有别的租户
+//
+// 它不做范围广删，也不改任何别的东西：判据是"归档里确实有这些行、内容逐字段
+// 一致"，任何一行对不上就整体拒绝。默认是干跑（只报告、不删）。
+//
+// 真实事故：跨范围同步把 4 文件/38 分块/21 资源/4 绑定写成来源租户的行，
+// 回滚只按目标范围回放，于是"已恢复"却留下这些行。这个工具的用途就是在
+// 修复后的代码上线前，用可审计的方式清掉它们。
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"log"
+	"os"
+	"strconv"
+	"strings"
+
+	"knowledge-governance/internal/appconfig"
+	"knowledge-governance/internal/libraryrelease"
+	"knowledge-governance/internal/librarysnapshot"
+	"knowledge-governance/internal/librarystack"
+	v071 "knowledge-governance/internal/releaseadapter/v071"
+)
+
+func main() {
+	scope := flag.String("scope", "internal", "internal|external")
+	snapshot := flag.String("snapshot", "", "sealed snapshot whose source scope is the residue (required)")
+	backupSnapshot := flag.String("backup-snapshot", "", "prebackup snapshot the target must be restored to (required)")
+	expectSource := flag.String("expect-source", "", "expected source scope as <tenant>/<knowledge-base> (required guard against typos)")
+	out := flag.String("out", "", "directory for the residue backup (required)")
+	apply := flag.Bool("apply", false, "actually delete the residue rows (default: dry run)")
+	reportPath := flag.String("report", "", "write the JSON report here")
+	flag.Parse()
+	if *snapshot == "" || *backupSnapshot == "" || *out == "" || *expectSource == "" {
+		log.Fatal("snapshot, backup-snapshot, out and expect-source are required")
+	}
+	config, err := appconfig.Load(os.Getenv)
+	if err != nil {
+		log.Fatalf("load config: %v", err)
+	}
+	ctx := context.Background()
+	report, err := run(ctx, config, *scope, *snapshot, *backupSnapshot, *expectSource, *out, *apply)
+	if *reportPath != "" && report != nil {
+		raw, marshalErr := json.MarshalIndent(report, "", "  ")
+		if marshalErr == nil {
+			raw = append(raw, '\n')
+			if writeErr := os.WriteFile(*reportPath, raw, 0o600); writeErr != nil {
+				log.Printf("report write failed: %v", writeErr)
+			}
+		}
+	}
+	if err != nil {
+		log.Fatalf("residue cleanup failed: %v", err)
+	}
+	log.Printf("residue cleanup ok: source=%s/%s tables=%v cleared=%d applied=%v verified=%v",
+		reportSource(report), report.SourceKB, report.Tables, report.Cleared, report.Applied, report.Verified)
+}
+
+func reportSource(report *v071.ResidueReport) string {
+	return strconv.FormatUint(report.SourceTenant, 10)
+}
+
+func run(ctx context.Context, config appconfig.Config, scope, snapshotID, backupSnapshotID, expectSource, out string,
+	apply bool) (*v071.ResidueReport, error) {
+	if scope != "internal" && scope != "external" {
+		return nil, fmt.Errorf("scope must be internal or external")
+	}
+	archive, err := librarysnapshot.Open(config.LibraryArchiveRoot, 256*1024*1024)
+	if err != nil {
+		return nil, fmt.Errorf("归档不可用: %w", err)
+	}
+	manifest, err := archive.Load(snapshotID)
+	archive.Close()
+	if err != nil {
+		return nil, fmt.Errorf("指定快照不可读: %w", err)
+	}
+	source := v071.ProductionScope{TenantID: manifest.TenantID, KnowledgeBaseID: manifest.KnowledgeBaseID}
+	want := strconv.FormatUint(source.TenantID, 10) + "/" + source.KnowledgeBaseID
+	// 范围写错会把"别人的行"当成残留：必须由调用方显式确认。
+	if strings.TrimSpace(expectSource) != want {
+		return nil, fmt.Errorf("来源范围与 -expect-source 不一致：归档为 %s，参数为 %s", want, expectSource)
+	}
+	target, _, err := librarystack.OpenProductionStack(ctx, config, scope, backupSnapshotID, "")
+	if err != nil {
+		return nil, fmt.Errorf("装配目标失败: %w", err)
+	}
+	defer target.Close()
+	report, err := target.ClearForeignScope(ctx, source, snapshotID, out, apply)
+	if err != nil {
+		return report, err
+	}
+	report.Foreign, _ = target.ForeignScopes(ctx)
+	if !apply {
+		return report, nil
+	}
+	if err := target.VerifyResidueCleared(ctx, source); err != nil {
+		return report, err
+	}
+	if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: backupSnapshotID}); err != nil {
+		return report, fmt.Errorf("清理后目标内容仍与预备份不一致（保持停写）: %w", err)
+	}
+	report.Verified = true
+	if len(report.Foreign) > 0 {
+		raw, _ := json.Marshal(report.Foreign)
+		return report, fmt.Errorf("清理后目标仍有其它租户的数据，需人工确认才能开放：%s", raw)
+	}
+	return report, nil
+}
@@ -59,6 +59,14 @@ type Expectation struct {
 	Digests    map[string]string
 }

+// TableFingerprint 是一张知识域表在**整实例**上的行数与身份集合摘要（不限租户/
+// 知识库）。恢复后必须与预备份时刻逐表一致：只比目标范围会让"写到范围外"的
+// 残留被当成已恢复（真实事故：恢复报成功，目标仍多 67 行）。
+type TableFingerprint struct {
+	Count  int64  `json:"count"`
+	Digest string `json:"digest"`
+}
+
 // BackupRef 指向一份可独立恢复的完整预备份。
 type BackupRef struct {
 	ID string `json:"id"`
@@ -67,6 +75,9 @@ type BackupRef struct {
 	SnapshotID string    `json:"snapshot_id"`
 	Bytes      int64     `json:"bytes"`
 	CreatedAt  time.Time `json:"created_at"`
+	// Universe 是预备份时刻的整实例指纹，恢复后逐表比对。空表示这次没有采集
+	// （例如 worker 被杀后用已持久化的快照接管），此时恢复只能靠内容校验兜底。
+	Universe map[string]TableFingerprint `json:"universe,omitempty"`
 }

 type Target interface {
@@ -224,8 +235,16 @@ func failAndRestore(ctx context.Context, outcome Outcome, target Target, recorde
 		recorder.Stage(ctx, "rollback", "failed", outcome.Rollback)
 		return outcome
 	}
+	// "恢复工具跑完"不是"恢复完成"：必须再校验一次目标内容与预备份一致，
+	// 否则页面会把半恢复状态报成已恢复（真实事故就是这样放开了维护）。
+	if verifyErr := target.Verify(ctx, Expectation{SnapshotID: outcome.Backup.SnapshotID}); verifyErr != nil {
+		outcome.State = "rollback_failed"
+		outcome.Rollback = "恢复后校验失败（目标可能仍是半写状态）：" + verifyErr.Error()
+		recorder.Stage(ctx, "rollback", "failed", outcome.Rollback)
+		return outcome
+	}
 	outcome.State = "rolled_back"
-	outcome.Rollback = "已恢复到预备份 " + outcome.Backup.ID
+	outcome.Rollback = "已恢复到预备份 " + outcome.Backup.ID + "，并逐表/索引/资源复核一致"
 	recorder.Stage(ctx, "rollback", "succeeded", outcome.Rollback)
 	return outcome
 }
@@ -13,9 +13,14 @@ type fakeTarget struct {
 	applyErr   map[Stage]error
 	verifyErr  error
 	restoreErr error
-	applied    []Stage
-	restored   int
-	currentN   int
+	// 恢复之后校验是否通过：用来区分"恢复跑完"和"恢复已验证"。
+	verifyPassesAfterRestore bool
+	applied                  []Stage
+	restored                 int
+	currentN                 int
+	verifyN                  int
+	// 记录最后一次 Verify 是否发生在 Restore 之后（恢复必须被校验兜底）。
+	verifiedAfterRestore bool
 }

 func (f *fakeTarget) Current(context.Context) (CurrentState, error) {
@@ -35,7 +40,17 @@ func (f *fakeTarget) Apply(_ context.Context, stage Stage, _ string) error {
 	return nil
 }

-func (f *fakeTarget) Verify(context.Context, Expectation) error { return f.verifyErr }
+func (f *fakeTarget) Verify(context.Context, Expectation) error {
+	f.verifyN++
+	f.verifiedAfterRestore = f.restored > 0
+	if f.verifyErr == nil {
+		return nil
+	}
+	if f.verifyPassesAfterRestore && f.restored > 0 {
+		return nil
+	}
+	return f.verifyErr
+}

 func (f *fakeTarget) Restore(context.Context, BackupRef) error {
 	f.restored++
@@ -185,12 +200,39 @@ func TestRunRollsBackWhenStageFails(t *testing.T) {
 	}
 }

-func TestRunRollsBackWhenVerificationFails(t *testing.T) {
-	target := &fakeTarget{verifyErr: errors.New("生产分块数不一致")}
-	outcome := Run(context.Background(), request(time.Now()), target, &fakeBackups{}, nil)
-	if outcome.State != "rolled_back" || target.restored != 1 {
+// 恢复跑完但恢复后校验仍失败：不能报 rolled_back。真实事故里页面就是在这种
+// 状态下显示"已恢复到预备份并验证"，而目标其实还留着 67 行残留。
+func TestRunDoesNotClaimRollbackWhenRecoveryVerificationFails(t *testing.T) {
+	target := &fakeTarget{applyErr: map[Stage]error{StageResources: errors.New("资源写失败")},
+		verifyErr: errors.New("生产分块数不一致")}
+	recorder := &stageLog{}
+	outcome := Run(context.Background(), request(time.Now()), target, &fakeBackups{}, recorder)
+	if outcome.State != "rollback_failed" || target.restored != 1 {
 		t.Fatalf("outcome = %+v restored=%d", outcome, target.restored)
 	}
+	if !strings.Contains(outcome.Rollback, "恢复后校验失败") {
+		t.Fatalf("rollback message = %q", outcome.Rollback)
+	}
+	if !recorder.has("rollback:failed") {
+		t.Fatalf("stage log = %v", recorder.entries)
+	}
+}
+
+// 只有"恢复到预备份且恢复后校验通过"才允许报 rolled_back。
+func TestRunReportsRolledBackOnlyAfterRecoveryVerifies(t *testing.T) {
+	target := &fakeTarget{applyErr: map[Stage]error{StageResources: errors.New("资源写失败")},
+		verifyErr: errors.New("半写状态"), verifyPassesAfterRestore: true}
+	recorder := &stageLog{}
+	outcome := Run(context.Background(), request(time.Now()), target, &fakeBackups{}, recorder)
+	if outcome.State != "rolled_back" {
+		t.Fatalf("outcome = %+v", outcome)
+	}
+	if !target.verifiedAfterRestore {
+		t.Fatal("restore must be followed by a content verification")
+	}
+	if !recorder.has("rollback:succeeded") {
+		t.Fatalf("stage log = %v", recorder.entries)
+	}
 }

 func TestRunReportsRollbackFailureDistinctly(t *testing.T) {
@@ -113,6 +113,12 @@ func (s *ProductionBackupStore) Prepare(ctx context.Context, _ string, current l
 	if err != nil {
 		return libraryrelease.BackupRef{}, err
 	}
+	// 整实例指纹:恢复之后要拿它证明"多写的、少写的都没有"。只按目标范围比,
+	// 写到范围外的行会被漏掉(真实事故:恢复报成功但目标多 67 行)。
+	universeBefore, err := universeFingerprint(ctx, s.pool)
+	if err != nil {
+		return libraryrelease.BackupRef{}, err
+	}
 	pg, err := ExportLibrary(ctx, s.pool, s.identity)
 	if err != nil {
 		return libraryrelease.BackupRef{}, fmt.Errorf("读取生产整库失败: %w", err)
@@ -154,6 +160,13 @@ func (s *ProductionBackupStore) Prepare(ctx context.Context, _ string, current l
 			return libraryrelease.BackupRef{}, fmt.Errorf("预备份期间生产 %s 内容发生变化，切点不一致", table)
 		}
 	}
+	universeAfter, err := universeFingerprint(ctx, s.pool)
+	if err != nil {
+		return libraryrelease.BackupRef{}, err
+	}
+	if diff := universeDiff(universeBefore, universeAfter); diff != "" {
+		return libraryrelease.BackupRef{}, fmt.Errorf("预备份期间整实例知识域发生变化（%s），切点不一致", diff)
+	}
 	for table, want := range current.Counts {
 		if table == "index" {
 			continue
@@ -171,7 +184,7 @@ func (s *ProductionBackupStore) Prepare(ctx context.Context, _ string, current l
 		total += part.Bytes
 	}
 	return libraryrelease.BackupRef{ID: "backup-" + snapshotID, SnapshotID: snapshotID, Bytes: total,
-		CreatedAt: createdAt}, nil
+		CreatedAt: createdAt, Universe: universeBefore}, nil
 }

 func (s *ProductionBackupStore) rowCounts(ctx context.Context) (map[string]int64, error) {
@@ -84,8 +84,16 @@ type ProductionLibraryTarget struct {
 	scope  ProductionScope
 	search *SearchExporter
 	now    func() time.Time
-	// storageBackend 是本次写入使用的目标存储后端 id（写入与校验共用）。
-	storageBackend string
+	// storageBackend / storageProvider 是本次写入使用的目标存储后端与提供方
+	// （写入与校验共用，缺一不可：只换桶不换提供方，目标取文件仍会走来源实例的路）。
+	storageBackend  string
+	storageProvider string
+	// snapshotTenant / snapshotKB 是这次要写入的快照来源身份。写入时按它做
+	// 来源→目标的范围映射；校验时按它反查"目标里还有没有来源范围的行"。
+	snapshotTenant uint64
+	snapshotKB     string
+	// targetEmbeddingModel 是目标知识库绑定的 embedding 模型 id（文档行要换成它）。
+	targetEmbeddingModel string
 }

 // resolveStorageBackend 在目标库里找本次写入的存储后端。找不到就报错：宁可不写，
@@ -98,24 +106,39 @@ func (t *ProductionLibraryTarget) resolveStorageBackend(ctx context.Context) err
 		if !validSearchIdentifier(t.config.ObjectsBackendID) {
 			return errors.New("目标存储后端 id 形状不合法")
 		}
-		t.storageBackend = t.config.ObjectsBackendID
+		provider, err := t.storageProviderOf(ctx, t.config.ObjectsBackendID)
+		if err != nil {
+			return err
+		}
+		t.storageBackend, t.storageProvider = t.config.ObjectsBackendID, provider
 		return nil
 	}
-	var id string
+	var id, provider string
 	err := t.pool.QueryRow(ctx, `
-SELECT id FROM storage_backends
+SELECT id, coalesce(provider,'') FROM storage_backends
 WHERE tenant_id=$1 AND coalesce(config->>'bucket_name','')=$2
-ORDER BY created_at LIMIT 1`, t.scope.TenantID, t.config.ObjectsBucket).Scan(&id)
+ORDER BY created_at LIMIT 1`, t.scope.TenantID, t.config.ObjectsBucket).Scan(&id, &provider)
 	if err != nil {
 		return errors.New("目标实例没有与目标桶匹配的存储后端，拒绝写入不可解析的资源定位符")
 	}
 	if !validSearchIdentifier(id) {
 		return errors.New("目标存储后端 id 形状不合法")
 	}
-	t.storageBackend = id
+	t.storageBackend, t.storageProvider = id, provider
 	return nil
 }

+// storageProviderOf 读出目标存储后端的提供方（minio / oss / s3…）。资源行里的
+// 提供方必须与它一致，否则目标应用会按来源提供方去找一个不存在的存储服务。
+func (t *ProductionLibraryTarget) storageProviderOf(ctx context.Context, id string) (string, error) {
+	var provider string
+	if err := t.pool.QueryRow(ctx, `SELECT coalesce(provider,'') FROM storage_backends WHERE id=$1 AND tenant_id=$2`,
+		id, t.scope.TenantID).Scan(&provider); err != nil {
+		return "", errors.New("目标实例没有这个存储后端，拒绝写入不可解析的资源定位符")
+	}
+	return provider, nil
+}
+
 func NewProductionLibraryTarget(ctx context.Context, config ProductionLibraryConfig, scope ProductionScope) (*ProductionLibraryTarget, error) {
 	if err := config.validate(scope); err != nil {
 		return nil, err
@@ -259,6 +282,7 @@ func (t *ProductionLibraryTarget) Apply(ctx context.Context, stage libraryreleas
 	if err := t.resolveStorageBackend(ctx); err != nil {
 		return err
 	}
+	t.resolveEmbeddingModel(ctx)
 	archive, err := librarysnapshot.Open(t.config.ArchiveRoot, 256*1024*1024)
 	if err != nil {
 		return errors.New("归档不可用")
@@ -268,6 +292,9 @@ func (t *ProductionLibraryTarget) Apply(ctx context.Context, stage libraryreleas
 	if err != nil {
 		return errors.New("预发布快照不可读")
 	}
+	if err := t.bindSnapshotScope(manifestScope{tenantID: manifest.TenantID, knowledgeBaseID: manifest.KnowledgeBaseID}); err != nil {
+		return err
+	}
 	switch stage {
 	case libraryrelease.StageDatabase, libraryrelease.StageWiki:
 		kind := "database"
@@ -356,10 +383,10 @@ func (t *ProductionLibraryTarget) insertRows(ctx context.Context, tx pgx.Tx, arc
 		if len(row) == 0 {
 			continue
 		}
-		if table == "resources" {
-			if err := t.applyResourceBucket(row); err != nil {
-				return err
-			}
+		// 行必须先映射到目标范围再落库：原样插入会把行写进来源租户/知识库，
+		// 目标范围的删除、校验与恢复都看不见它们。
+		if err := t.retargetRow(table, row); err != nil {
+			return err
 		}
 		if columns == nil {
 			columns = make([]string, 0, len(row))
@@ -408,53 +435,9 @@ type resourceLocation struct {
 }

 // rowTransform 是"写入时对行做过的变换"的统一入口：校验要用同一份，否则摘要比的是
-// 变换前的行、目标里是变换后的行，永远不一致。
+// 变换前的行、目标里是变换后的行，永远不一致。范围映射同样必须两边一致。
 func (t *ProductionLibraryTarget) rowTransform(table string) func(map[string]json.RawMessage) error {
-	if table == "resources" {
-		return t.applyResourceBucket
-	}
-	return nil
-}
-
-// applyResourceBucket 把资源行定位符里的桶换成本次写入的目标桶。
-// 写入与校验必须用同一个变换：只改写入侧会让"行内容摘要"对不上（真跑踩过）。
-func (t *ProductionLibraryTarget) applyResourceBucket(row map[string]json.RawMessage) error {
-	if t.config.ObjectsBucket == "" {
-		return nil
-	}
-	raw, ok := row["physical_path"]
-	if !ok {
-		return errors.New("资源行缺少定位符")
-	}
-	var locator string
-	if err := json.Unmarshal(raw, &locator); err != nil {
-		return errors.New("资源定位符不可解析")
-	}
-	rewritten, ok := rewriteLocatorTarget(locator, t.storageBackend, t.config.ObjectsBucket)
-	if !ok {
-		return fmt.Errorf("资源定位符无法改写到目标桶：%s", locator)
-	}
-	encoded, err := json.Marshal(rewritten)
-	if err != nil {
-		return err
-	}
-	row["physical_path"] = encoded
-	// 原版应用先按 resources.storage_backend_id 找存储后端，再回落到 KB，再落默认服务。
-	// 这一列留着来源 id，目标仍然 500（真跑：storage backend not found → invalid path）。
-	if t.storageBackend == "" {
-		return nil
-	}
-	if raw, ok := row["storage_backend_id"]; ok {
-		var current string
-		if json.Unmarshal(raw, &current) == nil && current != "" && current != t.storageBackend {
-			encodedBackend, err := json.Marshal(t.storageBackend)
-			if err != nil {
-				return err
-			}
-			row["storage_backend_id"] = encodedBackend
-		}
-	}
-	return nil
+	return func(row map[string]json.RawMessage) error { return t.retargetRow(table, row) }
 }

 func (t *ProductionLibraryTarget) importResources(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest) error {
@@ -591,6 +574,7 @@ func (t *ProductionLibraryTarget) Verify(ctx context.Context, expect libraryrele
 	if err := t.resolveStorageBackend(ctx); err != nil {
 		return err
 	}
+	t.resolveEmbeddingModel(ctx)
 	archive, err := librarysnapshot.Open(t.config.ArchiveRoot, 256*1024*1024)
 	if err != nil {
 		return errors.New("归档不可用")
@@ -600,6 +584,9 @@ func (t *ProductionLibraryTarget) Verify(ctx context.Context, expect libraryrele
 	if err != nil {
 		return errors.New("预发布快照不可读")
 	}
+	if err := t.bindSnapshotScope(manifestScope{tenantID: manifest.TenantID, knowledgeBaseID: manifest.KnowledgeBaseID}); err != nil {
+		return err
+	}
 	for _, part := range manifest.Parts {
 		if part.Kind != "database" && part.Kind != "wiki" {
 			continue
@@ -668,6 +655,72 @@ func (t *ProductionLibraryTarget) Verify(ctx context.Context, expect libraryrele
 	if err := t.verifyObjects(ctx, archive, manifest); err != nil {
 		return err
 	}
+	// 目标里不允许留下"快照来源范围"的行：那说明有写入逃出了目标范围，
+	// 而恢复只按目标范围回放，永远删不到它们（真实事故就是这样被漏掉的）。
+	if err := t.assertNoSourceScopeRows(ctx, archive, manifest); err != nil {
+		return err
+	}
+	return nil
+}
+
+// dropOrphanResources 清掉"本次恢复后已无任何归属"的目标租户资源行。
+//
+// 判据必须同时满足：属于目标租户、不在预备份快照里、没有任何 resource_binding、
+// 也没有任何分块正文以 resource://<handle> 引用它。这样的行对目标应用完全不可见
+// （没有知识库能按它取到文件），只可能是失败写入留下的残渣。
+func (t *ProductionLibraryTarget) dropOrphanResources(ctx context.Context, backupSnapshotID string) error {
+	archive, err := librarysnapshot.Open(t.config.ArchiveRoot, 256*1024*1024)
+	if err != nil {
+		return errors.New("归档不可用")
+	}
+	defer archive.Close()
+	manifest, err := archive.Load(backupSnapshotID)
+	if err != nil {
+		return errors.New("预备份快照不可读")
+	}
+	kept := []string{}
+	for _, part := range manifest.Parts {
+		if part.Kind != "database" || part.Name != "resources" {
+			continue
+		}
+		ids, err := archivedResourceIDs(archive, part)
+		if err != nil {
+			return err
+		}
+		kept = append(kept, ids...)
+	}
+	rows, err := t.pool.Query(ctx, `
+SELECT r.id FROM resources AS r
+WHERE r.tenant_id=$1
+  AND NOT (r.id = ANY($2::text[]))
+  AND NOT EXISTS (SELECT 1 FROM resource_bindings b WHERE b.resource_id=r.id)
+  AND NOT EXISTS (SELECT 1 FROM chunks c WHERE c.tenant_id=r.tenant_id AND c.deleted_at IS NULL
+                  AND coalesce(r.handle,'') <> ''
+                  AND position('resource://'||r.handle IN coalesce(c.content,'')) > 0)`,
+		t.scope.TenantID, kept)
+	if err != nil {
+		return fmt.Errorf("查找目标残留资源失败: %w", err)
+	}
+	orphans := make([]string, 0, 8)
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			rows.Close()
+			return err
+		}
+		orphans = append(orphans, id)
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return err
+	}
+	if len(orphans) == 0 {
+		return nil
+	}
+	if _, err := t.pool.Exec(ctx, `DELETE FROM resources WHERE tenant_id=$1 AND id = ANY($2::text[])`,
+		t.scope.TenantID, orphans); err != nil {
+		return fmt.Errorf("清理目标残留资源失败: %w", err)
+	}
 	return nil
 }

@@ -772,5 +825,21 @@ func (t *ProductionLibraryTarget) Restore(ctx context.Context, backup libraryrel
 			return fmt.Errorf("恢复阶段 %s 失败: %w", stage, err)
 		}
 	}
+	// resources 是多知识库共享表，回放阶段不删它：本次写入新增、恢复后已经
+	// 没有任何归属的资源行会残留。它们对应用不可见，但属于"多写的"。
+	if err := t.dropOrphanResources(ctx, backup.SnapshotID); err != nil {
+		return err
+	}
+	// 恢复工具把四个阶段都跑完不算恢复完成：整实例必须回到预备份时刻的样子。
+	// 只比目标范围会把"写到范围外"的残留当成已恢复（真实事故：恢复后仍多 67 行）。
+	if len(backup.Universe) > 0 {
+		after, err := universeFingerprint(ctx, t.pool)
+		if err != nil {
+			return fmt.Errorf("恢复后读取整库指纹失败: %w", err)
+		}
+		if diff := universeDiff(backup.Universe, after); diff != "" {
+			return fmt.Errorf("恢复后目标与预备份时刻不一致：%s（很可能有写入逃出了预备份范围，不能当作已恢复）", diff)
+		}
+	}
 	return nil
 }
@@ -4,12 +4,15 @@ import (
 	"bytes"
 	"context"
 	"crypto/rand"
+	"crypto/sha256"
 	"encoding/hex"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"io"
 	"net/http"
 	"os"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -390,3 +393,339 @@ func sealTargetFixture(t *testing.T, archive *librarysnapshot.Archive, scope str
 	}
 	return id
 }
+
+// memoryObjectStore 是"真实字节"的资源存储替身：Put/Open/Stat/Copy 都按摘要核对，
+// 可以强制写入失败，用来复现"资源阶段失败"这一半写场景。
+type memoryObjectStore struct {
+	objects map[string][]byte
+	failPut bool
+	puts    []string
+}
+
+func newMemoryObjectStore(objects map[string][]byte) *memoryObjectStore {
+	copied := map[string][]byte{}
+	for key, value := range objects {
+		copied[key] = append([]byte(nil), value...)
+	}
+	return &memoryObjectStore{objects: copied}
+}
+
+func objectKey(provider, path string) string { return provider + "|" + path }
+
+func (s *memoryObjectStore) Put(_ context.Context, provider, path string, reader io.Reader, size int64) error {
+	if s.failPut {
+		return errors.New("对象存储不可写")
+	}
+	body, err := io.ReadAll(reader)
+	if err != nil || int64(len(body)) != size {
+		return errors.New("对象字节数与声明不符")
+	}
+	s.objects[objectKey(provider, path)] = body
+	s.puts = append(s.puts, path)
+	return nil
+}
+
+func (s *memoryObjectStore) Open(_ context.Context, provider, path string) (io.ReadCloser, error) {
+	body, ok := s.objects[objectKey(provider, path)]
+	if !ok {
+		return nil, errors.New("对象不存在")
+	}
+	return io.NopCloser(bytes.NewReader(body)), nil
+}
+
+func (s *memoryObjectStore) Stat(_ context.Context, provider, path string) (int64, string, error) {
+	body, ok := s.objects[objectKey(provider, path)]
+	if !ok {
+		return 0, "", errors.New("对象不存在")
+	}
+	digest := sha256.Sum256(body)
+	return int64(len(body)), hex.EncodeToString(digest[:]), nil
+}
+
+func (s *memoryObjectStore) Copy(_ context.Context, provider, source, target string) error {
+	body, ok := s.objects[objectKey(provider, source)]
+	if !ok {
+		return errors.New("对象不存在")
+	}
+	s.objects[objectKey(provider, target)] = append([]byte(nil), body...)
+	return nil
+}
+
+// 真实事故的形状：来源 KB 与目标 KB 不同（租户也不同）、目标非空、资源阶段写失败。
+// 旧代码把归档行原样写进目标（落在来源租户/知识库），随后的回滚只清目标范围，
+// 于是"已恢复到预备份"却留下 4 文件 / 38 分块 / 21 资源 / 4 绑定。
+func TestLibraryCrossScopeSyncRollbackLeavesNoResidueIsolated(t *testing.T) {
+	dsn, base := os.Getenv("KG_LIBRARY_TEST_DATABASE_URL"), os.Getenv("KG_LIBRARY_TEST_ELASTIC_URL")
+	if dsn == "" || base == "" || os.Getenv("KG_LIBRARY_TEST_ISOLATED") != "1" {
+		t.Skip("requires isolated PG and ES opt-in")
+	}
+	cfg, err := pgxpool.ParseConfig(dsn)
+	if err != nil || !strings.HasPrefix(cfg.ConnConfig.Database, "kg_library_test_") {
+		t.Fatal("isolated database required")
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
+	defer cancel()
+	pool, err := pgxpool.NewWithConfig(ctx, cfg)
+	if err != nil {
+		t.Fatalf("open isolated PG: %v", err)
+	}
+	defer pool.Close()
+	for _, statement := range []string{
+		`DROP SCHEMA public CASCADE; CREATE SCHEMA public`,
+		libraryTestSchema,
+		`ALTER TABLE wiki_pages ADD COLUMN status text DEFAULT 'published'`,
+		`ALTER TABLE knowledge_tag_relations ADD COLUMN tenant_id bigint`,
+		`ALTER TABLE resources ADD COLUMN provider text, ADD COLUMN size bigint, ADD COLUMN storage_backend_id text`,
+		`ALTER TABLE resources ADD PRIMARY KEY (id)`,
+		`ALTER TABLE resource_bindings ADD COLUMN relation text`,
+		`INSERT INTO knowledge_bases VALUES('kb-target',10001,'{}',NULL),('kb-source',10000,'{}',NULL)`,
+		`CREATE TABLE storage_backends(id text PRIMARY KEY, tenant_id bigint, provider text, config jsonb, created_at timestamptz)`,
+		`INSERT INTO storage_backends VALUES('backend-target',10001,'minio','{"bucket_name":"kg-test-bucket"}',now())`,
+		// 目标非空：目标自己有文档/分块，另一个租户也有数据（回滚不得碰它）。
+		`INSERT INTO knowledges VALUES('doc-target',10001,'kb-target','completed',0,'',NULL)`,
+		`INSERT INTO knowledges VALUES('doc-foreign',20000,'kb-other','completed',0,'',NULL)`,
+		`INSERT INTO chunks VALUES('chunk-target',10001,'kb-target','doc-target','target content',NULL)`,
+		`INSERT INTO chunks VALUES('chunk-foreign',20000,'kb-other','doc-foreign','foreign content',NULL)`,
+		`UPDATE resource_bindings SET tenant_id=10000 WHERE tenant_id IS NULL`,
+	} {
+		if _, err := pool.Exec(ctx, statement); err != nil {
+			t.Fatalf("fixture schema: %v", err)
+		}
+	}
+
+	token := make([]byte, 8)
+	if _, err := rand.Read(token); err != nil {
+		t.Fatal(err)
+	}
+	index := "kg_library_test_" + hex.EncodeToString(token)
+	client := &http.Client{Timeout: 20 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error {
+		return http.ErrUseLastResponse
+	}}
+	call := func(method, path, body string, want int) {
+		t.Helper()
+		request, err := http.NewRequestWithContext(ctx, method, base+path, strings.NewReader(body))
+		if err != nil {
+			t.Fatal(err)
+		}
+		request.Header.Set("Content-Type", "application/json")
+		response, err := client.Do(request)
+		if err != nil {
+			t.Fatalf("fixture HTTP: %v", err)
+		}
+		defer response.Body.Close()
+		if response.StatusCode != want {
+			t.Fatalf("fixture %s %s => %d, want %d", method, path, response.StatusCode, want)
+		}
+	}
+	call("PUT", "/"+index, `{"mappings":{"properties":{"tenant_id":{"type":"long"},"knowledge_base_id":{"type":"keyword"},"knowledge_id":{"type":"keyword"},"chunk_id":{"type":"keyword"},"content":{"type":"text"},"embedding":{"type":"dense_vector","dims":2}}}}`, 200)
+	defer func() {
+		call("DELETE", "/"+index, "", 200)
+		call("HEAD", "/"+index, "", 404)
+	}()
+	// 目标当前索引自洽：目标分块必须有对应文档，否则预备份会（正确地）拒绝。
+	call("PUT", "/"+index+"/_doc/chunk-target?refresh=true",
+		`{"tenant_id":10001,"knowledge_base_id":"kb-target","knowledge_id":"doc-target","chunk_id":"chunk-target","source_id":"chunk-target","content":"target content","embedding":[0.1,0.2]}`, 201)
+
+	archiveRoot := t.TempDir()
+	if err := os.Chmod(archiveRoot, 0700); err != nil {
+		t.Fatal(err)
+	}
+	archive, err := librarysnapshot.Open(archiveRoot, 64*1024*1024)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer archive.Close()
+	snapshotID := sealCrossScopeFixture(t, archive, 10000, "kb-source")
+
+	store := newMemoryObjectStore(nil)
+	target, err := NewProductionLibraryTarget(ctx, ProductionLibraryConfig{
+		Postgres: dsn, SearchURL: base, SearchIndex: index, Objects: store, ObjectsReader: store,
+		ArchiveRoot: archiveRoot, ObjectsBucket: "kg-test-bucket", MaxDocuments: 100, HTTPClient: client,
+	}, ProductionScope{TenantID: 10001, KnowledgeBaseID: "kb-target"})
+	if err != nil {
+		t.Fatalf("target: %v", err)
+	}
+	defer target.Close()
+	mapping, err := target.Search().MappingSHA256(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	backups, err := NewProductionBackupStore(ProductionBackupConfig{
+		Pool: target.Pool(), Search: target.Search(), Objects: store, ArchiveRoot: archiveRoot,
+		Scope: "internal", Identity: target.Identity(), MaxDocuments: 100,
+		Compatibility: librarysnapshot.Compatibility{WeKnoraVersion: "v0.7.1", SchemaVersion: 74,
+			RetrieverEngine: "elasticsearch_v8", EmbeddingModel: "text-embedding-v4",
+			EmbeddingDimension: 1024, IndexMappingSHA256: mapping},
+	})
+	if err != nil {
+		t.Fatalf("backup store: %v", err)
+	}
+	now := time.Now().UTC()
+	request := libraryrelease.Request{Target: "internal", SnapshotID: snapshotID,
+		ParamsDigest: "digest-cross-scope", Now: now,
+		Grant: libraryrelease.Grant{Target: "internal", SnapshotID: snapshotID, ParamsDigest: "digest-cross-scope",
+			ExpiresAt: now.Add(10 * time.Minute), AuthorizedBy: "acceptance"}}
+
+	// 1) 资源阶段失败 → 必须回滚且不留下来源范围的行。
+	store.failPut = true
+	outcome := libraryrelease.Run(ctx, request, target, backups, nil)
+	if outcome.State != "rolled_back" {
+		t.Fatalf("outcome = %+v", outcome)
+	}
+	if outcome.FailedAt != libraryrelease.StageResources {
+		t.Fatalf("failed_at = %s, want resources", outcome.FailedAt)
+	}
+	if outcome.Backup.SnapshotID == "" || len(outcome.Backup.Universe) == 0 {
+		t.Fatalf("backup must carry the whole-instance fingerprint: %+v", outcome.Backup)
+	}
+	for _, table := range []string{"knowledges", "chunks", "wiki_pages"} {
+		var residue int64
+		if err := pool.QueryRow(ctx, fmt.Sprintf(`SELECT count(*) FROM %s WHERE tenant_id=$1 AND knowledge_base_id=$2`, table),
+			10000, "kb-source").Scan(&residue); err != nil {
+			t.Fatal(err)
+		}
+		if residue != 0 {
+			t.Fatalf("%s still carries %d source-scope rows after rollback", table, residue)
+		}
+	}
+	// 夹具在租户 10000 里本来就有自己的行（res-safe / binding-safe）：它们必须原样
+	// 留下，只有这次同步写进去的行（res-source / 绑到 doc-source）才算残留。
+	for query, want := range map[string]int64{
+		`SELECT count(*) FROM resources WHERE id='res-source'`:                                   0,
+		`SELECT count(*) FROM resource_bindings WHERE tenant_id=10000 AND owner_id='doc-source'`: 0,
+		`SELECT count(*) FROM resources WHERE tenant_id=10000`:                                   1,
+	} {
+		var got int64
+		if err := pool.QueryRow(ctx, query).Scan(&got); err != nil {
+			t.Fatal(err)
+		}
+		if got != want {
+			t.Fatalf("%s = %d, want %d", query, got, want)
+		}
+	}
+	var leftovers int64
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM resources WHERE tenant_id=10001`).Scan(&leftovers); err != nil {
+		t.Fatal(err)
+	}
+	if leftovers != 0 {
+		t.Fatalf("aborted write left %d resource rows in the target tenant", leftovers)
+	}
+	var foreign int64
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM chunks WHERE tenant_id=20000`).Scan(&foreign); err != nil {
+		t.Fatal(err)
+	}
+	// 夹具本来就有 1 行 other-scope(20000)，加上本用例的 chunk-foreign，共 2 行。
+	if foreign != 2 {
+		t.Fatalf("cross-tenant rows must survive rollback, got %d", foreign)
+	}
+	if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: outcome.Backup.SnapshotID}); err != nil {
+		t.Fatalf("rollback must verify against the prebackup: %v", err)
+	}
+
+	// 2) 同样的跨范围同步，资源能写 → 行必须落在目标范围，资源定位符指向目标存储。
+	store.failPut = false
+	succeed := libraryrelease.Run(ctx, request, target, backups, nil)
+	if succeed.State != "succeeded" {
+		t.Fatalf("second run outcome = %+v", succeed)
+	}
+	if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: snapshotID}); err != nil {
+		t.Fatalf("verify after cross-scope sync: %v", err)
+	}
+	var knowledgeKB, chunkKB, provider, locator string
+	var tenant int64
+	if err := pool.QueryRow(ctx, `SELECT tenant_id, knowledge_base_id FROM knowledges WHERE id='doc-source'`).
+		Scan(&tenant, &knowledgeKB); err != nil {
+		t.Fatalf("source document must exist in the target scope: %v", err)
+	}
+	if tenant != 10001 || knowledgeKB != "kb-target" {
+		t.Fatalf("cross-scope row landed at %d/%s", tenant, knowledgeKB)
+	}
+	if err := pool.QueryRow(ctx, `SELECT knowledge_base_id FROM chunks WHERE id='chunk-source'`).Scan(&chunkKB); err != nil {
+		t.Fatal(err)
+	}
+	if chunkKB != "kb-target" {
+		t.Fatalf("chunk landed in %s", chunkKB)
+	}
+	if err := pool.QueryRow(ctx, `SELECT provider, physical_path FROM resources WHERE id='res-source'`).
+		Scan(&provider, &locator); err != nil {
+		t.Fatal(err)
+	}
+	if provider != "minio" {
+		t.Fatalf("resource provider = %s, want minio", provider)
+	}
+	if !strings.Contains(locator, "storage://backend-target/minio://kg-test-bucket/10001/") {
+		t.Fatalf("resource locator was not retargeted: %s", locator)
+	}
+	if strings.Contains(locator, "weknora/10000") {
+		t.Fatalf("resource locator still points at the source tenant: %s", locator)
+	}
+	t.Logf("cross_scope_sync verified: snapshot=%s backup=%s locator=%s", snapshotID, succeed.Backup.SnapshotID, locator)
+}
+
+// sealCrossScopeFixture 造一份"来源范围与目标不同"的整库快照：文档/分块/资源/绑定/
+// Wiki/索引都带来源身份，资源部件是可核对的真实字节。
+func sealCrossScopeFixture(t *testing.T, archive *librarysnapshot.Archive, sourceTenant uint64, sourceKB string) string {
+	t.Helper()
+	encode := func(value any) []byte {
+		encoded, err := json.Marshal(value)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return encoded
+	}
+	const handle = "abcdefghijklmnopqrstuv"
+	objectBytes := []byte("12345678")
+	parts := make([]librarysnapshot.Part, 0, 12)
+	put := func(kind, name string, count int64, body []byte) {
+		part, err := archive.Put(bytes.NewReader(body))
+		if err != nil {
+			t.Fatal(err)
+		}
+		part.Kind, part.Name, part.Count = kind, name, count
+		parts = append(parts, part)
+	}
+	put("database", "knowledges", 1, encode([]map[string]any{{
+		"id": "doc-source", "tenant_id": sourceTenant, "knowledge_base_id": sourceKB,
+		"parse_status": "completed", "pending_subtasks_count": 0, "file_path": ""}}))
+	put("database", "chunks", 1, encode([]map[string]any{{
+		"id": "chunk-source", "tenant_id": sourceTenant, "knowledge_base_id": sourceKB,
+		"knowledge_id": "doc-source", "content": "见 resource://" + handle}}))
+	put("database", "knowledge_tags", 0, encode([]map[string]any{}))
+	put("database", "knowledge_tag_relations", 0, encode([]map[string]any{}))
+	put("database", "resources", 1, encode([]map[string]any{{
+		"id": "res-source", "handle": handle, "tenant_id": sourceTenant,
+		"storage_backend_id": "backend-source", "provider": "oss",
+		"physical_path": "storage://backend-source/oss://source-bucket/weknora/" +
+			formatTenant(sourceTenant) + "/res-source/a.png",
+		"size": int64(len(objectBytes))}}))
+	put("database", "resource_bindings", 1, encode([]map[string]any{{
+		"id": "bind-source", "resource_id": "res-source", "tenant_id": sourceTenant,
+		"owner_type": "knowledge", "owner_id": "doc-source", "relation": "attachment"}}))
+	put("resource", "res-source", 1, objectBytes)
+	put("wiki", "wiki_folders", 0, encode([]map[string]any{}))
+	put("wiki", "wiki_pages", 1, encode([]map[string]any{{
+		"id": "wiki-source", "tenant_id": sourceTenant, "knowledge_base_id": sourceKB,
+		"slug": "index", "content": "wiki"}}))
+	for _, name := range []string{"wiki_page_issues", "wiki_log_entries"} {
+		put("wiki", name, 0, encode([]map[string]any{}))
+	}
+	put("index", "chunks", 1, encode([]map[string]any{{"id": "chunk-source", "source": map[string]any{
+		"tenant_id": sourceTenant, "knowledge_base_id": sourceKB, "knowledge_id": "doc-source",
+		"chunk_id": "chunk-source", "source_id": "chunk-source", "content": "见 resource://" + handle,
+		"embedding": []float64{0.1, 0.2}}}}))
+	manifest := librarysnapshot.Manifest{ValidationPolicy: librarysnapshot.ArtifactAcceptancePolicy,
+		Format: librarysnapshot.Format, Scope: "internal", TenantID: sourceTenant, KnowledgeBaseID: sourceKB,
+		CreatedAt: time.Now().UTC(), Parts: parts,
+		Compatibility: librarysnapshot.Compatibility{
+			WeKnoraVersion: "v0.7.1", SchemaVersion: 74, RetrieverEngine: "elasticsearch_v8",
+			EmbeddingModel: "text-embedding-v4", EmbeddingDimension: 1024,
+			IndexMappingSHA256: strings.Repeat("a", 64),
+		}}
+	id, err := archive.Seal(manifest)
+	if err != nil {
+		t.Fatalf("seal cross-scope fixture: %v", err)
+	}
+	return id
+}
+
+func formatTenant(tenant uint64) string { return strconv.FormatUint(tenant, 10) }
@@ -0,0 +1,334 @@
+package v071
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+
+	"knowledge-governance/internal/librarysnapshot"
+)
+
+// ResidueReport 是"清理范围外残留"的审计记录：清掉了什么、依据是哪份归档、
+// 备份写到哪儿、清理后目标处于什么状态。没有这份记录就不许动目标。
+type ResidueReport struct {
+	SnapshotID   string            `json:"snapshot_id"`
+	SourceTenant uint64            `json:"source_tenant"`
+	SourceKB     string            `json:"source_kb"`
+	Tables       map[string]int    `json:"tables"`
+	BackupPath   string            `json:"backup_path"`
+	BackupSHA256 string            `json:"backup_sha256"`
+	Applied      bool              `json:"applied"`
+	Cleared      int               `json:"cleared"`
+	Verified     bool              `json:"verified"`
+	Foreign      map[string]string `json:"foreign_scopes,omitempty"`
+}
+
+// ClearForeignScope 清掉"归档里确实有、却仍以来源身份躺在目标里"的行。
+//
+// 它只做这一件事，而且必须全部条件成立才允许删：
+//  1. 目标这些行的内容与指定归档行逐字段一致（先证明它们是这次写入的副本，
+//     不是目标自己的数据）；
+//  2. 先把它们完整导出成 0600 的备份文件并记下摘要；
+//  3. 删除只按"归档身份集合 + 来源租户"精确命中，不做范围广删。
+//
+// 清完还要复核：来源身份不再存在、目标内容与预备份一致、整库指纹里没有别的租户。
+func (t *ProductionLibraryTarget) ClearForeignScope(ctx context.Context, source ProductionScope, snapshotID,
+	backupDir string, apply bool) (*ResidueReport, error) {
+	if source.TenantID == 0 || !validSearchIdentifier(source.KnowledgeBaseID) {
+		return nil, errors.New("来源范围不完整，拒绝清理")
+	}
+	if source.TenantID == t.scope.TenantID && source.KnowledgeBaseID == t.scope.KnowledgeBaseID {
+		return nil, errors.New("来源与目标范围相同，不需要清理")
+	}
+	if backupDir == "" {
+		return nil, errors.New("清理前必须指定备份目录")
+	}
+	if err := t.resolveStorageBackend(ctx); err != nil {
+		return nil, err
+	}
+	archive, err := librarysnapshot.Open(t.config.ArchiveRoot, 256*1024*1024)
+	if err != nil {
+		return nil, errors.New("归档不可用")
+	}
+	defer archive.Close()
+	manifest, err := archive.Load(snapshotID)
+	if err != nil {
+		return nil, errors.New("指定快照不可读")
+	}
+	report := &ResidueReport{SnapshotID: snapshotID, SourceTenant: source.TenantID, SourceKB: source.KnowledgeBaseID,
+		Tables: map[string]int{}, Applied: false}
+	backup := map[string][]map[string]json.RawMessage{}
+	digests := map[string]string{}
+	identities := map[string][]string{}
+	for _, part := range manifest.Parts {
+		if part.Kind != "database" && part.Kind != "wiki" {
+			continue
+		}
+		archived, archivedDigests, err := archivedRowsWithDigests(archive, part)
+		if err != nil {
+			return nil, err
+		}
+		// 事故里的残留行是"归档行 + 只把桶/后端换成目标"的旧变换结果：判等时必须
+		// 同时接受原样与这一种变换，否则会把真残留当成"不是这次写的"而拒绝清理。
+		if part.Name == "resources" {
+			for _, row := range archived {
+				legacy, legacyErr := t.legacyResourceRow(row)
+				if legacyErr != nil {
+					return nil, legacyErr
+				}
+				digest, digestErr := canonicalRowDigest(legacy)
+				if digestErr != nil {
+					return nil, digestErr
+				}
+				archivedDigests[digest] = true
+			}
+		}
+		if len(archived) == 0 {
+			continue
+		}
+		live, err := t.foreignScopeRows(ctx, part.Name, source)
+		if err != nil {
+			return nil, err
+		}
+		liveDigests := make(map[string]string, len(live))
+		for _, row := range live {
+			digest, err := canonicalRowDigest(row)
+			if err != nil {
+				return nil, err
+			}
+			liveDigests[digest] = archiveRowIdentity(part.Name, row)
+		}
+		// 目标里这一范围的行必须都能在归档里找到同样内容：否则它们不是这次写入的
+		// 副本，删了就是删别人的数据。
+		rows := make([]map[string]json.RawMessage, 0, len(live))
+		ids := make([]string, 0, len(live))
+		for digest, identity := range liveDigests {
+			if _, ok := archivedDigests[digest]; !ok {
+				return nil, fmt.Errorf("目标 %s 的行 %s 与指定归档不一致，拒绝按本次写入清理", part.Name, identity)
+			}
+			ids = append(ids, identity)
+		}
+		for _, row := range live {
+			rows = append(rows, row)
+		}
+		sort.Strings(ids)
+		report.Tables[part.Name] = len(rows)
+		backup[part.Name] = rows
+		identities[part.Name] = ids
+		digests[part.Name] = combineRowDigests(sortedKeys(liveDigests))
+	}
+	total := 0
+	for _, count := range report.Tables {
+		total += count
+	}
+	if total == 0 {
+		report.Verified = true
+		return report, nil
+	}
+	if err := os.MkdirAll(backupDir, 0o700); err != nil {
+		return nil, fmt.Errorf("创建备份目录失败: %w", err)
+	}
+	payload, err := json.MarshalIndent(map[string]any{
+		"snapshot_id": snapshotID, "source_tenant": source.TenantID, "source_kb": source.KnowledgeBaseID,
+		"target_tenant": t.scope.TenantID, "target_kb": t.scope.KnowledgeBaseID,
+		"tables": report.Tables, "rows": backup, "digests": digests,
+	}, "", "  ")
+	if err != nil {
+		return nil, err
+	}
+	report.BackupPath = filepath.Join(backupDir, "foreign-scope-residue-"+snapshotID[:12]+".json")
+	if err := os.WriteFile(report.BackupPath, payload, 0o600); err != nil {
+		return nil, fmt.Errorf("写残留备份失败: %w", err)
+	}
+	sum := sha256.Sum256(payload)
+	report.BackupSHA256 = hex.EncodeToString(sum[:])
+	if !apply {
+		return report, nil
+	}
+	// 删除顺序：子表先、父表后（resource_bindings → resources，chunks → knowledges）。
+	for index := len(productionLibraryTables) - 1; index >= 0; index-- {
+		table := productionLibraryTables[index]
+		ids, ok := identities[table]
+		if !ok || len(ids) == 0 {
+			continue
+		}
+		statement := fmt.Sprintf(`DELETE FROM %s WHERE id = ANY($1::text[])`, table)
+		if !tableHasTenant(table) {
+			// knowledge_tag_relations 没有 id：身份就是父文档 + 标签。
+			statement = `DELETE FROM knowledge_tag_relations WHERE (knowledge_id||':'||tag_id) = ANY($1::text[])`
+		}
+		if _, err := t.pool.Exec(ctx, statement, ids); err != nil {
+			return nil, fmt.Errorf("清理 %s 失败: %w", table, err)
+		}
+	}
+	report.Applied = true
+	report.Cleared = total
+	return report, nil
+}
+
+// legacyResourceRow 复现旧实现写资源行时的变换：只换定位符里的存储后端与桶，
+// 租户/知识库/provider 原样保留。判等要认这一种形状，事故残留才是"可证明"的。
+func (t *ProductionLibraryTarget) legacyResourceRow(row map[string]json.RawMessage) (map[string]json.RawMessage, error) {
+	copied := make(map[string]json.RawMessage, len(row))
+	for key, value := range row {
+		copied[key] = value
+	}
+	if t.config.ObjectsBucket == "" {
+		return copied, nil
+	}
+	raw, ok := copied["physical_path"]
+	if !ok {
+		return nil, errors.New("资源行缺少定位符")
+	}
+	var locator string
+	if err := json.Unmarshal(raw, &locator); err != nil {
+		return nil, errors.New("资源定位符不可解析")
+	}
+	if rewritten, ok := rewriteLocatorTarget(locator, t.storageBackend, t.config.ObjectsBucket); ok {
+		encoded, err := json.Marshal(rewritten)
+		if err != nil {
+			return nil, err
+		}
+		copied["physical_path"] = encoded
+	}
+	if t.storageBackend == "" {
+		return copied, nil
+	}
+	if raw, ok := copied["storage_backend_id"]; ok {
+		var current string
+		if json.Unmarshal(raw, &current) == nil && current != "" && current != t.storageBackend {
+			encoded, err := json.Marshal(t.storageBackend)
+			if err != nil {
+				return nil, err
+			}
+			copied["storage_backend_id"] = encoded
+		}
+	}
+	return copied, nil
+}
+
+// VerifyResidueCleared 复核清理结果：来源身份不再存在、其它租户的解释得清、
+// 且（可选）目标内容与预备份完全一致。
+func (t *ProductionLibraryTarget) VerifyResidueCleared(ctx context.Context, source ProductionScope) error {
+	for _, table := range productionLibraryTables {
+		if !tableHasTenant(table) {
+			continue
+		}
+		statement := fmt.Sprintf(`SELECT count(*) FROM %s WHERE tenant_id=$1`, table)
+		arguments := []any{source.TenantID}
+		if tablesWithKB[table] {
+			statement += " AND knowledge_base_id=$2"
+			arguments = append(arguments, source.KnowledgeBaseID)
+		}
+		var count int64
+		if err := t.pool.QueryRow(ctx, statement, arguments...).Scan(&count); err != nil {
+			return fmt.Errorf("复核 %s 失败: %w", table, err)
+		}
+		if count > 0 {
+			return fmt.Errorf("目标 %s 仍有 %d 行属于来源范围 %d/%s", table, count, source.TenantID, source.KnowledgeBaseID)
+		}
+	}
+	return nil
+}
+
+// ForeignScopes 列出目标里所有"不是目标租户"的租户，用于人工确认没有别的泄漏。
+func (t *ProductionLibraryTarget) ForeignScopes(ctx context.Context) (map[string]string, error) {
+	foreign := map[string]string{}
+	for _, table := range productionLibraryTables {
+		if !tableHasTenant(table) {
+			continue
+		}
+		rows, err := t.pool.Query(ctx, fmt.Sprintf(
+			`SELECT tenant_id, count(*) FROM %s WHERE tenant_id <> $1 GROUP BY 1`, table), t.scope.TenantID)
+		if err != nil {
+			return nil, fmt.Errorf("读取 %s 租户分布失败: %w", table, err)
+		}
+		for rows.Next() {
+			var tenant, count int64
+			if err := rows.Scan(&tenant, &count); err != nil {
+				rows.Close()
+				return nil, err
+			}
+			foreign[fmt.Sprintf("%s:%d", table, tenant)] = fmt.Sprintf("%d 行", count)
+		}
+		rows.Close()
+		if err := rows.Err(); err != nil {
+			return nil, err
+		}
+	}
+	return foreign, nil
+}
+
+// foreignScopeRows 读出目标里属于来源范围的行（resources / resource_bindings
+// 只有租户列，按租户取；其余按租户+知识库取）。
+func (t *ProductionLibraryTarget) foreignScopeRows(ctx context.Context, table string, source ProductionScope) (
+	[]map[string]json.RawMessage, error) {
+	if !tableHasTenant(table) {
+		return nil, nil
+	}
+	statement := fmt.Sprintf(`SELECT to_jsonb(%s) FROM %s WHERE %s`, table, table, scopeWhere(table))
+	arguments := []any{source.TenantID, source.KnowledgeBaseID}
+	if table == "resources" || table == "resource_bindings" {
+		statement = fmt.Sprintf(`SELECT to_jsonb(%s) FROM %s WHERE tenant_id=$1`, table, table)
+		arguments = arguments[:1]
+	}
+	rows, err := t.pool.Query(ctx, statement, arguments...)
+	if err != nil {
+		return nil, fmt.Errorf("读取目标 %s 来源范围行失败: %w", table, err)
+	}
+	defer rows.Close()
+	result := make([]map[string]json.RawMessage, 0, 8)
+	for rows.Next() {
+		var row map[string]json.RawMessage
+		if err := rows.Scan(&row); err != nil {
+			return nil, err
+		}
+		result = append(result, row)
+	}
+	return result, rows.Err()
+}
+
+// archivedRowsWithDigests 读出一份归档部件的行与规范化内容摘要。
+func archivedRowsWithDigests(archive *librarysnapshot.Archive, part librarysnapshot.Part) (
+	[]map[string]json.RawMessage, map[string]bool, error) {
+	reader, err := archive.OpenPart(part)
+	if err != nil {
+		return nil, nil, err
+	}
+	defer reader.Close()
+	decoder := json.NewDecoder(reader)
+	token, err := decoder.Token()
+	if err != nil || token != json.Delim('[') {
+		return nil, nil, errors.New("归档部件不是行数组")
+	}
+	rows := make([]map[string]json.RawMessage, 0, 64)
+	digests := map[string]bool{}
+	for decoder.More() {
+		var row map[string]json.RawMessage
+		if err := decoder.Decode(&row); err != nil {
+			return nil, nil, err
+		}
+		digest, err := canonicalRowDigest(row)
+		if err != nil {
+			return nil, nil, err
+		}
+		digests[digest] = true
+		rows = append(rows, row)
+	}
+	return rows, digests, nil
+}
+
+func sortedKeys(values map[string]string) []string {
+	keys := make([]string, 0, len(values))
+	for key := range values {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+	return keys
+}
@@ -0,0 +1,262 @@
+package v071
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+)
+
+// 整库同步是"把来源知识库的内容搬到目标的租户/知识库"，但归档行带着来源身份
+// （tenant_id / knowledge_base_id）。原样插入会让行落在来源范围里：目标范围的
+// 删除、校验、预备份与恢复全都看不见它们（真实事故：同步失败后"已恢复到预备份"，
+// 目标却多出 4 文件 / 38 分块 / 21 资源 / 4 绑定，行所属租户还是来源租户）。
+//
+// 因此写入与校验必须共用这一个改写入口，并且对不上就失败——宁可不写，也不写
+// 一行目标范围外的数据。
+
+// tablesWithKB 是真实 WeKnora v0.7.1 里带 knowledge_base_id 的知识域表。
+// resources / resource_bindings 是多知识库共享表，只有 tenant_id；
+// knowledge_tag_relations 两列都没有，归属完全由父表 knowledges 决定。
+var tablesWithKB = map[string]bool{
+	"knowledges": true, "chunks": true, "knowledge_tags": true,
+	"wiki_folders": true, "wiki_pages": true, "wiki_page_issues": true, "wiki_log_entries": true,
+}
+
+// tableHasTenant 报告这张表是否自带租户列。
+func tableHasTenant(table string) bool { return table != "knowledge_tag_relations" }
+
+// bindSnapshotScope 记下快照的来源身份。没有来源身份就无法判断"这行要搬到哪个
+// 范围"，宁可失败也不要按行内容猜。
+func (t *ProductionLibraryTarget) bindSnapshotScope(manifest manifestScope) error {
+	if manifest.tenantID == 0 || !validSearchIdentifier(manifest.knowledgeBaseID) {
+		return errors.New("快照缺少来源范围身份（tenant_id / knowledge_base_id），无法确定目标范围映射")
+	}
+	t.snapshotTenant = manifest.tenantID
+	t.snapshotKB = manifest.knowledgeBaseID
+	return nil
+}
+
+// manifestScope 只是把快照的两个范围字段抽出来，避免调用点重复解构。
+type manifestScope struct {
+	tenantID        uint64
+	knowledgeBaseID string
+}
+
+// retargetRow 把一行归档数据改写成目标范围的数据：tenant_id / knowledge_base_id
+// 换成目标身份，资源行还要把定位符、存储后端、提供方换成目标实例自己的值。
+func (t *ProductionLibraryTarget) retargetRow(table string, row map[string]json.RawMessage) error {
+	if !tableHasTenant(table) {
+		return nil
+	}
+	if err := t.retargetTenantField(row); err != nil {
+		return fmt.Errorf("%s：%w", table, err)
+	}
+	if tablesWithKB[table] {
+		if err := t.retargetKnowledgeBaseField(row); err != nil {
+			return fmt.Errorf("%s：%w", table, err)
+		}
+	}
+	if table == "resources" {
+		return t.rewriteResourceIdentity(row)
+	}
+	if table == "knowledges" {
+		return t.retargetEmbeddingModelField(row)
+	}
+	return nil
+}
+
+// retargetEmbeddingModelField 把文档的 embedding 模型换成目标知识库绑定的那一个。
+// 来源模型 id 在目标实例里通常不存在（目标 models 表可能是空的），留着它等于让
+// 应用拿一个解析不到的模型去查——库里的向量与查询向量就不是同一套了。
+func (t *ProductionLibraryTarget) retargetEmbeddingModelField(row map[string]json.RawMessage) error {
+	if t.targetEmbeddingModel == "" {
+		return nil
+	}
+	if _, ok := row["embedding_model_id"]; !ok {
+		return nil
+	}
+	encoded, err := json.Marshal(t.targetEmbeddingModel)
+	if err != nil {
+		return errors.New("编码目标 embedding 模型失败")
+	}
+	row["embedding_model_id"] = encoded
+	return nil
+}
+
+// resolveEmbeddingModel 读目标知识库自己的 embedding 模型 id。读不到不算错误
+// （目标可能确实没绑模型），此时保持来源值不动，由契约校验去发现不一致。
+func (t *ProductionLibraryTarget) resolveEmbeddingModel(ctx context.Context) {
+	var model string
+	if err := t.pool.QueryRow(ctx,
+		`SELECT coalesce(embedding_model_id,'') FROM knowledge_bases WHERE id=$1 AND tenant_id=$2`,
+		t.scope.KnowledgeBaseID, t.scope.TenantID).Scan(&model); err == nil {
+		t.targetEmbeddingModel = model
+	}
+}
+
+func (t *ProductionLibraryTarget) retargetTenantField(row map[string]json.RawMessage) error {
+	raw, ok := row["tenant_id"]
+	if !ok {
+		return errors.New("行缺少 tenant_id，无法确定要写入哪个范围")
+	}
+	var current uint64
+	if err := json.Unmarshal(raw, &current); err != nil {
+		return errors.New("行 tenant_id 不可解析")
+	}
+	if t.snapshotTenant == 0 {
+		return errors.New("快照来源租户未知，拒绝猜测目标租户")
+	}
+	if current != t.snapshotTenant {
+		return fmt.Errorf("行 tenant_id=%d 不属于快照来源租户 %d", current, t.snapshotTenant)
+	}
+	encoded, err := json.Marshal(t.scope.TenantID)
+	if err != nil {
+		return errors.New("编码目标租户失败")
+	}
+	row["tenant_id"] = encoded
+	return nil
+}
+
+func (t *ProductionLibraryTarget) retargetKnowledgeBaseField(row map[string]json.RawMessage) error {
+	raw, ok := row["knowledge_base_id"]
+	if !ok {
+		return errors.New("行缺少 knowledge_base_id，无法确定要写入哪个知识库")
+	}
+	var current string
+	if err := json.Unmarshal(raw, &current); err != nil || current == "" {
+		return errors.New("行 knowledge_base_id 不可解析")
+	}
+	if t.snapshotKB == "" {
+		return errors.New("快照来源知识库未知，拒绝猜测目标知识库")
+	}
+	if current != t.snapshotKB {
+		return fmt.Errorf("行 knowledge_base_id=%s 不属于快照来源知识库 %s", current, t.snapshotKB)
+	}
+	encoded, err := json.Marshal(t.scope.KnowledgeBaseID)
+	if err != nil {
+		return errors.New("编码目标知识库失败")
+	}
+	row["knowledge_base_id"] = encoded
+	return nil
+}
+
+// rewriteResourceIdentity 把资源行的定位符与提供方换成目标实例的值。
+//
+// 只换桶不够：来源行的 scheme（oss/minio…）与对象键里的租户段也是来源实例的，
+// 目标实例按它取文件会落到别的提供方或别的租户前缀上（真跑：storage backend
+// not found → invalid path → 500 code1007）。
+func (t *ProductionLibraryTarget) rewriteResourceIdentity(row map[string]json.RawMessage) error {
+	raw, ok := row["physical_path"]
+	if !ok {
+		return errors.New("资源行缺少定位符")
+	}
+	var locator string
+	if err := json.Unmarshal(raw, &locator); err != nil {
+		return errors.New("资源定位符不可解析")
+	}
+	rewritten, err := retargetLocator(locator, t.storageBackend, t.storageProvider, t.config.ObjectsBucket,
+		t.snapshotTenant, t.scope.TenantID)
+	if err != nil {
+		return err
+	}
+	if rewritten != locator {
+		encoded, err := json.Marshal(rewritten)
+		if err != nil {
+			return errors.New("编码目标定位符失败")
+		}
+		row["physical_path"] = encoded
+	}
+	if t.storageBackend != "" {
+		if _, ok := row["storage_backend_id"]; ok {
+			encoded, err := json.Marshal(t.storageBackend)
+			if err != nil {
+				return errors.New("编码目标存储后端失败")
+			}
+			row["storage_backend_id"] = encoded
+		}
+	}
+	if t.storageProvider != "" {
+		if _, ok := row["provider"]; ok {
+			encoded, err := json.Marshal(t.storageProvider)
+			if err != nil {
+				return errors.New("编码目标提供方失败")
+			}
+			row["provider"] = encoded
+		}
+	}
+	return nil
+}
+
+// retargetLocator 把 "storage://<backend>/<scheme>://<bucket>/<key>" 整体换成目标
+// 实例的真实取值：后端 id、scheme/提供方、桶名，以及对象键里的租户段。
+// 桶名为空表示目标没有配置对象存储（测试夹具），此时不动定位符。
+func retargetLocator(locator, backendID, provider, bucket string, sourceTenant, targetTenant uint64) (string, error) {
+	if bucket == "" || locator == "" {
+		return locator, nil
+	}
+	if locator != strings.TrimSpace(locator) {
+		return "", errors.New("资源定位符带首尾空白")
+	}
+	prefix := ""
+	rest := locator
+	if strings.HasPrefix(locator, "storage://") {
+		trimmed := strings.TrimPrefix(locator, "storage://")
+		parts := strings.SplitN(trimmed, "/", 2)
+		if len(parts) != 2 || !validSearchIdentifier(parts[0]) || parts[1] == "" {
+			return "", errors.New("资源定位符结构不可解析")
+		}
+		if backendID != "" {
+			if !validSearchIdentifier(backendID) {
+				return "", errors.New("目标存储后端 id 形状不合法")
+			}
+			parts[0] = backendID
+		}
+		prefix, rest = "storage://"+parts[0]+"/", parts[1]
+	}
+	separator := strings.Index(rest, "://")
+	if separator < 0 {
+		// 裸对象键没有桶与租户段：落盘按配置桶，改写无意义。
+		return locator, nil
+	}
+	scheme := rest[:separator]
+	if !validSearchIdentifier(scheme) {
+		return "", errors.New("资源定位符的提供方不可解析")
+	}
+	schemeless := rest[separator+3:]
+	boundary := strings.Index(schemeless, "/")
+	if boundary <= 0 || boundary == len(schemeless)-1 {
+		return "", errors.New("资源定位符缺少对象键")
+	}
+	key := schemeless[boundary+1:]
+	if provider != "" {
+		if !validSearchIdentifier(provider) {
+			return "", errors.New("目标提供方形状不合法")
+		}
+		scheme = provider
+	}
+	return prefix + scheme + "://" + bucket + "/" + retargetKeyTenant(key, sourceTenant, targetTenant), nil
+}
+
+// retargetKeyTenant 把对象键里的来源租户段换成目标租户，并丢掉来源实例自己的
+// 路径前缀（例如 MINIO_PATH_PREFIX=weknora/），让键与目标实例的写法一致。
+// 找不到来源租户段时原样返回：此时键本身不含跨租户引用。
+func retargetKeyTenant(key string, sourceTenant, targetTenant uint64) string {
+	if sourceTenant == 0 || targetTenant == 0 || sourceTenant == targetTenant {
+		return key
+	}
+	source, target := strconv.FormatUint(sourceTenant, 10), strconv.FormatUint(targetTenant, 10)
+	segments := strings.Split(key, "/")
+	for index := 0; index < len(segments) && index < 2; index++ {
+		// 只有"后面还有内容"的段才是路径前缀；整键就是租户号时不该动它。
+		if index >= len(segments)-1 || segments[index] != source {
+			continue
+		}
+		kept := segments[index:]
+		kept[0] = target
+		return strings.Join(kept, "/")
+	}
+	return key
+}
@@ -0,0 +1,194 @@
+package v071
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+)
+
+// 真实事故的形状：来源 KB 与目标 KB 不同（租户也不同），归档行原样写进目标，
+// 目标范围的删除/校验/恢复都看不到它们。这里把"必须改写哪些列"钉死。
+
+func scopeTargetFixture() *ProductionLibraryTarget {
+	return &ProductionLibraryTarget{
+		scope:           ProductionScope{TenantID: 10001, KnowledgeBaseID: "kb-target"},
+		config:          ProductionLibraryConfig{ObjectsBucket: "kgseal-1029a"},
+		storageBackend:  "backend-target",
+		storageProvider: "minio",
+		snapshotTenant:  10000,
+		snapshotKB:      "kb-source",
+	}
+}
+
+func rawRow(t *testing.T, fields map[string]any) map[string]json.RawMessage {
+	t.Helper()
+	row := make(map[string]json.RawMessage, len(fields))
+	for key, value := range fields {
+		encoded, err := json.Marshal(value)
+		if err != nil {
+			t.Fatalf("encode %s: %v", key, err)
+		}
+		row[key] = encoded
+	}
+	return row
+}
+
+func fieldText(t *testing.T, row map[string]json.RawMessage, key string) string {
+	t.Helper()
+	var value string
+	if err := json.Unmarshal(row[key], &value); err != nil {
+		t.Fatalf("%s not a string: %v", key, err)
+	}
+	return value
+}
+
+func fieldUint(t *testing.T, row map[string]json.RawMessage, key string) uint64 {
+	t.Helper()
+	var value uint64
+	if err := json.Unmarshal(row[key], &value); err != nil {
+		t.Fatalf("%s not a number: %v", key, err)
+	}
+	return value
+}
+
+func TestRetargetRowMapsSourceScopeToTarget(t *testing.T) {
+	target := scopeTargetFixture()
+	for _, table := range []string{"knowledges", "chunks", "knowledge_tags", "wiki_pages", "wiki_folders"} {
+		row := rawRow(t, map[string]any{"id": "row-1", "tenant_id": 10000, "knowledge_base_id": "kb-source"})
+		if err := target.retargetRow(table, row); err != nil {
+			t.Fatalf("%s retarget: %v", table, err)
+		}
+		if got := fieldUint(t, row, "tenant_id"); got != 10001 {
+			t.Fatalf("%s tenant_id = %d, want 10001", table, got)
+		}
+		if got := fieldText(t, row, "knowledge_base_id"); got != "kb-target" {
+			t.Fatalf("%s knowledge_base_id = %s, want kb-target", table, got)
+		}
+	}
+	// 共享表只有租户列：不得凭空写一个 knowledge_base_id 出来。
+	binding := rawRow(t, map[string]any{"id": "bind-1", "tenant_id": 10000, "owner_type": "knowledge", "owner_id": "row-1"})
+	if err := target.retargetRow("resource_bindings", binding); err != nil {
+		t.Fatalf("resource_bindings retarget: %v", err)
+	}
+	if got := fieldUint(t, binding, "tenant_id"); got != 10001 {
+		t.Fatalf("binding tenant_id = %d, want 10001", got)
+	}
+	if _, ok := binding["knowledge_base_id"]; ok {
+		t.Fatal("resource_bindings must not gain a knowledge_base_id column")
+	}
+	// knowledge_tag_relations 没有范围列：归属走父表，改写必须是空操作。
+	relation := rawRow(t, map[string]any{"knowledge_id": "row-1", "tag_id": "tag-1"})
+	if err := target.retargetRow("knowledge_tag_relations", relation); err != nil {
+		t.Fatalf("knowledge_tag_relations retarget: %v", err)
+	}
+	if len(relation) != 2 {
+		t.Fatalf("knowledge_tag_relations must stay untouched, got %v", relation)
+	}
+}
+
+// 文档行的 embedding 模型必须换成目标知识库绑定的那一个：来源模型 id 在目标
+// 实例里通常不存在，留着它检索时的查询向量与库里向量就不是同一套。
+func TestRetargetRowUsesTargetEmbeddingModel(t *testing.T) {
+	target := scopeTargetFixture()
+	target.targetEmbeddingModel = "model-target"
+	row := rawRow(t, map[string]any{"id": "doc-1", "tenant_id": 10000, "knowledge_base_id": "kb-source",
+		"embedding_model_id": "model-source"})
+	if err := target.retargetRow("knowledges", row); err != nil {
+		t.Fatalf("retarget: %v", err)
+	}
+	if got := fieldText(t, row, "embedding_model_id"); got != "model-target" {
+		t.Fatalf("embedding_model_id = %s, want model-target", got)
+	}
+	// 目标没绑模型时不猜：保持来源值，交给契约校验去发现不一致。
+	plain := scopeTargetFixture()
+	row = rawRow(t, map[string]any{"id": "doc-1", "tenant_id": 10000, "knowledge_base_id": "kb-source",
+		"embedding_model_id": "model-source"})
+	if err := plain.retargetRow("knowledges", row); err != nil {
+		t.Fatalf("retarget without target model: %v", err)
+	}
+	if got := fieldText(t, row, "embedding_model_id"); got != "model-source" {
+		t.Fatalf("embedding_model_id = %s, want unchanged", got)
+	}
+}
+
+func TestRetargetRowFailsClosedOutsideSnapshotScope(t *testing.T) {
+	target := scopeTargetFixture()
+	cases := map[string]map[string]any{
+		"foreign tenant":       {"id": "x", "tenant_id": 20000, "knowledge_base_id": "kb-source"},
+		"foreign knowledge":    {"id": "x", "tenant_id": 10000, "knowledge_base_id": "kb-other"},
+		"missing tenant":       {"id": "x", "knowledge_base_id": "kb-source"},
+		"missing knowledge kb": {"id": "x", "tenant_id": 10000},
+	}
+	for name, fields := range cases {
+		if err := target.retargetRow("knowledges", rawRow(t, fields)); err == nil {
+			t.Fatalf("%s: retarget must fail closed", name)
+		}
+	}
+	// 快照来源未知时不许猜目标范围。
+	unknown := &ProductionLibraryTarget{scope: ProductionScope{TenantID: 10001, KnowledgeBaseID: "kb-target"}}
+	if err := unknown.retargetRow("knowledges", rawRow(t, map[string]any{"id": "x", "tenant_id": 10000, "knowledge_base_id": "kb"})); err == nil {
+		t.Fatal("unknown snapshot scope must fail closed")
+	}
+}
+
+func TestRetargetRowRewritesResourceStorageIdentity(t *testing.T) {
+	target := scopeTargetFixture()
+	row := rawRow(t, map[string]any{
+		"id":                 "res-1",
+		"tenant_id":          10000,
+		"storage_backend_id": "backend-source",
+		"provider":           "oss",
+		"physical_path":      "storage://backend-source/oss://yzh-weknora-test/weknora/10000/res-1/a.png",
+	})
+	if err := target.retargetRow("resources", row); err != nil {
+		t.Fatalf("retarget resource: %v", err)
+	}
+	if got := fieldUint(t, row, "tenant_id"); got != 10001 {
+		t.Fatalf("resource tenant = %d, want 10001", got)
+	}
+	want := "storage://backend-target/minio://kgseal-1029a/10001/res-1/a.png"
+	if got := fieldText(t, row, "physical_path"); got != want {
+		t.Fatalf("physical_path = %q, want %q", got, want)
+	}
+	if got := fieldText(t, row, "provider"); got != "minio" {
+		t.Fatalf("provider = %q, want minio", got)
+	}
+	if got := fieldText(t, row, "storage_backend_id"); got != "backend-target" {
+		t.Fatalf("storage_backend_id = %q, want backend-target", got)
+	}
+}
+
+func TestRetargetKeyTenantDropsSourcePrefix(t *testing.T) {
+	cases := []struct{ key, want string }{
+		{"weknora/10000/exports/a.png", "10001/exports/a.png"},
+		{"10000/knowledge/file.md", "10001/knowledge/file.md"},
+		{"other/prefix/a.png", "other/prefix/a.png"},
+		{"10000", "10000"},
+	}
+	for _, tc := range cases {
+		if got := retargetKeyTenant(tc.key, 10000, 10001); got != tc.want {
+			t.Fatalf("retargetKeyTenant(%q) = %q, want %q", tc.key, got, tc.want)
+		}
+	}
+	// 同租户不编码：键必须原样返回。
+	if got := retargetKeyTenant("weknora/10000/a.png", 10000, 10000); got != "weknora/10000/a.png" {
+		t.Fatalf("same tenant must keep key, got %q", got)
+	}
+}
+
+func TestRetargetLocatorRejectsMalformed(t *testing.T) {
+	if _, err := retargetLocator(" not-a-locator ", "backend", "minio", "bucket", 10000, 10001); err == nil {
+		t.Fatal("locator with surrounding spaces must be rejected")
+	}
+	if _, err := retargetLocator("storage://backend/oss://", "backend", "minio", "bucket", 10000, 10001); err == nil {
+		t.Fatal("locator without object key must be rejected")
+	}
+	// 没有目标桶（测试夹具）时不改写，保持原值可继续用于采集/校验。
+	locator := "storage://backend/oss://bucket/key"
+	if got, err := retargetLocator(locator, "backend", "minio", "", 10000, 10001); err != nil || got != locator {
+		t.Fatalf("empty bucket must be a no-op, got %q err=%v", got, err)
+	}
+	if !strings.Contains(retargetKeyTenant("weknora/10000/a", 10000, 10001), "10001") {
+		t.Fatal("tenant segment must be replaced")
+	}
+}
@@ -0,0 +1,126 @@
+package v071
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sort"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"knowledge-governance/internal/libraryrelease"
+	"knowledge-governance/internal/librarysnapshot"
+)
+
+// universeFingerprint 是"整实例"的知识域指纹：不限租户/知识库的行数 + 身份集合摘要。
+//
+// 为什么不能只比目标范围：恢复只按目标范围删除/回放，写到范围外的残留它看不见
+// （真实事故：作业报"已恢复到预备份"，目标却多出 4 文件 / 38 分块 / 21 资源 /
+// 4 绑定）。整实例指纹让"多写了什么"必然暴露，同时也能证明恢复没有误删别的
+// 知识库的行。
+func universeFingerprint(ctx context.Context, pool *pgxpool.Pool) (map[string]libraryrelease.TableFingerprint, error) {
+	result := make(map[string]libraryrelease.TableFingerprint, len(productionLibraryTables))
+	for _, table := range productionLibraryTables {
+		identity := identityExpr(table)
+		statement := fmt.Sprintf(
+			`SELECT count(*), coalesce(md5(string_agg((%s)::text, E'\n' ORDER BY (%s)::text)), '') FROM %s`,
+			identity, identity, table)
+		var fingerprint libraryrelease.TableFingerprint
+		if err := pool.QueryRow(ctx, statement).Scan(&fingerprint.Count, &fingerprint.Digest); err != nil {
+			return nil, fmt.Errorf("读取整库 %s 指纹失败: %w", table, err)
+		}
+		result[table] = fingerprint
+	}
+	return result, nil
+}
+
+// archivedRowIdentities 读出一份归档部件的行身份（与校验用的身份口径一致）。
+func archivedRowIdentities(archive *librarysnapshot.Archive, part librarysnapshot.Part) ([]string, error) {
+	reader, err := archive.OpenPart(part)
+	if err != nil {
+		return nil, err
+	}
+	defer reader.Close()
+	decoder := json.NewDecoder(reader)
+	token, err := decoder.Token()
+	if err != nil || token != json.Delim('[') {
+		return nil, errors.New("归档部件不是行数组")
+	}
+	ids := make([]string, 0, 64)
+	for decoder.More() {
+		var row map[string]json.RawMessage
+		if err := decoder.Decode(&row); err != nil {
+			return nil, err
+		}
+		identity := archiveRowIdentity(part.Name, row)
+		if identity == "" {
+			continue
+		}
+		ids = append(ids, identity)
+	}
+	return ids, nil
+}
+
+// universeDiff 返回第一处不一致的表；空字符串表示整实例与参照完全一致。
+func universeDiff(want, got map[string]libraryrelease.TableFingerprint) string {
+	tables := make([]string, 0, len(want))
+	for table := range want {
+		tables = append(tables, table)
+	}
+	sort.Strings(tables)
+	for _, table := range tables {
+		current, ok := got[table]
+		if !ok {
+			return fmt.Sprintf("%s（目标读不到）", table)
+		}
+		if current.Count != want[table].Count {
+			return fmt.Sprintf("%s 行数 %d，参照 %d", table, current.Count, want[table].Count)
+		}
+		if current.Digest != want[table].Digest {
+			return fmt.Sprintf("%s 身份集合与参照不同（同数量改写也会被这条拦住）", table)
+		}
+	}
+	return ""
+}
+
+// assertNoSourceScopeRows 找出"这次要写的这些行，仍然以来源身份躺在目标里"。
+//
+// 判据按归档里的身份集合逐表核对，而不是"这个租户/知识库有没有行"：同一个租户
+// 里本来就可能住着别的知识库（共享资源、别的 KB 的绑定），按范围数会把它们误
+// 报成残留。这里只认"本次写集合里、仍属于来源租户"的行——它们只可能来自逃出
+// 目标范围的写入，而恢复只按目标范围回放，永远删不到它们。
+func (t *ProductionLibraryTarget) assertNoSourceScopeRows(ctx context.Context, archive *librarysnapshot.Archive,
+	manifest librarysnapshot.Manifest) error {
+	if t.snapshotTenant == 0 {
+		return nil
+	}
+	if t.snapshotTenant == t.scope.TenantID && t.snapshotKB == t.scope.KnowledgeBaseID {
+		return nil
+	}
+	for _, part := range manifest.Parts {
+		if part.Kind != "database" && part.Kind != "wiki" {
+			continue
+		}
+		if !tableHasTenant(part.Name) {
+			continue
+		}
+		ids, err := archivedRowIdentities(archive, part)
+		if err != nil {
+			return err
+		}
+		if len(ids) == 0 {
+			continue
+		}
+		statement := fmt.Sprintf(`SELECT count(*) FROM %s WHERE tenant_id=$1 AND id = ANY($2::text[])`, part.Name)
+		var count int64
+		if err := t.pool.QueryRow(ctx, statement, t.snapshotTenant, ids).Scan(&count); err != nil {
+			return fmt.Errorf("检查目标 %s 的来源范围残留失败: %w", part.Name, err)
+		}
+		if count > 0 {
+			return fmt.Errorf("目标 %s 残留 %d 行仍属于来源范围 %d/%s：有写入逃出了目标范围，恢复不会清掉它们",
+				part.Name, count, t.snapshotTenant, t.snapshotKB)
+		}
+	}
+	return nil
+}
@@ -60,8 +60,11 @@ func TestRewriteLocatorSwapsStorageBackendWithBucket(t *testing.T) {
 // 行变换用解析到的目标后端与目标桶：写入与校验共用这一份，摘要才不会两边打架。
 func TestResourceRowTransformRewritesBackendAndBucket(t *testing.T) {
 	target := &ProductionLibraryTarget{config: ProductionLibraryConfig{ObjectsBucket: "kgseal-1029a"},
-		storageBackend: "target-backend"}
-	row := map[string]json.RawMessage{"physical_path": json.RawMessage(`"storage://source-backend/minio://kgmaint-1028a/10001/k/ab.md"`)}
+		storageBackend: "target-backend", snapshotTenant: 10001, snapshotKB: "kb",
+		scope: ProductionScope{TenantID: 10001, KnowledgeBaseID: "kb"}}
+	row := map[string]json.RawMessage{
+		"tenant_id":     json.RawMessage(`10001`),
+		"physical_path": json.RawMessage(`"storage://source-backend/minio://kgmaint-1028a/10001/k/ab.md"`)}
 	if err := target.rowTransform("resources")(row); err != nil {
 		t.Fatalf("row transform failed: %v", err)
 	}
@@ -78,8 +81,10 @@ func TestResourceRowTransformRewritesBackendAndBucket(t *testing.T) {
 // 只改定位符不改这一列，目标仍是 storage backend not found（真跑 500）。
 func TestResourceRowTransformRewritesBackendColumn(t *testing.T) {
 	target := &ProductionLibraryTarget{config: ProductionLibraryConfig{ObjectsBucket: "kgseal-1029a"},
-		storageBackend: "target-backend"}
+		storageBackend: "target-backend", snapshotTenant: 10001, snapshotKB: "kb",
+		scope: ProductionScope{TenantID: 10001, KnowledgeBaseID: "kb"}}
 	row := map[string]json.RawMessage{
+		"tenant_id":          json.RawMessage(`10001`),
 		"physical_path":      json.RawMessage(`"storage://source-backend/minio://kgmaint-1028a/1/k/ab.md"`),
 		"storage_backend_id": json.RawMessage(`"source-backend"`),
 	}