Merge main into feature/demo-showcase

2026-09-05 15:37:03 +08:00 · 2026-08-11 15:46:58 +08:00
parent 3a181c7416 0aa3cc92df
commit 2e20c511b9
218 changed files with 14685 additions and 6184 deletions
@@ -2,15 +2,22 @@ package service

 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"math"
+	"path"
 	"strings"
 	"time"

 	"backupx/server/internal/apperror"
+	"backupx/server/internal/backup"
 	"backupx/server/internal/model"
 	"backupx/server/internal/repository"
+	"backupx/server/internal/storage"
 	"backupx/server/internal/storage/codec"
 )

@@ -23,6 +30,7 @@ type AgentService struct {
 	storageRepo repository.StorageTargetRepository
 	cmdRepo     repository.AgentCommandRepository
 	restoreRepo repository.RestoreRecordRepository
+	registry    *storage.Registry
 	cipher      *codec.ConfigCipher
 }

@@ -33,6 +41,7 @@ func NewAgentService(
 	storageRepo repository.StorageTargetRepository,
 	cmdRepo repository.AgentCommandRepository,
 	cipher *codec.ConfigCipher,
+	registry *storage.Registry,
 ) *AgentService {
 	return &AgentService{
 		nodeRepo:    nodeRepo,
@@ -40,6 +49,7 @@ func NewAgentService(
 		recordRepo:  recordRepo,
 		storageRepo: storageRepo,
 		cmdRepo:     cmdRepo,
+		registry:    registry,
 		cipher:      cipher,
 	}
 }
@@ -145,10 +155,11 @@ type AgentTaskSpec struct {

 // AgentStorageTargetConfig 存储目标配置（已解密）
 type AgentStorageTargetConfig struct {
-	ID     uint            `json:"id"`
-	Type   string          `json:"type"`
-	Name   string          `json:"name"`
-	Config json.RawMessage `json:"config"`
+	ID           uint            `json:"id"`
+	Type         string          `json:"type"`
+	Name         string          `json:"name"`
+	Config       json.RawMessage `json:"config"`
+	TransferMode string          `json:"transferMode"`
 }

 // GetTaskSpec 返回 Agent 执行任务所需的完整规格。
@@ -187,11 +198,22 @@ func (s *AgentService) GetTaskSpec(ctx context.Context, node *model.Node, taskID
 		if err != nil {
 			return nil, fmt.Errorf("decrypt storage config: %w", err)
 		}
+		transferMode := storage.TransferModeDirect
+		if strings.EqualFold(target.Type, storage.TypeLocalDisk) {
+			var localConfig storage.LocalDiskConfig
+			if err := json.Unmarshal(configRaw, &localConfig); err != nil {
+				return nil, fmt.Errorf("decode local disk config: %w", err)
+			}
+			if localConfig.MasterRelay {
+				transferMode = storage.TransferModeMasterRelay
+			}
+		}
 		storageTargets = append(storageTargets, AgentStorageTargetConfig{
-			ID:     target.ID,
-			Type:   target.Type,
-			Name:   target.Name,
-			Config: json.RawMessage(configRaw),
+			ID:           target.ID,
+			Type:         target.Type,
+			Name:         target.Name,
+			Config:       json.RawMessage(configRaw),
+			TransferMode: transferMode,
 		})
 	}
 	return &AgentTaskSpec{
@@ -214,6 +236,102 @@ func (s *AgentService) GetTaskSpec(ctx context.Context, node *model.Node, taskID
 	}, nil
 }

+// UploadArtifact receives a remote Agent artifact as a stream and writes it
+// with a provider created on the Master. The first supported use is local_disk,
+// whose configured path belongs to the Master rather than the source Agent.
+func (s *AgentService) UploadArtifact(ctx context.Context, node *model.Node, recordID, targetID uint, objectKey string, size int64, checksum string, reader io.Reader) error {
+	if node == nil || reader == nil || s.registry == nil {
+		return apperror.BadRequest("AGENT_ARTIFACT_INVALID", "中转上传参数不完整", nil)
+	}
+	record, err := s.recordRepo.FindByID(ctx, recordID)
+	if err != nil {
+		return err
+	}
+	if record == nil {
+		return apperror.New(404, "BACKUP_RECORD_NOT_FOUND", "记录不存在", nil)
+	}
+	task, err := s.taskRepo.FindByID(ctx, record.TaskID)
+	if err != nil {
+		return err
+	}
+	if task == nil || !recordBelongsToNode(record, task, node.ID) {
+		return apperror.Unauthorized("BACKUP_RECORD_FORBIDDEN", "记录不属于当前节点", nil)
+	}
+	if isBackupRecordTerminal(record.Status) {
+		return apperror.BadRequest("BACKUP_RECORD_TERMINAL", "备份记录已结束，不能继续上传产物", nil)
+	}
+	allowedTarget := false
+	for _, configuredTargetID := range collectTargetIDs(task) {
+		if configuredTargetID == targetID {
+			allowedTarget = true
+			break
+		}
+	}
+	if !allowedTarget {
+		return apperror.Unauthorized("BACKUP_STORAGE_TARGET_FORBIDDEN", "存储目标不属于该任务", nil)
+	}
+	target, err := s.storageRepo.FindByID(ctx, targetID)
+	if err != nil {
+		return err
+	}
+	if target == nil || !strings.EqualFold(target.Type, storage.TypeLocalDisk) {
+		return apperror.BadRequest("AGENT_ARTIFACT_RELAY_UNSUPPORTED", "仅 Master 本地磁盘目标需要中转上传", nil)
+	}
+	configMap := map[string]any{}
+	if err := s.cipher.DecryptJSON(target.ConfigCiphertext, &configMap); err != nil {
+		return fmt.Errorf("decrypt storage config: %w", err)
+	}
+	masterRelay, _ := configMap["masterRelay"].(bool)
+	if !masterRelay {
+		return apperror.BadRequest("AGENT_ARTIFACT_RELAY_UNSUPPORTED", "该本地磁盘目标配置为 Agent 直接写入", nil)
+	}
+	cleanKey, keyErr := s.validateArtifactKey(record, task, objectKey, true)
+	if keyErr != nil {
+		return keyErr
+	}
+	checksum = strings.TrimSpace(checksum)
+	checksumBytes, checksumErr := hex.DecodeString(checksum)
+	if size < 0 || size == math.MaxInt64 || checksumErr != nil || len(checksumBytes) != sha256.Size {
+		return apperror.BadRequest("AGENT_ARTIFACT_INVALID", "中转上传需要有效的大小和 SHA-256", checksumErr)
+	}
+	if target.QuotaBytes > 0 {
+		usage, usageErr := s.recordRepo.StorageUsage(ctx)
+		if usageErr != nil {
+			return fmt.Errorf("read storage usage: %w", usageErr)
+		}
+		currentUsed := int64(0)
+		for _, item := range usage {
+			if item.StorageTargetID == targetID {
+				currentUsed = item.TotalSize
+				break
+			}
+		}
+		if currentUsed >= target.QuotaBytes || size > target.QuotaBytes-currentUsed {
+			return apperror.BadRequest("BACKUP_STORAGE_QUOTA_EXCEEDED", fmt.Sprintf("超出存储目标配额（当前 %d，新增 %d，配额 %d）", currentUsed, size, target.QuotaBytes), nil)
+		}
+	}
+	provider, err := s.registry.Create(ctx, target.Type, configMap)
+	if err != nil {
+		return fmt.Errorf("create master relay provider: %w", err)
+	}
+	limited := io.LimitReader(reader, size+1)
+	hashed := newHashingReader(limited)
+	metadata := map[string]string{
+		"taskId":       fmt.Sprintf("%d", task.ID),
+		"recordId":     fmt.Sprintf("%d", record.ID),
+		"sourceNodeId": fmt.Sprintf("%d", node.ID),
+		"transferMode": storage.TransferModeMasterRelay,
+	}
+	if err := provider.Upload(ctx, cleanKey, hashed, size, metadata); err != nil {
+		return errors.Join(fmt.Errorf("relay artifact to master storage: %w", err), provider.Delete(ctx, cleanKey))
+	}
+	if hashed.n != size || !strings.EqualFold(hashed.Sum(), checksum) {
+		deleteErr := provider.Delete(ctx, cleanKey)
+		return errors.Join(fmt.Errorf("relayed artifact integrity mismatch: received %d of %d bytes", hashed.n, size), deleteErr)
+	}
+	return nil
+}
+
 func (s *AgentService) ensureTaskSpecAccess(ctx context.Context, node *model.Node, task *model.BackupTask) error {
 	if task.NodeID == node.ID {
 		return nil
@@ -236,6 +354,7 @@ type AgentRecordUpdate struct {
 	Checksum             string                    `json:"checksum,omitempty"`
 	StoragePath          string                    `json:"storagePath,omitempty"`
 	StorageTargetID      uint                      `json:"storageTargetId,omitempty"`
+	StorageTransferMode  string                    `json:"storageTransferMode,omitempty"`
 	StorageUploadResults []StorageUploadResultItem `json:"storageUploadResults,omitempty"`
 	ErrorMessage         string                    `json:"errorMessage,omitempty"`
 	LogAppend            string                    `json:"logAppend,omitempty"` // 增量日志，追加到 record.log_content
@@ -260,6 +379,99 @@ func (s *AgentService) UpdateRecord(ctx context.Context, node *model.Node, recor
 	if isBackupRecordTerminal(record.Status) {
 		return nil
 	}
+	allowedTargets := make(map[uint]struct{})
+	for _, targetID := range collectTargetIDs(task) {
+		allowedTargets[targetID] = struct{}{}
+	}
+	targetCache := make(map[uint]*model.StorageTarget)
+	validateTransferMode := func(targetID uint, transferMode string) (string, error) {
+		if _, ok := allowedTargets[targetID]; !ok {
+			return "", apperror.Unauthorized("BACKUP_STORAGE_TARGET_FORBIDDEN", "存储目标不属于该任务", nil)
+		}
+		target := targetCache[targetID]
+		if target == nil {
+			var findErr error
+			target, findErr = s.storageRepo.FindByID(ctx, targetID)
+			if findErr != nil {
+				return "", findErr
+			}
+			if target == nil {
+				return "", apperror.BadRequest("BACKUP_STORAGE_TARGET_INVALID", "存储目标不存在", nil)
+			}
+			targetCache[targetID] = target
+		}
+		expectedMode := storage.TransferModeDirect
+		if strings.EqualFold(target.Type, storage.TypeLocalDisk) {
+			var localConfig storage.LocalDiskConfig
+			if err := s.cipher.DecryptJSON(target.ConfigCiphertext, &localConfig); err != nil {
+				return "", fmt.Errorf("decrypt storage config: %w", err)
+			}
+			if localConfig.MasterRelay {
+				expectedMode = storage.TransferModeMasterRelay
+			}
+		}
+		if transferMode != "" && transferMode != expectedMode {
+			return "", apperror.BadRequest("AGENT_STORAGE_TRANSFER_MODE_INVALID", "Agent 上报的存储传输模式与目标配置不一致", nil)
+		}
+		return expectedMode, nil
+	}
+	selectedTransferMode := ""
+	if update.StorageTargetID > 0 {
+		if _, ok := allowedTargets[update.StorageTargetID]; !ok {
+			return apperror.Unauthorized("BACKUP_STORAGE_TARGET_FORBIDDEN", "存储目标不属于该任务", nil)
+		}
+		var modeErr error
+		selectedTransferMode, modeErr = validateTransferMode(update.StorageTargetID, update.StorageTransferMode)
+		if modeErr != nil {
+			return modeErr
+		}
+	} else if update.StorageTransferMode != "" {
+		return apperror.BadRequest("AGENT_STORAGE_TRANSFER_MODE_INVALID", "传输模式缺少对应的存储目标", nil)
+	}
+	for index := range update.StorageUploadResults {
+		result := &update.StorageUploadResults[index]
+		expectedMode, err := validateTransferMode(result.StorageTargetID, result.TransferMode)
+		if err != nil {
+			return err
+		}
+		if result.FileSize < 0 || (result.Status != "" && result.Status != "success" && result.Status != "failed") {
+			return apperror.BadRequest("AGENT_ARTIFACT_INVALID", "Agent 上报的存储结果无效", nil)
+		}
+		if result.StoragePath != "" {
+			normalizedPath, err := s.validateArtifactKey(record, task, result.StoragePath, expectedMode == storage.TransferModeMasterRelay)
+			if err != nil {
+				return err
+			}
+			result.StoragePath = normalizedPath
+		}
+		result.TransferMode = expectedMode
+	}
+	if update.StoragePath != "" {
+		if update.StorageTargetID == 0 {
+			return apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "存储路径缺少对应的存储目标", nil)
+		}
+		cleanStoragePath, err := s.validateArtifactKey(record, task, update.StoragePath, selectedTransferMode == storage.TransferModeMasterRelay)
+		if err != nil {
+			return err
+		}
+		if update.FileName != "" && path.Base(cleanStoragePath) != update.FileName {
+			return apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "Agent 上报的文件名与存储路径不一致", nil)
+		}
+		update.StoragePath = cleanStoragePath
+	}
+	if update.Status != "" && update.Status != model.BackupRecordStatusRunning && update.Status != model.BackupRecordStatusSuccess && update.Status != model.BackupRecordStatusFailed {
+		return apperror.BadRequest("BACKUP_RECORD_STATUS_INVALID", "Agent 上报的备份状态无效", nil)
+	}
+	if update.FileSize < 0 || (update.FileName != "" && (path.Base(update.FileName) != update.FileName || strings.Contains(update.FileName, "\\"))) {
+		return apperror.BadRequest("AGENT_ARTIFACT_INVALID", "Agent 上报的备份文件信息无效", nil)
+	}
+	if update.Checksum != "" {
+		checksumBytes, checksumErr := hex.DecodeString(strings.TrimSpace(update.Checksum))
+		if checksumErr != nil || len(checksumBytes) != sha256.Size {
+			return apperror.BadRequest("AGENT_ARTIFACT_INVALID", "Agent 上报的 SHA-256 无效", checksumErr)
+		}
+		update.Checksum = strings.ToLower(strings.TrimSpace(update.Checksum))
+	}
 	if update.Status != "" {
 		record.Status = update.Status
 	}
@@ -277,6 +489,7 @@ func (s *AgentService) UpdateRecord(ctx context.Context, node *model.Node, recor
 	}
 	if update.StorageTargetID > 0 {
 		record.StorageTargetID = update.StorageTargetID
+		record.StorageTransferMode = selectedTransferMode
 	}
 	if len(update.StorageUploadResults) > 0 {
 		if resultsJSON, marshalErr := json.Marshal(update.StorageUploadResults); marshalErr == nil {
@@ -312,6 +525,27 @@ func (s *AgentService) UpdateRecord(ctx context.Context, node *model.Node, recor
 	return nil
 }

+func (s *AgentService) validateArtifactKey(record *model.BackupRecord, task *model.BackupTask, objectKey string, requireRecordNamespace bool) (string, error) {
+	if record == nil || task == nil {
+		return "", apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "无法确认中转对象归属", nil)
+	}
+	rawKey := objectKey
+	cleanKey := path.Clean(rawKey)
+	fileName := path.Base(cleanKey)
+	if rawKey == "" || strings.TrimSpace(rawKey) != rawKey || cleanKey == "." || path.IsAbs(cleanKey) || strings.HasPrefix(cleanKey, "../") || cleanKey != rawKey || strings.Contains(rawKey, "\\") || fileName == "." || fileName == "/" {
+		return "", apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "中转上传对象路径不安全", nil)
+	}
+	expectedKey := backup.BuildRecordStorageKey(task.Type, record.StartedAt, record.ID, fileName)
+	if requireRecordNamespace {
+		legacyKey := backup.BuildStorageKey(task.Type, record.StartedAt, fileName)
+		if cleanKey != expectedKey && cleanKey != legacyKey {
+			return "", apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "中转上传对象不属于当前备份记录", nil)
+		}
+		return expectedKey, nil
+	}
+	return cleanKey, nil
+}
+
 func recordBelongsToNode(record *model.BackupRecord, task *model.BackupTask, nodeID uint) bool {
 	if record.NodeID != 0 {
 		return record.NodeID == nodeID
@@ -1,19 +1,26 @@
 package service

 import (
+	"bytes"
 	"context"
+	"crypto/sha256"
 	"errors"
+	"fmt"
+	"os"
 	"path/filepath"
 	"strings"
 	"testing"
 	"time"

+	"backupx/server/internal/backup"
 	"backupx/server/internal/config"
 	"backupx/server/internal/database"
 	"backupx/server/internal/logger"
 	"backupx/server/internal/model"
 	"backupx/server/internal/repository"
+	"backupx/server/internal/storage"
 	"backupx/server/internal/storage/codec"
+	storageRclone "backupx/server/internal/storage/rclone"
 	"gorm.io/gorm"
 )

@@ -42,7 +49,7 @@ func newAgentServicePoolTestHarness(t *testing.T) (*AgentService, *gorm.DB, repo
 	if err := nodeRepo.Create(context.Background(), other); err != nil {
 		t.Fatalf("create other node: %v", err)
 	}
-	targetConfig, err := cipher.EncryptJSON(map[string]any{"basePath": t.TempDir()})
+	targetConfig, err := cipher.EncryptJSON(map[string]any{"basePath": t.TempDir(), "masterRelay": true})
 	if err != nil {
 		t.Fatalf("EncryptJSON returned error: %v", err)
 	}
@@ -76,7 +83,8 @@ func newAgentServicePoolTestHarness(t *testing.T) (*AgentService, *gorm.DB, repo
 	if err := recordRepo.Create(context.Background(), record); err != nil {
 		t.Fatalf("create record: %v", err)
 	}
-	return NewAgentService(nodeRepo, taskRepo, recordRepo, storageRepo, cmdRepo, cipher), db, recordRepo, cmdRepo, owner, other
+	storageRegistry := storage.NewRegistry(storageRclone.NewLocalDiskFactory())
+	return NewAgentService(nodeRepo, taskRepo, recordRepo, storageRepo, cmdRepo, cipher, storageRegistry), db, recordRepo, cmdRepo, owner, other
 }

 func TestAgentServicePooledTaskUsesRecordNodeForSpecAndRecordUpdates(t *testing.T) {
@@ -90,19 +98,27 @@ func TestAgentServicePooledTaskUsesRecordNodeForSpecAndRecordUpdates(t *testing.
 	if spec.TaskID != 1 || len(spec.StorageTargets) != 1 {
 		t.Fatalf("unexpected spec: %#v", spec)
 	}
+	if spec.StorageTargets[0].TransferMode != storage.TransferModeMasterRelay {
+		t.Fatalf("expected local disk to use Master relay, got %#v", spec.StorageTargets[0])
+	}
 	if _, err := svc.GetTaskSpec(ctx, other, 1); err == nil {
 		t.Fatal("expected non-owner node to be forbidden from pooled task spec")
 	}
+	record, err := records.FindByID(ctx, 1)
+	if err != nil {
+		t.Fatalf("FindByID record returned error: %v", err)
+	}
+	storagePath := backup.BuildRecordStorageKey("file", record.StartedAt, record.ID, "backup.tar.gz")

 	if err := svc.UpdateRecord(ctx, owner, 1, AgentRecordUpdate{
-		Status:          model.BackupRecordStatusSuccess,
-		FileName:        "backup.tar.gz",
-		FileSize:        123,
-		StoragePath:     "tasks/1/backup.tar.gz",
-		StorageTargetID: 2,
+		Status:              model.BackupRecordStatusSuccess,
+		FileName:            "backup.tar.gz",
+		FileSize:            123,
+		StoragePath:         storagePath,
+		StorageTargetID:     1,
+		StorageTransferMode: storage.TransferModeMasterRelay,
 		StorageUploadResults: []StorageUploadResultItem{
-			{StorageTargetID: 1, StorageTargetName: "first", Status: "failed", Error: "boom"},
-			{StorageTargetID: 2, StorageTargetName: "second", Status: "success", StoragePath: "tasks/1/backup.tar.gz", FileSize: 123},
+			{StorageTargetID: 1, StorageTargetName: "local", Status: "success", StoragePath: storagePath, FileSize: 123, TransferMode: storage.TransferModeMasterRelay},
 		},
 	}); err != nil {
 		t.Fatalf("owner UpdateRecord returned error: %v", err)
@@ -114,10 +130,13 @@ func TestAgentServicePooledTaskUsesRecordNodeForSpecAndRecordUpdates(t *testing.
 	if updated.Status != model.BackupRecordStatusSuccess || updated.NodeID != owner.ID {
 		t.Fatalf("unexpected updated record: %#v", updated)
 	}
-	if updated.StorageTargetID != 2 {
-		t.Fatalf("expected successful storage target id 2, got %d", updated.StorageTargetID)
+	if updated.StorageTargetID != 1 {
+		t.Fatalf("expected successful storage target id 1, got %d", updated.StorageTargetID)
 	}
-	if !strings.Contains(updated.StorageUploadResults, `"storageTargetName":"second"`) {
+	if updated.StorageTransferMode != storage.TransferModeMasterRelay {
+		t.Fatalf("expected Master relay transfer mode, got %q", updated.StorageTransferMode)
+	}
+	if !strings.Contains(updated.StorageUploadResults, `"storageTargetName":"local"`) {
 		t.Fatalf("expected upload results to be persisted, got %q", updated.StorageUploadResults)
 	}
 	if err := svc.UpdateRecord(ctx, other, 1, AgentRecordUpdate{LogAppend: "bad"}); err == nil {
@@ -125,6 +144,157 @@ func TestAgentServicePooledTaskUsesRecordNodeForSpecAndRecordUpdates(t *testing.
 	}
 }

+func TestAgentServiceRelaysRemoteArtifactToMasterLocalDisk(t *testing.T) {
+	svc, _, records, _, owner, other := newAgentServicePoolTestHarness(t)
+	ctx := context.Background()
+	payload := []byte("artifact from remote source server")
+	digest := sha256.Sum256(payload)
+	checksum := fmt.Sprintf("%x", digest[:])
+	record, err := records.FindByID(ctx, 1)
+	if err != nil {
+		t.Fatalf("FindByID record returned error: %v", err)
+	}
+	objectKey := backup.BuildRecordStorageKey("file", record.StartedAt, record.ID, "remote-source.tar")
+
+	if err := svc.UploadArtifact(ctx, owner, 1, 1, objectKey, int64(len(payload)), checksum, bytes.NewReader(payload)); err != nil {
+		t.Fatalf("UploadArtifact returned error: %v", err)
+	}
+	target, err := svc.storageRepo.FindByID(ctx, 1)
+	if err != nil || target == nil {
+		t.Fatalf("FindByID target: target=%#v err=%v", target, err)
+	}
+	config := map[string]any{}
+	if err := svc.cipher.DecryptJSON(target.ConfigCiphertext, &config); err != nil {
+		t.Fatalf("DecryptJSON target config: %v", err)
+	}
+	basePath, _ := config["basePath"].(string)
+	stored, err := os.ReadFile(filepath.Join(basePath, filepath.FromSlash(objectKey)))
+	if err != nil {
+		t.Fatalf("read relayed artifact: %v", err)
+	}
+	if !bytes.Equal(stored, payload) {
+		t.Fatalf("relayed artifact differs: got %q", stored)
+	}
+	if err := svc.UploadArtifact(ctx, other, 1, 1, objectKey, int64(len(payload)), checksum, bytes.NewReader(payload)); err == nil {
+		t.Fatal("expected a different node to be forbidden from relaying the artifact")
+	}
+
+	legacyPayload := []byte("artifact from an older Agent")
+	legacyDigest := sha256.Sum256(legacyPayload)
+	legacyKey := backup.BuildStorageKey("file", record.StartedAt, "legacy-agent.tar")
+	canonicalKey := backup.BuildRecordStorageKey("file", record.StartedAt, record.ID, "legacy-agent.tar")
+	if err := svc.UploadArtifact(ctx, owner, record.ID, target.ID, legacyKey, int64(len(legacyPayload)), fmt.Sprintf("%x", legacyDigest[:]), bytes.NewReader(legacyPayload)); err != nil {
+		t.Fatalf("UploadArtifact legacy key returned error: %v", err)
+	}
+	stored, err = os.ReadFile(filepath.Join(basePath, filepath.FromSlash(canonicalKey)))
+	if err != nil || !bytes.Equal(stored, legacyPayload) {
+		t.Fatalf("legacy Agent artifact was not normalized: data=%q err=%v", stored, err)
+	}
+	if _, err := os.Stat(filepath.Join(basePath, filepath.FromSlash(legacyKey))); !errors.Is(err, os.ErrNotExist) {
+		t.Fatalf("legacy object key should not be written directly: %v", err)
+	}
+	if err := svc.UpdateRecord(ctx, owner, record.ID, AgentRecordUpdate{
+		Status:          model.BackupRecordStatusSuccess,
+		FileName:        "legacy-agent.tar",
+		FileSize:        int64(len(legacyPayload)),
+		Checksum:        fmt.Sprintf("%x", legacyDigest[:]),
+		StoragePath:     legacyKey,
+		StorageTargetID: target.ID,
+		StorageUploadResults: []StorageUploadResultItem{{
+			StorageTargetID: target.ID,
+			Status:          "success",
+			StoragePath:     legacyKey,
+			FileSize:        int64(len(legacyPayload)),
+		}},
+	}); err != nil {
+		t.Fatalf("UpdateRecord legacy key returned error: %v", err)
+	}
+	updated, err := records.FindByID(ctx, record.ID)
+	if err != nil {
+		t.Fatalf("FindByID updated record returned error: %v", err)
+	}
+	if updated.StoragePath != canonicalKey || updated.StorageTransferMode != storage.TransferModeMasterRelay || !strings.Contains(updated.StorageUploadResults, canonicalKey) {
+		t.Fatalf("legacy Agent record was not normalized: %#v", updated)
+	}
+}
+
+func TestAgentServiceRejectsArtifactOutsideRecordNamespace(t *testing.T) {
+	svc, _, records, _, owner, _ := newAgentServicePoolTestHarness(t)
+	ctx := context.Background()
+	record, err := records.FindByID(ctx, 1)
+	if err != nil {
+		t.Fatalf("FindByID record returned error: %v", err)
+	}
+	target, err := svc.storageRepo.FindByID(ctx, 1)
+	if err != nil || target == nil {
+		t.Fatalf("FindByID target: target=%#v err=%v", target, err)
+	}
+	config := map[string]any{}
+	if err := svc.cipher.DecryptJSON(target.ConfigCiphertext, &config); err != nil {
+		t.Fatalf("DecryptJSON target config: %v", err)
+	}
+	basePath, _ := config["basePath"].(string)
+	victimKey := backup.BuildRecordStorageKey("file", record.StartedAt, record.ID+1, "victim.tar")
+	victimPath := filepath.Join(basePath, filepath.FromSlash(victimKey))
+	if err := os.MkdirAll(filepath.Dir(victimPath), 0o755); err != nil {
+		t.Fatalf("MkdirAll victim parent: %v", err)
+	}
+	if err := os.WriteFile(victimPath, []byte("keep me"), 0o600); err != nil {
+		t.Fatalf("WriteFile victim: %v", err)
+	}
+	payload := []byte("overwrite")
+	digest := sha256.Sum256(payload)
+	if err := svc.UploadArtifact(ctx, owner, record.ID, target.ID, victimKey, int64(len(payload)), fmt.Sprintf("%x", digest[:]), bytes.NewReader(payload)); err == nil {
+		t.Fatal("expected another record namespace to be rejected")
+	}
+	stored, err := os.ReadFile(victimPath)
+	if err != nil {
+		t.Fatalf("ReadFile victim: %v", err)
+	}
+	if string(stored) != "keep me" {
+		t.Fatalf("victim object changed: %q", stored)
+	}
+	if err := svc.UpdateRecord(ctx, owner, record.ID, AgentRecordUpdate{StoragePath: victimKey, StorageTargetID: target.ID, StorageTransferMode: storage.TransferModeMasterRelay}); err == nil {
+		t.Fatal("expected another record namespace in status update to be rejected")
+	}
+}
+
+func TestAgentServiceKeepsExistingLocalDiskTargetsAgentLocal(t *testing.T) {
+	svc, _, records, _, owner, _ := newAgentServicePoolTestHarness(t)
+	ctx := context.Background()
+	target, err := svc.storageRepo.FindByID(ctx, 1)
+	if err != nil || target == nil {
+		t.Fatalf("FindByID target: target=%#v err=%v", target, err)
+	}
+	legacyConfig, err := svc.cipher.EncryptJSON(map[string]any{"basePath": t.TempDir()})
+	if err != nil {
+		t.Fatalf("EncryptJSON legacy target: %v", err)
+	}
+	target.ConfigCiphertext = legacyConfig
+	if err := svc.storageRepo.Update(ctx, target); err != nil {
+		t.Fatalf("Update legacy target: %v", err)
+	}
+
+	spec, err := svc.GetTaskSpec(ctx, owner, 1)
+	if err != nil {
+		t.Fatalf("GetTaskSpec returned error: %v", err)
+	}
+	if len(spec.StorageTargets) != 1 || spec.StorageTargets[0].TransferMode != storage.TransferModeDirect {
+		t.Fatalf("expected legacy local disk to stay Agent-local, got %#v", spec.StorageTargets)
+	}
+	payload := []byte("must not be relayed")
+	digest := sha256.Sum256(payload)
+	record, findErr := records.FindByID(ctx, 1)
+	if findErr != nil {
+		t.Fatalf("FindByID record returned error: %v", findErr)
+	}
+	objectKey := backup.BuildRecordStorageKey("file", record.StartedAt, record.ID, "legacy.tar")
+	err = svc.UploadArtifact(ctx, owner, 1, 1, objectKey, int64(len(payload)), fmt.Sprintf("%x", digest[:]), bytes.NewReader(payload))
+	if err == nil {
+		t.Fatal("expected relay upload to be rejected for an Agent-local target")
+	}
+}
+
 func TestAgentServiceUpdateRecordRefreshesTaskSummaryOnTerminalStatus(t *testing.T) {
 	for _, status := range []string{model.BackupRecordStatusSuccess, model.BackupRecordStatusFailed} {
 		t.Run(status, func(t *testing.T) {
@@ -5,11 +5,13 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"hash"
 	"io"
 	"os"
 	"path/filepath"
+	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -49,6 +51,7 @@ type StorageUploadResultItem struct {
 	Status            string `json:"status"`
 	StoragePath       string `json:"storagePath,omitempty"`
 	FileSize          int64  `json:"fileSize,omitempty"`
+	TransferMode      string `json:"transferMode,omitempty"`
 	Error             string `json:"error,omitempty"`
 }

@@ -62,6 +65,17 @@ type DownloadedArtifact struct {
 	Reader   io.ReadCloser
 }

+type temporaryArtifactReader struct {
+	*os.File
+	directory string
+}
+
+func (r *temporaryArtifactReader) Close() error {
+	closeErr := r.File.Close()
+	removeErr := os.RemoveAll(r.directory)
+	return errors.Join(closeErr, removeErr)
+}
+
 // collectTargetIDs 获取任务关联的所有存储目标 ID
 func collectTargetIDs(task *model.BackupTask) []uint {
 	if len(task.StorageTargets) > 0 {
@@ -102,6 +116,10 @@ type BackupExecutionService struct {
 	bandwidthLimit string // rclone 带宽限制（全局默认，节点配置可覆盖）
 	metrics        *metrics.Metrics
 	taskLocks      sync.Map
+	// repositoryLocks serializes immutable index updates per storage target.
+	// Repository mode is intentionally single-writer in v1 to avoid orphaned
+	// duplicate packs when two local tasks discover the same missing chunk.
+	repositoryLocks sync.Map
 }

 // SetMetrics 注入 Prometheus 采集器。nil 时所有埋点退化为 no-op。
@@ -211,6 +229,25 @@ func (s *BackupExecutionService) DownloadRecord(ctx context.Context, recordID ui
 	if err != nil {
 		return nil, err
 	}
+	if record.BackupKind == model.BackupKindRepository {
+		tempDir, err := os.MkdirTemp(s.tempDir, "repository-download-*")
+		if err != nil {
+			return nil, apperror.Internal("BACKUP_RECORD_DOWNLOAD_FAILED", "无法创建 CDC 导出目录", err)
+		}
+		exportName := fmt.Sprintf("backupx-record-%d.tar", record.ID)
+		exportPath := filepath.Join(tempDir, exportName)
+		store := backup.NewRepositoryStore(s.cipher.Key())
+		if err := store.ExportTar(ctx, provider, record.StoragePath, record.Checksum, exportPath); err != nil {
+			cleanupErr := os.RemoveAll(tempDir)
+			return nil, apperror.Internal("BACKUP_RECORD_DOWNLOAD_FAILED", "无法从 CDC 仓库导出归档", errors.Join(err, cleanupErr))
+		}
+		file, err := os.Open(exportPath)
+		if err != nil {
+			cleanupErr := os.RemoveAll(tempDir)
+			return nil, apperror.Internal("BACKUP_RECORD_DOWNLOAD_FAILED", "无法打开 CDC 导出归档", errors.Join(err, cleanupErr))
+		}
+		return &DownloadedArtifact{FileName: exportName, Reader: &temporaryArtifactReader{File: file, directory: tempDir}}, nil
+	}
 	reader, err := provider.Download(ctx, record.StoragePath)
 	if err != nil {
 		return nil, apperror.Internal("BACKUP_RECORD_DOWNLOAD_FAILED", "无法下载备份文件", err)
@@ -234,6 +271,16 @@ func (s *BackupExecutionService) RestoreRecord(ctx context.Context, recordID uin
 	if task == nil {
 		return apperror.New(404, "BACKUP_TASK_NOT_FOUND", "关联的备份任务不存在，无法执行恢复", fmt.Errorf("backup task %d not found", record.TaskID))
 	}
+	if record.BackupKind == model.BackupKindRepository {
+		spec, specErr := s.buildTaskSpec(task, record.StartedAt)
+		if specErr != nil {
+			return specErr
+		}
+		if err := backup.NewRepositoryStore(s.cipher.Key()).Restore(ctx, provider, record.StoragePath, record.Checksum, spec, backup.NopLogWriter{}); err != nil {
+			return apperror.Internal("BACKUP_RECORD_RESTORE_FAILED", "从 CDC 仓库恢复备份失败", err)
+		}
+		return nil
+	}
 	tempDir, err := os.MkdirTemp("", "backupx-restore-*")
 	if err != nil {
 		return apperror.Internal("BACKUP_RECORD_RESTORE_FAILED", "无法创建恢复目录", err)
@@ -291,6 +338,58 @@ func (s *BackupExecutionService) DeleteRecord(ctx context.Context, recordID uint
 				fmt.Sprintf("该全量备份仍有 %d 个差异备份依赖它，删除会导致这些差异无法恢复。请先删除相关差异备份或等待其过期。", deps), nil)
 		}
 	}
+	if record.BackupKind == model.BackupKindRepository {
+		copies := []StorageUploadResultItem{{
+			StorageTargetID: record.StorageTargetID,
+			Status:          model.BackupRecordStatusSuccess,
+			StoragePath:     record.StoragePath,
+		}}
+		if strings.TrimSpace(record.StorageUploadResults) != "" {
+			if err := json.Unmarshal([]byte(record.StorageUploadResults), &copies); err != nil {
+				return apperror.Internal("BACKUP_RECORD_DELETE_FAILED", "无法解析 CDC 仓库副本信息，已停止删除以避免遗留数据", err)
+			}
+		}
+		copyPaths := make(map[uint]string, len(copies))
+		for _, copy := range copies {
+			if strings.EqualFold(copy.Status, model.BackupRecordStatusSuccess) && strings.TrimSpace(copy.StoragePath) != "" {
+				copyPaths[copy.StorageTargetID] = copy.StoragePath
+			}
+		}
+		targetIDs := make([]uint, 0, len(copyPaths))
+		for targetID := range copyPaths {
+			targetIDs = append(targetIDs, targetID)
+		}
+		sort.Slice(targetIDs, func(i, j int) bool { return targetIDs[i] < targetIDs[j] })
+		unlocks := make([]func(), 0, len(targetIDs))
+		for _, targetID := range targetIDs {
+			unlocks = append(unlocks, s.acquireRepositoryLock(targetID))
+		}
+		defer func() {
+			for index := len(unlocks) - 1; index >= 0; index-- {
+				unlocks[index]()
+			}
+		}()
+		providers := make(map[uint]storage.StorageProvider, len(targetIDs))
+		for _, targetID := range targetIDs {
+			provider, resolveErr := s.resolveProvider(ctx, targetID)
+			if resolveErr != nil {
+				return resolveErr
+			}
+			if deleteErr := provider.Delete(ctx, copyPaths[targetID]); deleteErr != nil {
+				return apperror.Internal("BACKUP_RECORD_DELETE_FAILED", "无法删除 CDC 仓库快照", deleteErr)
+			}
+			providers[targetID] = provider
+		}
+		for targetID, provider := range providers {
+			if _, pruneErr := backup.NewRepositoryStore(s.cipher.Key()).Prune(ctx, provider); pruneErr != nil {
+				return apperror.Internal("BACKUP_REPOSITORY_PRUNE_FAILED", fmt.Sprintf("无法清理存储目标 %d 的 CDC 仓库；记录暂时保留以便重试", targetID), pruneErr)
+			}
+		}
+		if err := s.records.Delete(ctx, recordID); err != nil {
+			return apperror.Internal("BACKUP_RECORD_DELETE_FAILED", "无法删除备份记录", err)
+		}
+		return nil
+	}
 	if remote, err := s.deleteRemoteLocalDiskObject(ctx, record); err != nil {
 		return err
 	} else if !remote && strings.TrimSpace(record.StoragePath) != "" {
@@ -312,6 +411,9 @@ func (s *BackupExecutionService) deleteRemoteLocalDiskObject(ctx context.Context
 	if strings.TrimSpace(record.StoragePath) == "" || s.nodeRepo == nil {
 		return false, nil
 	}
+	if record.StorageTransferMode == storage.TransferModeMasterRelay {
+		return false, nil
+	}
 	node, err := s.nodeRepo.FindByID(ctx, record.NodeID)
 	if err != nil || node == nil || node.IsLocal {
 		return false, nil
@@ -384,6 +486,9 @@ func (s *BackupExecutionService) startTask(ctx context.Context, id uint, async b
 			return nil, perr
 		}
 	}
+	if strings.EqualFold(task.BackupMode, model.BackupModeRepository) && s.resolveRemoteNode(ctx, resolvedNodeID) != nil {
+		return nil, apperror.BadRequest("BACKUP_TASK_REPOSITORY_REMOTE_UNSUPPORTED", "CDC 仓库模式当前仅支持 Master 本机单写者执行", nil)
+	}
 	startedAt := s.now()
 	// 取第一个存储目标 ID 做兼容
 	primaryTargetID := task.StorageTargetID
@@ -630,6 +735,141 @@ func (s *BackupExecutionService) resolveDifferentialBase(ctx context.Context, ta
 	return 0, backup.Manifest{}, false
 }

+type repositoryTaskResult struct {
+	fileName        string
+	logicalSize     int64
+	checksum        string
+	storagePath     string
+	storageTargetID uint
+	manifestJSON    string
+	uploadResults   []StorageUploadResultItem
+	providers       map[uint]storage.StorageProvider
+}
+
+func (s *BackupExecutionService) executeRepositoryTask(ctx context.Context, task *model.BackupTask, recordID uint, startedAt time.Time, spec backup.TaskSpec, logger *backup.ExecutionLogger) (*repositoryTaskResult, error) {
+	store := backup.NewRepositoryStore(s.cipher.Key())
+	plan, err := store.BuildPlan(ctx, spec, logger)
+	if err != nil {
+		return nil, err
+	}
+	defer func() {
+		if closeErr := plan.Close(); closeErr != nil {
+			logger.Warnf("清理 CDC 临时计划失败：%v", closeErr)
+		}
+	}()
+	manifestBytes, err := backup.EncodeManifest(plan.Manifest)
+	if err != nil {
+		return nil, fmt.Errorf("encode repository manifest: %w", err)
+	}
+	targetIDs := collectTargetIDs(task)
+	if len(targetIDs) == 0 {
+		return nil, fmt.Errorf("没有关联的存储目标")
+	}
+	storageUsage, usageErr := s.storageUsageSnapshot(ctx)
+	if usageErr != nil {
+		logger.Warnf("读取存储目标用量失败，跳过本次软配额校验：%v", usageErr)
+		storageUsage = map[uint]int64{}
+	}
+
+	snapshotKey := store.SnapshotKey(task.ID, recordID, startedAt)
+	result := &repositoryTaskResult{
+		fileName:      filepath.Base(snapshotKey),
+		logicalSize:   plan.LogicalSize,
+		storagePath:   snapshotKey,
+		manifestJSON:  string(manifestBytes),
+		uploadResults: make([]StorageUploadResultItem, 0, len(targetIDs)),
+		providers:     make(map[uint]storage.StorageProvider),
+	}
+	var failures []string
+	for _, targetID := range targetIDs {
+		target, findErr := s.targets.FindByID(ctx, targetID)
+		targetName := fmt.Sprintf("target-%d", targetID)
+		if findErr == nil && target != nil {
+			targetName = target.Name
+		}
+		if findErr != nil || target == nil {
+			message := "存储目标不存在"
+			if findErr != nil {
+				message = findErr.Error()
+			}
+			result.uploadResults = append(result.uploadResults, StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: message})
+			failures = append(failures, fmt.Sprintf("%s: %s", targetName, message))
+			continue
+		}
+		provider, resolveErr := s.resolveProviderForNode(ctx, targetID, task.NodeID)
+		if resolveErr != nil {
+			result.uploadResults = append(result.uploadResults, StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: resolveErr.Error()})
+			failures = append(failures, fmt.Sprintf("%s: %v", targetName, resolveErr))
+			continue
+		}
+		logger.Infof("同步 CDC 仓库到存储目标：%s", targetName)
+		unlock := s.acquireRepositoryLock(targetID)
+		estimatedSize, estimateErr := store.EstimateUploadSize(ctx, provider, plan)
+		if estimateErr != nil {
+			unlock()
+			result.uploadResults = append(result.uploadResults, StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: estimateErr.Error()})
+			failures = append(failures, fmt.Sprintf("%s: %v", targetName, estimateErr))
+			continue
+		}
+		if target.QuotaBytes > 0 && storageUsage[targetID]+estimatedSize > target.QuotaBytes {
+			unlock()
+			message := fmt.Sprintf("超出存储目标配额（%d + 预计 %d > %d）", storageUsage[targetID], estimatedSize, target.QuotaBytes)
+			result.uploadResults = append(result.uploadResults, StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: message})
+			failures = append(failures, fmt.Sprintf("%s: %s", targetName, message))
+			continue
+		}
+		upload, uploadErr := store.Upload(ctx, provider, plan, snapshotKey)
+		unlock()
+		if uploadErr != nil {
+			result.uploadResults = append(result.uploadResults, StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: uploadErr.Error()})
+			failures = append(failures, fmt.Sprintf("%s: %v", targetName, uploadErr))
+			logger.Warnf("存储目标 %s CDC 仓库同步失败：%v", targetName, uploadErr)
+			continue
+		}
+		result.uploadResults = append(result.uploadResults, StorageUploadResultItem{
+			StorageTargetID: targetID, StorageTargetName: targetName, Status: "success",
+			StoragePath: upload.SnapshotKey, FileSize: upload.UploadedBytes,
+		})
+		result.providers[targetID] = provider
+		if result.storageTargetID == 0 {
+			result.storageTargetID = targetID
+			result.checksum = upload.Checksum
+		}
+		logger.Infof("存储目标 %s CDC 同步完成：新块 %d/%d，复用 %d bytes，实际上传 %d bytes", targetName, upload.NewChunks, upload.UniqueChunks, upload.ReusedBytes, upload.UploadedBytes)
+	}
+	if result.storageTargetID == 0 {
+		return nil, fmt.Errorf("所有存储目标 CDC 仓库同步均失败：%s", strings.Join(failures, "; "))
+	}
+	if len(failures) > 0 {
+		logger.Warnf("部分存储目标 CDC 仓库同步失败：%s", strings.Join(failures, "; "))
+	}
+	if s.dependentsResolver != nil {
+		go func(upstreamID uint, upstreamName string) {
+			dependents, resolveErr := s.dependentsResolver.TriggerDependents(context.Background(), upstreamID)
+			if resolveErr != nil {
+				logger.Warnf("解析任务 %s 的下游依赖失败：%v", upstreamName, resolveErr)
+				return
+			}
+			for _, dependentID := range dependents {
+				if _, runErr := s.RunTaskByID(context.Background(), dependentID); runErr != nil {
+					logger.Warnf("触发下游任务 #%d 失败（上游: %s）：%v", dependentID, upstreamName, runErr)
+				} else {
+					logger.Infof("已触发下游任务 #%d（上游: %s）", dependentID, upstreamName)
+				}
+			}
+		}(task.ID, task.Name)
+	}
+	return result, nil
+}
+
+func (s *BackupExecutionService) acquireRepositoryLock(targetID uint) func() {
+	created := &sync.Mutex{}
+	actual, _ := s.repositoryLocks.LoadOrStore(targetID, created)
+	lock := actual.(*sync.Mutex)
+	lock.Lock()
+	return lock.Unlock
+}
+
 func (s *BackupExecutionService) executeTask(ctx context.Context, task *model.BackupTask, recordID uint, startedAt time.Time) {
 	// 节点级并发限流：当任务绑定节点且节点配置了 MaxConcurrent>0，
 	// 该节点上所有任务共享一个节点专属 semaphore，互相排队
@@ -658,18 +898,33 @@ func (s *BackupExecutionService) executeTask(ctx context.Context, task *model.Ba
 	backupKind := model.BackupKindFull
 	var baseRecordID uint
 	var manifestJSON string
+	var repositoryProviders map[uint]storage.StorageProvider
 	completeRecord := func() {
+		readyForRepositoryRetention := status == model.BackupRecordStatusSuccess
 		if finalizeErr := s.finalizeRecord(ctx, task, recordID, startedAt, status, errMessage, logger.String(), fileName, fileSize, checksum, storagePath, selectedStorageTargetID); finalizeErr != nil {
 			logger.Errorf("写回备份记录失败：%v", finalizeErr)
+			readyForRepositoryRetention = false
 		}
 		// 采集任务执行结果到 Prometheus（耗时 + 产出字节 + 状态计数）
 		s.metrics.ObserveTaskRun(task.Type, status, time.Since(startedAt).Seconds(), fileSize)
 		// 写入多目标上传结果
 		if len(uploadResults) > 0 {
-			if resultsJSON, marshalErr := json.Marshal(uploadResults); marshalErr == nil {
-				if record, findErr := s.records.FindByID(ctx, recordID); findErr == nil && record != nil {
-					record.StorageUploadResults = string(resultsJSON)
-					_ = s.records.Update(ctx, record)
+			resultsJSON, marshalErr := json.Marshal(uploadResults)
+			if marshalErr != nil {
+				logger.Warnf("序列化多目标上传结果失败：%v", marshalErr)
+				readyForRepositoryRetention = false
+			} else if record, findErr := s.records.FindByID(ctx, recordID); findErr != nil || record == nil {
+				if findErr != nil {
+					logger.Warnf("读取备份记录以写回多目标结果失败：%v", findErr)
+				} else {
+					logger.Warnf("备份记录 #%d 不存在，无法写回多目标结果", recordID)
+				}
+				readyForRepositoryRetention = false
+			} else {
+				record.StorageUploadResults = string(resultsJSON)
+				if updateErr := s.records.Update(ctx, record); updateErr != nil {
+					logger.Warnf("写回多目标上传结果失败：%v", updateErr)
+					readyForRepositoryRetention = false
 				}
 			}
 		}
@@ -681,6 +936,36 @@ func (s *BackupExecutionService) executeTask(ctx context.Context, task *model.Ba
 				record.Manifest = manifestJSON
 				if updErr := s.records.Update(ctx, record); updErr != nil {
 					logger.Warnf("写回差异链信息失败：%v", updErr)
+					readyForRepositoryRetention = false
+				}
+			} else {
+				if findErr != nil {
+					logger.Warnf("读取备份记录以写回备份类型失败：%v", findErr)
+				} else {
+					logger.Warnf("备份记录 #%d 不存在，无法写回备份类型", recordID)
+				}
+				readyForRepositoryRetention = false
+			}
+		}
+		if readyForRepositoryRetention && backupKind == model.BackupKindRepository && s.retention != nil && len(repositoryProviders) > 0 {
+			targetIDs := make([]uint, 0, len(repositoryProviders))
+			for targetID := range repositoryProviders {
+				targetIDs = append(targetIDs, targetID)
+			}
+			sort.Slice(targetIDs, func(i, j int) bool { return targetIDs[i] < targetIDs[j] })
+			unlocks := make([]func(), 0, len(targetIDs))
+			for _, targetID := range targetIDs {
+				unlocks = append(unlocks, s.acquireRepositoryLock(targetID))
+			}
+			cleanupResult, cleanupErr := s.retention.CleanupProviders(ctx, task, repositoryProviders)
+			for index := len(unlocks) - 1; index >= 0; index-- {
+				unlocks[index]()
+			}
+			if cleanupErr != nil {
+				logger.Warnf("执行 CDC 仓库保留策略失败：%v", cleanupErr)
+			} else {
+				for _, warning := range cleanupResult.Warnings {
+					logger.Warnf("CDC 仓库保留策略警告：%s", warning)
 				}
 			}
 		}
@@ -701,6 +986,26 @@ func (s *BackupExecutionService) executeTask(ctx context.Context, task *model.Ba
 		logger.Errorf("构建任务运行时配置失败：%v", err)
 		return
 	}
+	if task.Type == model.BackupTaskTypeFile && strings.EqualFold(task.BackupMode, model.BackupModeRepository) {
+		backupKind = model.BackupKindRepository
+		repositoryResult, repositoryErr := s.executeRepositoryTask(ctx, task, recordID, startedAt, spec, logger)
+		if repositoryErr != nil {
+			errMessage = repositoryErr.Error()
+			logger.Errorf("执行 CDC 仓库备份失败：%v", repositoryErr)
+			return
+		}
+		fileName = repositoryResult.fileName
+		fileSize = repositoryResult.logicalSize
+		checksum = repositoryResult.checksum
+		storagePath = repositoryResult.storagePath
+		selectedStorageTargetID = repositoryResult.storageTargetID
+		uploadResults = repositoryResult.uploadResults
+		repositoryProviders = repositoryResult.providers
+		manifestJSON = repositoryResult.manifestJSON
+		status = model.BackupRecordStatusSuccess
+		logger.Infof("CDC 仓库备份执行完成")
+		return
+	}
 	// 差异备份：解析基线全量，命中则切换为差异模式（仅本机文件任务）。
 	if baseID, baseManifest, ok := s.resolveDifferentialBase(ctx, task); ok {
 		spec.Differential = true
@@ -1,6 +1,7 @@
 package service

 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -188,6 +189,108 @@ func TestBackupExecutionServiceSQLiteBackupRemainsFull(t *testing.T) {
 	}
 }

+func TestBackupExecutionServiceRepositoryModeRoundTrip(t *testing.T) {
+	executionService, recordService, tasks, _, records, sourceDir, storageDir := newExecutionTestServices(t)
+	ctx := context.Background()
+	task, err := tasks.FindByID(ctx, 1)
+	if err != nil {
+		t.Fatalf("FindByID task returned error: %v", err)
+	}
+	task.BackupMode = model.BackupModeRepository
+	task.Compression = "zstd"
+	if err := tasks.Update(ctx, task); err != nil {
+		t.Fatalf("Update repository task returned error: %v", err)
+	}
+	large := make([]byte, 4<<20)
+	for index := range large {
+		large[index] = byte((index * 31) % 251)
+	}
+	largePath := filepath.Join(sourceDir, "large.bin")
+	if err := os.WriteFile(largePath, large, 0o640); err != nil {
+		t.Fatalf("write large fixture: %v", err)
+	}
+
+	first, err := executionService.RunTaskByIDSync(ctx, task.ID)
+	if err != nil {
+		t.Fatalf("first repository backup returned error: %v", err)
+	}
+	if first.Status != model.BackupRecordStatusSuccess || first.BackupKind != model.BackupKindRepository {
+		t.Fatalf("unexpected first repository record: %#v", first)
+	}
+	if !strings.HasPrefix(first.StoragePath, ".backupx/repository/v1/snapshots/") {
+		t.Fatalf("unexpected repository snapshot path: %s", first.StoragePath)
+	}
+
+	large[2<<20] ^= 0xff
+	if err := os.WriteFile(largePath, large, 0o640); err != nil {
+		t.Fatalf("modify large fixture: %v", err)
+	}
+	second, err := executionService.RunTaskByIDSync(ctx, task.ID)
+	if err != nil {
+		t.Fatalf("second repository backup returned error: %v", err)
+	}
+	stored, err := records.FindByID(ctx, second.ID)
+	if err != nil {
+		t.Fatalf("FindByID repository record returned error: %v", err)
+	}
+	if stored == nil || stored.BackupKind != model.BackupKindRepository || stored.Manifest == "" {
+		t.Fatalf("repository metadata was not persisted: %#v", stored)
+	}
+
+	download, err := recordService.Download(ctx, second.ID)
+	if err != nil {
+		t.Fatalf("export repository snapshot returned error: %v", err)
+	}
+	exported, readErr := io.ReadAll(download.Reader)
+	closeErr := download.Reader.Close()
+	if readErr != nil || closeErr != nil {
+		t.Fatalf("read repository export: read=%v close=%v", readErr, closeErr)
+	}
+	if len(exported) == 0 || !strings.HasSuffix(download.FileName, ".tar") {
+		t.Fatalf("unexpected repository export: name=%s size=%d", download.FileName, len(exported))
+	}
+
+	if err := os.WriteFile(largePath, bytes.Repeat([]byte{0}, len(large)), 0o640); err != nil {
+		t.Fatalf("damage source before restore: %v", err)
+	}
+	if err := executionService.RestoreRecord(ctx, second.ID); err != nil {
+		t.Fatalf("restore repository record returned error: %v", err)
+	}
+	restored, err := os.ReadFile(largePath)
+	if err != nil {
+		t.Fatalf("read restored source: %v", err)
+	}
+	if !bytes.Equal(restored, large) {
+		t.Fatalf("repository restore did not reproduce the source")
+	}
+
+	if err := recordService.Delete(ctx, first.ID); err != nil {
+		t.Fatalf("delete first repository record: %v", err)
+	}
+	if err := recordService.Delete(ctx, second.ID); err != nil {
+		t.Fatalf("delete second repository record: %v", err)
+	}
+	packRoot := filepath.Join(storageDir, filepath.FromSlash(".backupx/repository/v1/packs"))
+	remainingPacks := 0
+	if err := filepath.Walk(packRoot, func(_ string, info os.FileInfo, walkErr error) error {
+		if walkErr != nil {
+			if os.IsNotExist(walkErr) {
+				return nil
+			}
+			return walkErr
+		}
+		if info != nil && !info.IsDir() {
+			remainingPacks++
+		}
+		return nil
+	}); err != nil {
+		t.Fatalf("inspect repository packs: %v", err)
+	}
+	if remainingPacks != 0 {
+		t.Fatalf("repository prune left %d packs after deleting all snapshots", remainingPacks)
+	}
+}
+
 func TestBackupExecutionServiceNodePoolSelectionDoesNotPersistTaskNodeID(t *testing.T) {
 	executionService, _, tasks, _, records, _, _ := newExecutionTestServices(t)
 	ctx := context.Background()
@@ -359,6 +462,56 @@ func TestBackupExecutionServiceRestoreRecordRejectsRemoteLocalDisk(t *testing.T)
 	}
 }

+func TestBackupExecutionServiceDownloadsMasterRelayedLocalDiskRecord(t *testing.T) {
+	executionService, _, tasks, _, records, _, storageDir := newExecutionTestServices(t)
+	ctx := context.Background()
+	executionService.SetClusterDependencies(&nodeRepoStub{nodes: []model.Node{
+		{ID: 10, Name: "edge-a", Token: "edge-a-token", Status: model.NodeStatusOnline},
+	}}, &fakeDispatcher{})
+	task, err := tasks.FindByID(ctx, 1)
+	if err != nil {
+		t.Fatalf("FindByID task returned error: %v", err)
+	}
+	storagePath := "file/2026/05/09/relayed.tar"
+	artifactPath := filepath.Join(storageDir, filepath.FromSlash(storagePath))
+	if err := os.MkdirAll(filepath.Dir(artifactPath), 0o755); err != nil {
+		t.Fatalf("MkdirAll artifact parent returned error: %v", err)
+	}
+	content := []byte("stored on Master")
+	if err := os.WriteFile(artifactPath, content, 0o600); err != nil {
+		t.Fatalf("WriteFile artifact returned error: %v", err)
+	}
+	completedAt := time.Now().UTC()
+	record := &model.BackupRecord{
+		TaskID:              task.ID,
+		StorageTargetID:     task.StorageTargetID,
+		NodeID:              10,
+		Status:              model.BackupRecordStatusSuccess,
+		FileName:            "relayed.tar",
+		FileSize:            int64(len(content)),
+		StoragePath:         storagePath,
+		StorageTransferMode: storage.TransferModeMasterRelay,
+		StartedAt:           completedAt.Add(-time.Second),
+		CompletedAt:         &completedAt,
+	}
+	if err := records.Create(ctx, record); err != nil {
+		t.Fatalf("Create record returned error: %v", err)
+	}
+
+	download, err := executionService.DownloadRecord(ctx, record.ID)
+	if err != nil {
+		t.Fatalf("DownloadRecord returned error: %v", err)
+	}
+	got, readErr := io.ReadAll(download.Reader)
+	closeErr := download.Reader.Close()
+	if readErr != nil || closeErr != nil {
+		t.Fatalf("read relayed artifact: read=%v close=%v", readErr, closeErr)
+	}
+	if !bytes.Equal(got, content) {
+		t.Fatalf("downloaded content = %q, want %q", got, content)
+	}
+}
+
 func TestBackupExecutionServiceRecordsFirstSuccessfulStorageTarget(t *testing.T) {
 	executionService, _, tasks, targets, records, _, _ := newExecutionTestServices(t)
 	ctx := context.Background()
@@ -23,22 +23,23 @@ type BackupRecordListInput struct {
 }

 type BackupRecordSummary struct {
-	ID                uint       `json:"id"`
-	TaskID            uint       `json:"taskId"`
-	TaskName          string     `json:"taskName"`
-	StorageTargetID   uint       `json:"storageTargetId"`
-	StorageTargetName string     `json:"storageTargetName"`
-	Status            string     `json:"status"`
-	FileName          string     `json:"fileName"`
-	FileSize          int64      `json:"fileSize"`
-	Checksum          string     `json:"checksum"`
-	StoragePath       string     `json:"storagePath"`
-	DurationSeconds   int        `json:"durationSeconds"`
-	ErrorMessage      string     `json:"errorMessage"`
-	StartedAt         time.Time  `json:"startedAt"`
-	CompletedAt       *time.Time `json:"completedAt,omitempty"`
-	Locked            bool       `json:"locked"`
-	BackupKind        string     `json:"backupKind"`
+	ID                  uint       `json:"id"`
+	TaskID              uint       `json:"taskId"`
+	TaskName            string     `json:"taskName"`
+	StorageTargetID     uint       `json:"storageTargetId"`
+	StorageTargetName   string     `json:"storageTargetName"`
+	Status              string     `json:"status"`
+	FileName            string     `json:"fileName"`
+	FileSize            int64      `json:"fileSize"`
+	Checksum            string     `json:"checksum"`
+	StoragePath         string     `json:"storagePath"`
+	StorageTransferMode string     `json:"storageTransferMode,omitempty"`
+	DurationSeconds     int        `json:"durationSeconds"`
+	ErrorMessage        string     `json:"errorMessage"`
+	StartedAt           time.Time  `json:"startedAt"`
+	CompletedAt         *time.Time `json:"completedAt,omitempty"`
+	Locked              bool       `json:"locked"`
+	BackupKind          string     `json:"backupKind"`
 }

 type BackupRecordDetail struct {
@@ -184,22 +185,23 @@ func (s *BackupRecordService) SetLock(ctx context.Context, id uint, locked bool)

 func toBackupRecordSummary(item *model.BackupRecord) BackupRecordSummary {
 	return BackupRecordSummary{
-		ID:                item.ID,
-		TaskID:            item.TaskID,
-		TaskName:          item.Task.Name,
-		StorageTargetID:   item.StorageTargetID,
-		StorageTargetName: item.StorageTarget.Name,
-		Status:            item.Status,
-		FileName:          item.FileName,
-		FileSize:          item.FileSize,
-		Checksum:          item.Checksum,
-		StoragePath:       item.StoragePath,
-		DurationSeconds:   item.DurationSeconds,
-		ErrorMessage:      item.ErrorMessage,
-		StartedAt:         item.StartedAt,
-		CompletedAt:       item.CompletedAt,
-		Locked:            item.Locked,
-		BackupKind:        item.BackupKind,
+		ID:                  item.ID,
+		TaskID:              item.TaskID,
+		TaskName:            item.Task.Name,
+		StorageTargetID:     item.StorageTargetID,
+		StorageTargetName:   item.StorageTarget.Name,
+		Status:              item.Status,
+		FileName:            item.FileName,
+		FileSize:            item.FileSize,
+		Checksum:            item.Checksum,
+		StoragePath:         item.StoragePath,
+		StorageTransferMode: item.StorageTransferMode,
+		DurationSeconds:     item.DurationSeconds,
+		ErrorMessage:        item.ErrorMessage,
+		StartedAt:           item.StartedAt,
+		CompletedAt:         item.CompletedAt,
+		Locked:              item.Locked,
+		BackupKind:          item.BackupKind,
 	}
 }

@@ -57,8 +57,8 @@ type BackupTaskUpsertInput struct {
 	KeepWeekly  int `json:"keepWeekly"`
 	KeepMonthly int `json:"keepMonthly"`
 	KeepYearly  int `json:"keepYearly"`
-	// BackupMode 备份模式：full（默认）/ differential（差异，仅文件类型本机任务）
-	BackupMode           string `json:"backupMode" binding:"omitempty,oneof=full differential"`
+	// BackupMode 备份模式：full（默认）/ differential（差异归档）/ repository（CDC 去重仓库）
+	BackupMode           string `json:"backupMode" binding:"omitempty,oneof=full differential repository"`
 	DiffFullIntervalDays int    `json:"diffFullIntervalDays"`
 	// 备份复制目标存储 ID 列表（3-2-1 规则）
 	ReplicationTargetIDs []uint `json:"replicationTargetIds"`
@@ -414,21 +414,50 @@ func (s *BackupTaskService) cleanupRemoteFiles(ctx context.Context, taskID uint)
 	recordCount = len(records)
 	// 缓存 provider 避免同一存储目标重复创建连接
 	providerCache := make(map[uint]storage.StorageProvider)
+	repositoryProviders := make(map[uint]storage.StorageProvider)
 	for _, record := range records {
-		if strings.TrimSpace(record.StoragePath) == "" {
-			continue
+		copies := []StorageUploadResultItem{{
+			StorageTargetID: record.StorageTargetID,
+			Status:          model.BackupRecordStatusSuccess,
+			StoragePath:     record.StoragePath,
+		}}
+		if strings.TrimSpace(record.StorageUploadResults) != "" {
+			var storedCopies []StorageUploadResultItem
+			if unmarshalErr := json.Unmarshal([]byte(record.StorageUploadResults), &storedCopies); unmarshalErr == nil {
+				copies = storedCopies
+			}
 		}
-		provider, ok := providerCache[record.StorageTargetID]
-		if !ok {
-			provider, err = s.resolveStorageProvider(ctx, record.StorageTargetID)
-			if err != nil {
+		seenTargets := make(map[uint]struct{}, len(copies))
+		for _, copy := range copies {
+			if !strings.EqualFold(copy.Status, model.BackupRecordStatusSuccess) || strings.TrimSpace(copy.StoragePath) == "" {
 				continue
 			}
-			providerCache[record.StorageTargetID] = provider
+			if _, seen := seenTargets[copy.StorageTargetID]; seen {
+				continue
+			}
+			seenTargets[copy.StorageTargetID] = struct{}{}
+			provider, ok := providerCache[copy.StorageTargetID]
+			if !ok {
+				provider, err = s.resolveStorageProvider(ctx, copy.StorageTargetID)
+				if err != nil {
+					continue
+				}
+				providerCache[copy.StorageTargetID] = provider
+			}
+			if err := provider.Delete(ctx, copy.StoragePath); err == nil {
+				cleanedFiles++
+				if record.BackupKind == model.BackupKindRepository {
+					repositoryProviders[copy.StorageTargetID] = provider
+				}
+			}
 		}
-		if err := provider.Delete(ctx, record.StoragePath); err == nil {
-			cleanedFiles++
+	}
+	for _, provider := range repositoryProviders {
+		pruned, pruneErr := backup.NewRepositoryStore(s.cipher.Key()).Prune(ctx, provider)
+		if pruneErr != nil {
+			continue
 		}
+		cleanedFiles += pruned.DeletedIndexes + pruned.DeletedPacks
 	}
 	return recordCount, cleanedFiles
 }
@@ -530,6 +559,17 @@ func (s *BackupTaskService) validateInput(ctx context.Context, existing *model.B
 			return apperror.BadRequest("BACKUP_TASK_DIFF_REMOTE_UNSUPPORTED", "差异备份当前仅支持本机 Master 执行，请将任务固定在本机或改用全量备份。", nil)
 		}
 	}
+	if strings.EqualFold(strings.TrimSpace(input.BackupMode), model.BackupModeRepository) {
+		if input.Type != model.BackupTaskTypeFile {
+			return apperror.BadRequest("BACKUP_TASK_REPOSITORY_UNSUPPORTED", "CDC 仓库模式仅支持文件目录类型任务", nil)
+		}
+		if strings.TrimSpace(input.NodePoolTag) != "" || (fixedNode != nil && !fixedNode.IsLocal) {
+			return apperror.BadRequest("BACKUP_TASK_REPOSITORY_REMOTE_UNSUPPORTED", "CDC 仓库模式当前采用单写者索引，仅支持 Master 本机执行。远程服务器备份请暂用全量模式。", nil)
+		}
+		if len(input.ReplicationTargetIDs) > 0 {
+			return apperror.BadRequest("BACKUP_TASK_REPOSITORY_REPLICATION_UNSUPPORTED", "CDC 仓库快照不能使用对象级复制；请直接为任务选择多个存储目标以生成完整仓库副本。", nil)
+		}
+	}
 	if input.RetentionDays < 0 {
 		return apperror.BadRequest("BACKUP_TASK_INVALID", "保留天数不能小于 0", nil)
 	}
@@ -935,10 +975,17 @@ func decodeExtraConfig(value string) (map[string]any, error) {
 	return result, nil
 }

-// normalizeBackupMode 归一化备份模式：仅文件类型可启用差异，其余一律全量（双保险，防绕过校验）。
+// normalizeBackupMode 归一化备份模式：仅文件类型可启用差异或 CDC 仓库，
+// 其余一律全量（双保险，防绕过校验）。
 func normalizeBackupMode(mode, taskType string) string {
-	if strings.EqualFold(strings.TrimSpace(mode), model.BackupModeDifferential) && normalizeBackupTaskType(taskType) == model.BackupTaskTypeFile {
+	if normalizeBackupTaskType(taskType) != model.BackupTaskTypeFile {
+		return model.BackupModeFull
+	}
+	switch strings.ToLower(strings.TrimSpace(mode)) {
+	case model.BackupModeDifferential:
 		return model.BackupModeDifferential
+	case model.BackupModeRepository:
+		return model.BackupModeRepository
 	}
 	return model.BackupModeFull
 }
@@ -126,14 +126,14 @@ func (s *DashboardService) Timeline(ctx context.Context, days int) ([]repository
 // 判定规则：任务设置了 SLAHoursRPO > 0，且距最近一次 success 备份的时间 > SLAHoursRPO。
 // 从未成功过的任务（LastSuccessAt = nil）若启用也视为违约（from createdAt 起算）。
 type SLAViolation struct {
-	TaskID                 uint       `json:"taskId"`
-	TaskName               string     `json:"taskName"`
-	NodeID                 uint       `json:"nodeId"`
-	NodeName               string     `json:"nodeName,omitempty"`
-	SLAHoursRPO            int        `json:"slaHoursRpo"`
-	LastSuccessAt          *time.Time `json:"lastSuccessAt,omitempty"`
-	HoursSinceLastSuccess  float64    `json:"hoursSinceLastSuccess"`
-	NeverSucceeded         bool       `json:"neverSucceeded"`
+	TaskID                uint       `json:"taskId"`
+	TaskName              string     `json:"taskName"`
+	NodeID                uint       `json:"nodeId"`
+	NodeName              string     `json:"nodeName,omitempty"`
+	SLAHoursRPO           int        `json:"slaHoursRpo"`
+	LastSuccessAt         *time.Time `json:"lastSuccessAt,omitempty"`
+	HoursSinceLastSuccess float64    `json:"hoursSinceLastSuccess"`
+	NeverSucceeded        bool       `json:"neverSucceeded"`
 }

 // SLAComplianceReport Dashboard 的 SLA 合规概览。
@@ -204,15 +204,15 @@ func roundHours(value float64) float64 {

 // ClusterNodeSummary 集群节点简报（Dashboard 用）。
 type ClusterNodeSummary struct {
-	ID             uint      `json:"id"`
-	Name           string    `json:"name"`
-	Hostname       string    `json:"hostname"`
-	Status         string    `json:"status"`
-	IsLocal        bool      `json:"isLocal"`
-	AgentVersion   string    `json:"agentVersion"`
-	VersionStatus  string    `json:"versionStatus"` // current | outdated | unknown
-	LastSeen       time.Time `json:"lastSeen"`
-	TaskCount      int64     `json:"taskCount"`
+	ID            uint      `json:"id"`
+	Name          string    `json:"name"`
+	Hostname      string    `json:"hostname"`
+	Status        string    `json:"status"`
+	IsLocal       bool      `json:"isLocal"`
+	AgentVersion  string    `json:"agentVersion"`
+	VersionStatus string    `json:"versionStatus"` // current | outdated | unknown
+	LastSeen      time.Time `json:"lastSeen"`
+	TaskCount     int64     `json:"taskCount"`
 }

 // ClusterOverview Dashboard 集群概览卡片。
@@ -312,9 +312,9 @@ func (s *DashboardService) Breakdown(ctx context.Context, days int) (*BreakdownS
 		}
 	}
 	result := &BreakdownStats{
-		ByType:   makeBreakdown(typeCounts, typeLabel),
-		ByNode:   makeBreakdownByUint(nodeCounts, nodeNames, "节点 #"),
-		ByStatus: []BreakdownItem{},
+		ByType:    makeBreakdown(typeCounts, typeLabel),
+		ByNode:    makeBreakdownByUint(nodeCounts, nodeNames, "节点 #"),
+		ByStatus:  []BreakdownItem{},
 		ByStorage: []BreakdownItem{},
 	}
 	// 按状态（最近 days 天记录）
@@ -140,6 +140,11 @@ func validateCrossNodeLocalDisk(ctx context.Context, nodeRepo repository.NodeRep
 	if record == nil || record.NodeID == 0 || nodeRepo == nil {
 		return nil
 	}
+	// 中转模式的对象实际落在 Master 配置的本地磁盘，Master 可以安全访问。
+	// 空值和 direct 均按旧版 Agent 本地落盘处理，保持升级兼容。
+	if record.StorageTransferMode == storage.TransferModeMasterRelay {
+		return nil
+	}
 	node, err := nodeRepo.FindByID(ctx, record.NodeID)
 	if err != nil || node == nil || node.IsLocal {
 		return nil
@@ -6,6 +6,7 @@ import (
 	"encoding/base64"
 	"encoding/hex"
 	"fmt"
+	"net/url"
 	"strings"
 	"time"

@@ -27,13 +28,16 @@ func NewInstallTokenService(repo repository.AgentInstallTokenRepository, nodeRep

 // InstallTokenInput 生成一次性安装令牌的输入。
 type InstallTokenInput struct {
-	NodeID       uint
-	Mode         string
-	Arch         string
-	AgentVersion string
-	DownloadSrc  string
-	TTLSeconds   int
-	CreatedByID  uint
+	NodeID         uint
+	Mode           string
+	Arch           string
+	AgentVersion   string
+	DownloadSrc    string
+	TTLSeconds     int
+	CreatedByID    uint
+	AgentMasterURL string
+	ProxyURL       string
+	CACertFile     string
 }

 // InstallTokenOutput 生成结果。
@@ -112,14 +116,17 @@ func (s *InstallTokenService) Create(ctx context.Context, in InstallTokenInput)
 	}
 	expiresAt := time.Now().UTC().Add(time.Duration(in.TTLSeconds) * time.Second)
 	record := &model.AgentInstallToken{
-		Token:       token,
-		NodeID:      in.NodeID,
-		Mode:        in.Mode,
-		Arch:        in.Arch,
-		AgentVer:    in.AgentVersion,
-		DownloadSrc: in.DownloadSrc,
-		ExpiresAt:   expiresAt,
-		CreatedByID: in.CreatedByID,
+		Token:          token,
+		NodeID:         in.NodeID,
+		Mode:           in.Mode,
+		Arch:           in.Arch,
+		AgentVer:       in.AgentVersion,
+		DownloadSrc:    in.DownloadSrc,
+		AgentMasterURL: strings.TrimRight(strings.TrimSpace(in.AgentMasterURL), "/"),
+		ProxyURL:       strings.TrimSpace(in.ProxyURL),
+		CACertFile:     strings.TrimSpace(in.CACertFile),
+		ExpiresAt:      expiresAt,
+		CreatedByID:    in.CreatedByID,
 	}
 	if err := s.repo.Create(ctx, record); err != nil {
 		return nil, err
@@ -130,9 +137,15 @@ func (s *InstallTokenService) Create(ctx context.Context, in InstallTokenInput)
 // CreateCommand 创建 install token，并返回 UI 展示安装命令所需的 URL 与嵌入式脚本。
 func (s *InstallTokenService) CreateCommand(ctx context.Context, in InstallCommandInput) (*InstallCommandOutput, error) {
 	masterURL := strings.TrimRight(strings.TrimSpace(in.MasterURL), "/")
-	if masterURL == "" {
-		return nil, apperror.BadRequest("INSTALL_TOKEN_INVALID", "masterURL 必填", nil)
+	deliveryURL, parseErr := url.Parse(masterURL)
+	if masterURL == "" || parseErr != nil || (deliveryURL.Scheme != "http" && deliveryURL.Scheme != "https") || deliveryURL.Host == "" || deliveryURL.User != nil || deliveryURL.RawQuery != "" || deliveryURL.Fragment != "" || strings.ContainsAny(masterURL, " \t\r\n\"'`$\\") {
+		return nil, apperror.BadRequest("INSTALL_TOKEN_INVALID", "masterURL 必须是安全的完整 HTTP(S) 地址", parseErr)
 	}
+	agentMasterURL := strings.TrimRight(strings.TrimSpace(in.AgentMasterURL), "/")
+	if agentMasterURL == "" {
+		agentMasterURL = masterURL
+	}
+	in.AgentMasterURL = agentMasterURL
 	if err := s.validate(in.InstallTokenInput); err != nil {
 		return nil, err
 	}
@@ -144,12 +157,15 @@ func (s *InstallTokenService) CreateCommand(ctx context.Context, in InstallComma
 		return nil, apperror.New(404, "NODE_NOT_FOUND", "节点不存在", nil)
 	}
 	if _, err := renderInstallCommandScript(masterURL, node, &model.AgentInstallToken{
-		Mode:        in.Mode,
-		Arch:        in.Arch,
-		AgentVer:    in.AgentVersion,
-		DownloadSrc: in.DownloadSrc,
+		Mode:           in.Mode,
+		Arch:           in.Arch,
+		AgentVer:       in.AgentVersion,
+		DownloadSrc:    in.DownloadSrc,
+		AgentMasterURL: in.AgentMasterURL,
+		ProxyURL:       in.ProxyURL,
+		CACertFile:     in.CACertFile,
 	}); err != nil {
-		return nil, err
+		return nil, apperror.BadRequest("INSTALL_TOKEN_INVALID", "Agent 连接配置无效", err)
 	}
 	out, err := s.Create(ctx, in.InstallTokenInput)
 	if err != nil {
@@ -164,20 +180,24 @@ func (s *InstallTokenService) CreateCommand(ctx context.Context, in InstallComma
 		ExpiresAt:    out.ExpiresAt,
 		Node:         out.Node,
 		Record:       out.Record,
-		URL:          masterURL + "/api/install/" + out.Token,
-		FallbackURL:  masterURL + "/install/" + out.Token,
+		URL:          agentMasterURL + "/api/install/" + out.Token,
+		FallbackURL:  agentMasterURL + "/install/" + out.Token,
 		ScriptBase64: base64.StdEncoding.EncodeToString([]byte(script)),
 	}
 	if out.Record.Mode == model.InstallModeDocker {
-		result.ComposeURL = masterURL + "/api/install/" + out.Token + "/compose.yml"
-		result.FallbackComposeURL = masterURL + "/install/" + out.Token + "/compose.yml"
+		result.ComposeURL = agentMasterURL + "/api/install/" + out.Token + "/compose.yml"
+		result.FallbackComposeURL = agentMasterURL + "/install/" + out.Token + "/compose.yml"
 	}
 	return result, nil
 }

 func renderInstallCommandScript(masterURL string, node *model.Node, record *model.AgentInstallToken) (string, error) {
+	agentMasterURL := strings.TrimRight(strings.TrimSpace(record.AgentMasterURL), "/")
+	if agentMasterURL == "" {
+		agentMasterURL = masterURL
+	}
 	return installscript.RenderScript(installscript.Context{
-		MasterURL:     masterURL,
+		MasterURL:     agentMasterURL,
 		AgentToken:    node.Token,
 		AgentVersion:  record.AgentVer,
 		Mode:          record.Mode,
@@ -185,6 +205,8 @@ func renderInstallCommandScript(masterURL string, node *model.Node, record *mode
 		DownloadBase:  installscript.DownloadBaseFor(record.DownloadSrc),
 		InstallPrefix: "/opt/backupx-agent",
 		NodeID:        node.ID,
+		ProxyURL:      record.ProxyURL,
+		CACertFile:    record.CACertFile,
 	})
 }

@@ -259,6 +281,9 @@ func (s *InstallTokenService) validate(in InstallTokenInput) error {
 		return apperror.BadRequest("INSTALL_TOKEN_INVALID",
 			fmt.Sprintf("ttlSeconds 需在 %d-%d", InstallTokenMinTTL, InstallTokenMaxTTL), nil)
 	}
+	if len(in.AgentMasterURL) > 2048 || len(in.ProxyURL) > 2048 || len(in.CACertFile) > 512 {
+		return apperror.BadRequest("INSTALL_TOKEN_INVALID", "连接配置过长", nil)
+	}
 	return nil
 }

@@ -2,7 +2,9 @@ package service

 import (
 	"context"
+	"encoding/base64"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"

@@ -177,13 +179,16 @@ func TestInstallTokenServiceCreateCommandBuildsURLsAndScript(t *testing.T) {

 	out, err := svc.CreateCommand(context.Background(), InstallCommandInput{
 		InstallTokenInput: InstallTokenInput{
-			NodeID:       node.ID,
-			Mode:         model.InstallModeDocker,
-			Arch:         model.InstallArchAuto,
-			AgentVersion: "v1.7.0",
-			DownloadSrc:  model.InstallSourceGitHub,
-			TTLSeconds:   900,
-			CreatedByID:  1,
+			NodeID:         node.ID,
+			Mode:           model.InstallModeDocker,
+			Arch:           model.InstallArchAuto,
+			AgentVersion:   "v1.7.0",
+			DownloadSrc:    model.InstallSourceGitHub,
+			TTLSeconds:     900,
+			CreatedByID:    1,
+			AgentMasterURL: "http://127.0.0.1:18340",
+			ProxyURL:       "socks5h://127.0.0.1:1080",
+			CACertFile:     "/etc/pki/internal-ca.pem",
 		},
 		MasterURL: "https://public.example.com/base",
 	})
@@ -193,15 +198,66 @@ func TestInstallTokenServiceCreateCommandBuildsURLsAndScript(t *testing.T) {
 	if out.Token == "" || out.ScriptBase64 == "" {
 		t.Fatalf("missing token or script: %+v", out)
 	}
-	if out.URL != "https://public.example.com/base/api/install/"+out.Token {
+	if out.URL != "http://127.0.0.1:18340/api/install/"+out.Token {
 		t.Fatalf("bad url: %s", out.URL)
 	}
-	if out.FallbackURL != "https://public.example.com/base/install/"+out.Token {
+	if out.FallbackURL != "http://127.0.0.1:18340/install/"+out.Token {
 		t.Fatalf("bad fallback url: %s", out.FallbackURL)
 	}
-	if out.ComposeURL != "https://public.example.com/base/api/install/"+out.Token+"/compose.yml" {
+	if out.ComposeURL != "http://127.0.0.1:18340/api/install/"+out.Token+"/compose.yml" {
 		t.Fatalf("bad compose url: %s", out.ComposeURL)
 	}
+	script, err := base64.StdEncoding.DecodeString(out.ScriptBase64)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, want := range []string{
+		`MASTER_URL="http://127.0.0.1:18340"`,
+		`PROXY_URL="socks5h://127.0.0.1:1080"`,
+		`CA_CERT_FILE="/etc/pki/internal-ca.pem"`,
+	} {
+		if !strings.Contains(string(script), want) {
+			t.Fatalf("script missing %q", want)
+		}
+	}
+}
+
+func TestInstallTokenServiceRejectsUnsafeConnectionBeforeCreate(t *testing.T) {
+	db := openInstallTokenTestDB(t)
+	nodeRepo := repository.NewNodeRepository(db)
+	node := &model.Node{Name: "restricted", Token: "deadbeefcafebabe0123456789abcdef0123456789abcdef0123456789abcdef"}
+	if err := nodeRepo.Create(context.Background(), node); err != nil {
+		t.Fatal(err)
+	}
+	tokenRepo := repository.NewAgentInstallTokenRepository(db)
+	svc := NewInstallTokenService(tokenRepo, nodeRepo)
+	_, err := svc.CreateCommand(context.Background(), InstallCommandInput{
+		InstallTokenInput: InstallTokenInput{
+			NodeID: node.ID, Mode: model.InstallModeSystemd, Arch: model.InstallArchAuto,
+			AgentVersion: "v2.4.0", DownloadSrc: model.InstallSourceGitHub, TTLSeconds: 900,
+			ProxyURL: "http://user:pass@proxy.example.com",
+		},
+		MasterURL: "https://public.example.com",
+	})
+	if err == nil {
+		t.Fatal("expected proxy credentials to be rejected")
+	}
+	count, countErr := tokenRepo.CountCreatedSince(context.Background(), node.ID, time.Now().UTC().Add(-time.Hour))
+	if countErr != nil || count != 0 {
+		t.Fatalf("invalid connection created token records: count=%d err=%v", count, countErr)
+	}
+
+	_, err = svc.CreateCommand(context.Background(), InstallCommandInput{
+		InstallTokenInput: InstallTokenInput{
+			NodeID: node.ID, Mode: model.InstallModeSystemd, Arch: model.InstallArchAuto,
+			AgentVersion: "v2.4.0", DownloadSrc: model.InstallSourceGitHub, TTLSeconds: 900,
+			AgentMasterURL: "https://master.internal",
+		},
+		MasterURL: "http://public.example.com/$unsafe",
+	})
+	if err == nil {
+		t.Fatal("expected unsafe public delivery URL to be rejected")
+	}
 }

 func TestInstallTokenServiceRateLimit(t *testing.T) {
@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 	"strings"
@@ -321,6 +322,13 @@ func (s *RestoreService) restoreArtifact(ctx context.Context, record *model.Back
 	if err != nil {
 		return fmt.Errorf("创建存储客户端失败：%w", err)
 	}
+	if record.BackupKind == model.BackupKindRepository {
+		logger.Infof("读取 CDC 仓库快照：%s", record.StoragePath)
+		if err := backup.NewRepositoryStore(s.cipher.Key()).Restore(ctx, provider, record.StoragePath, record.Checksum, spec, logger); err != nil {
+			return fmt.Errorf("恢复 CDC 仓库快照失败：%w", err)
+		}
+		return nil
+	}
 	recDir, err := os.MkdirTemp(parentTempDir, fmt.Sprintf("rec-%d-*", record.ID))
 	if err != nil {
 		return fmt.Errorf("创建恢复子目录失败：%w", err)
@@ -368,6 +376,9 @@ func (s *RestoreService) buildRestoreChain(ctx context.Context, record *model.Ba
 }

 func backupKindLabel(kind string) string {
+	if kind == model.BackupKindRepository {
+		return "CDC 仓库快照"
+	}
 	if kind == model.BackupKindDifferential {
 		return "差异"
 	}
@@ -591,15 +602,22 @@ func (s *RestoreService) GetAgentRestoreSpec(ctx context.Context, node *model.No
 	if target == nil {
 		return nil, apperror.BadRequest("BACKUP_STORAGE_TARGET_INVALID", "存储目标不存在", nil)
 	}
-	configRaw, err := s.cipher.Decrypt(target.ConfigCiphertext)
-	if err != nil {
-		return nil, fmt.Errorf("decrypt storage config: %w", err)
-	}
 	// 拆开 sourcePaths
 	sourcePaths := []string{}
 	if strings.TrimSpace(task.SourcePaths) != "" {
 		_ = json.Unmarshal([]byte(task.SourcePaths), &sourcePaths)
 	}
+	transferMode := storage.TransferModeDirect
+	if backupRecord.StorageTransferMode == storage.TransferModeMasterRelay {
+		transferMode = storage.TransferModeMasterRelay
+	}
+	var configRaw []byte
+	if transferMode == storage.TransferModeDirect {
+		configRaw, err = s.cipher.Decrypt(target.ConfigCiphertext)
+		if err != nil {
+			return nil, fmt.Errorf("decrypt storage config: %w", err)
+		}
+	}
 	return &AgentRestoreSpec{
 		RestoreRecordID: restore.ID,
 		BackupRecordID:  backupRecord.ID,
@@ -618,10 +636,11 @@ func (s *RestoreService) GetAgentRestoreSpec(ctx context.Context, node *model.No
 		Compression:     task.Compression,
 		Encrypt:         task.Encrypt,
 		Storage: AgentStorageTargetConfig{
-			ID:     target.ID,
-			Type:   target.Type,
-			Name:   target.Name,
-			Config: json.RawMessage(configRaw),
+			ID:           target.ID,
+			Type:         target.Type,
+			Name:         target.Name,
+			Config:       json.RawMessage(configRaw),
+			TransferMode: transferMode,
 		},
 		StoragePath: backupRecord.StoragePath,
 		FileName:    backupRecord.FileName,
@@ -629,6 +648,63 @@ func (s *RestoreService) GetAgentRestoreSpec(ctx context.Context, node *model.No
 	}, nil
 }

+type AgentArtifactDownload struct {
+	Reader io.ReadCloser
+	Size   int64
+}
+
+// DownloadAgentArtifact opens a Master-local object for authenticated streaming
+// back to the Agent that owns the restore record.
+func (s *RestoreService) DownloadAgentArtifact(ctx context.Context, node *model.Node, restoreID uint) (*AgentArtifactDownload, error) {
+	if node == nil {
+		return nil, apperror.Unauthorized("RESTORE_RECORD_FORBIDDEN", "恢复记录不属于当前节点", nil)
+	}
+	restore, err := s.restores.FindByID(ctx, restoreID)
+	if err != nil {
+		return nil, err
+	}
+	if restore == nil {
+		return nil, apperror.New(404, "RESTORE_RECORD_NOT_FOUND", "恢复记录不存在", nil)
+	}
+	if restore.NodeID != node.ID {
+		return nil, apperror.Unauthorized("RESTORE_RECORD_FORBIDDEN", "恢复记录不属于当前节点", nil)
+	}
+	if isRestoreRecordTerminal(restore.Status) {
+		return nil, apperror.BadRequest("RESTORE_RECORD_TERMINAL", "恢复记录已结束，不能继续下载产物", nil)
+	}
+	record, err := s.records.FindByID(ctx, restore.BackupRecordID)
+	if err != nil {
+		return nil, err
+	}
+	if record == nil {
+		return nil, apperror.New(404, "BACKUP_RECORD_NOT_FOUND", "源备份记录不存在", nil)
+	}
+	target, err := s.targets.FindByID(ctx, record.StorageTargetID)
+	if err != nil {
+		return nil, err
+	}
+	if target == nil || !strings.EqualFold(target.Type, storage.TypeLocalDisk) || record.StorageTransferMode != storage.TransferModeMasterRelay {
+		return nil, apperror.BadRequest("AGENT_ARTIFACT_RELAY_UNSUPPORTED", "该存储目标应由 Agent 直接下载", nil)
+	}
+	configMap := map[string]any{}
+	if err := s.cipher.DecryptJSON(target.ConfigCiphertext, &configMap); err != nil {
+		return nil, fmt.Errorf("decrypt storage config: %w", err)
+	}
+	provider, err := s.storageRegistry.Create(ctx, target.Type, configMap)
+	if err != nil {
+		return nil, fmt.Errorf("create master relay provider: %w", err)
+	}
+	reader, err := provider.Download(ctx, record.StoragePath)
+	if err != nil {
+		return nil, fmt.Errorf("open master relay artifact: %w", err)
+	}
+	size := record.FileSize
+	if size <= 0 {
+		size = -1
+	}
+	return &AgentArtifactDownload{Reader: reader, Size: size}, nil
+}
+
 // UpdateAgentRestore Agent 回传状态/日志。
 func (s *RestoreService) UpdateAgentRestore(ctx context.Context, node *model.Node, restoreID uint, update AgentRestoreUpdate) error {
 	restore, err := s.restores.FindByID(ctx, restoreID)
@@ -1,8 +1,10 @@
 package service

 import (
+	"bytes"
 	"context"
 	"encoding/json"
+	"io"
 	"os"
 	"path/filepath"
 	"strings"
@@ -427,16 +429,27 @@ func TestRestoreServiceAgentRestoreAccessUsesRestoreRecordNode(t *testing.T) {
 	}
 	startedAt := time.Now().UTC()
 	completedAt := startedAt.Add(time.Second)
+	artifact := []byte("central backup artifact")
+	storagePath := "file/2026/05/09/remote.tar.gz"
+	artifactPath := filepath.Join(h.storageDir, filepath.FromSlash(storagePath))
+	if err := os.MkdirAll(filepath.Dir(artifactPath), 0o755); err != nil {
+		t.Fatalf("MkdirAll artifact parent: %v", err)
+	}
+	if err := os.WriteFile(artifactPath, artifact, 0o600); err != nil {
+		t.Fatalf("WriteFile artifact: %v", err)
+	}
 	backupRecord := &model.BackupRecord{
-		TaskID:          task.ID,
-		StorageTargetID: task.StorageTargetID,
-		NodeID:          owner.ID,
-		Status:          model.BackupRecordStatusSuccess,
-		FileName:        "remote.tar.gz",
-		StoragePath:     "file/2026/05/09/remote.tar.gz",
-		Checksum:        "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
-		StartedAt:       startedAt,
-		CompletedAt:     &completedAt,
+		TaskID:              task.ID,
+		StorageTargetID:     task.StorageTargetID,
+		NodeID:              owner.ID,
+		Status:              model.BackupRecordStatusSuccess,
+		FileName:            "remote.tar.gz",
+		StoragePath:         storagePath,
+		FileSize:            int64(len(artifact)),
+		StorageTransferMode: storage.TransferModeMasterRelay,
+		Checksum:            "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
+		StartedAt:           startedAt,
+		CompletedAt:         &completedAt,
 	}
 	if err := h.records.Create(ctx, backupRecord); err != nil {
 		t.Fatalf("Create backup record: %v", err)
@@ -464,6 +477,21 @@ func TestRestoreServiceAgentRestoreAccessUsesRestoreRecordNode(t *testing.T) {
 	if spec.Checksum != backupRecord.Checksum {
 		t.Fatalf("expected spec.Checksum=%q, got %q", backupRecord.Checksum, spec.Checksum)
 	}
+	if spec.Storage.TransferMode != storage.TransferModeMasterRelay {
+		t.Fatalf("expected Master relay restore, got %#v", spec.Storage)
+	}
+	download, err := h.service.DownloadAgentArtifact(ctx, owner, restore.ID)
+	if err != nil {
+		t.Fatalf("DownloadAgentArtifact returned error: %v", err)
+	}
+	downloaded, readErr := io.ReadAll(download.Reader)
+	closeErr := download.Reader.Close()
+	if readErr != nil || closeErr != nil {
+		t.Fatalf("read relayed restore artifact: read=%v close=%v", readErr, closeErr)
+	}
+	if !bytes.Equal(downloaded, artifact) {
+		t.Fatalf("relayed restore artifact differs: %q", downloaded)
+	}
 	if _, err := h.service.GetAgentRestoreSpec(ctx, other, restore.ID); err == nil {
 		t.Fatal("expected non-owner node to be forbidden from restore spec")
 	}
@@ -14,9 +14,9 @@ import (

 // TaskExportService 管理备份任务的 JSON 导入 / 导出。
 // 用途：
-//   1. 集群迁移（旧 Master → 新 Master 的任务配置搬迁）
-//   2. 灾备恢复（任务配置本地文件化，Master 宕机后重建）
-//   3. 配置审计（版本化 Git 管理 JSON 快照）
+//  1. 集群迁移（旧 Master → 新 Master 的任务配置搬迁）
+//  2. 灾备恢复（任务配置本地文件化，Master 宕机后重建）
+//  3. 配置审计（版本化 Git 管理 JSON 快照）
 //
 // 出于安全考虑，导出/导入不包含任何敏感字段：
 //   - 数据库密码（DBPasswordCiphertext）：跳过，导入后需人工填补
@@ -40,35 +40,35 @@ func NewTaskExportService(

 // ExportedTask 导出格式：按名称引用存储/节点，不含敏感数据。
 type ExportedTask struct {
-	Name               string         `json:"name"`
-	Type               string         `json:"type"`
-	Enabled            bool           `json:"enabled"`
-	CronExpr           string         `json:"cronExpr,omitempty"`
-	SourcePath         string         `json:"sourcePath,omitempty"`
-	SourcePaths        []string       `json:"sourcePaths,omitempty"`
-	ExcludePatterns    []string       `json:"excludePatterns,omitempty"`
-	DBHost             string         `json:"dbHost,omitempty"`
-	DBPort             int            `json:"dbPort,omitempty"`
-	DBUser             string         `json:"dbUser,omitempty"`
-	DBName             string         `json:"dbName,omitempty"`
-	DBPath             string         `json:"dbPath,omitempty"`
-	ExtraConfig        map[string]any `json:"extraConfig,omitempty"`
+	Name            string         `json:"name"`
+	Type            string         `json:"type"`
+	Enabled         bool           `json:"enabled"`
+	CronExpr        string         `json:"cronExpr,omitempty"`
+	SourcePath      string         `json:"sourcePath,omitempty"`
+	SourcePaths     []string       `json:"sourcePaths,omitempty"`
+	ExcludePatterns []string       `json:"excludePatterns,omitempty"`
+	DBHost          string         `json:"dbHost,omitempty"`
+	DBPort          int            `json:"dbPort,omitempty"`
+	DBUser          string         `json:"dbUser,omitempty"`
+	DBName          string         `json:"dbName,omitempty"`
+	DBPath          string         `json:"dbPath,omitempty"`
+	ExtraConfig     map[string]any `json:"extraConfig,omitempty"`
 	// 按名称引用：导入时按名称查找对应 ID
-	StorageTargetNames    []string `json:"storageTargetNames"`
-	ReplicationTargetNames []string `json:"replicationTargetNames,omitempty"`
-	NodeName              string   `json:"nodeName,omitempty"`
-	DependsOnTaskNames    []string `json:"dependsOnTaskNames,omitempty"`
-	Tags                  string   `json:"tags,omitempty"`
-	Compression           string   `json:"compression,omitempty"`
-	Encrypt               bool     `json:"encrypt,omitempty"`
-	RetentionDays         int      `json:"retentionDays,omitempty"`
-	MaxBackups            int      `json:"maxBackups,omitempty"`
-	VerifyEnabled         bool     `json:"verifyEnabled,omitempty"`
-	VerifyCronExpr        string   `json:"verifyCronExpr,omitempty"`
-	VerifyMode            string   `json:"verifyMode,omitempty"`
-	SLAHoursRPO           int      `json:"slaHoursRpo,omitempty"`
-	AlertOnConsecutiveFails int    `json:"alertOnConsecutiveFails,omitempty"`
-	MaintenanceWindows    string   `json:"maintenanceWindows,omitempty"`
+	StorageTargetNames      []string `json:"storageTargetNames"`
+	ReplicationTargetNames  []string `json:"replicationTargetNames,omitempty"`
+	NodeName                string   `json:"nodeName,omitempty"`
+	DependsOnTaskNames      []string `json:"dependsOnTaskNames,omitempty"`
+	Tags                    string   `json:"tags,omitempty"`
+	Compression             string   `json:"compression,omitempty"`
+	Encrypt                 bool     `json:"encrypt,omitempty"`
+	RetentionDays           int      `json:"retentionDays,omitempty"`
+	MaxBackups              int      `json:"maxBackups,omitempty"`
+	VerifyEnabled           bool     `json:"verifyEnabled,omitempty"`
+	VerifyCronExpr          string   `json:"verifyCronExpr,omitempty"`
+	VerifyMode              string   `json:"verifyMode,omitempty"`
+	SLAHoursRPO             int      `json:"slaHoursRpo,omitempty"`
+	AlertOnConsecutiveFails int      `json:"alertOnConsecutiveFails,omitempty"`
+	MaintenanceWindows      string   `json:"maintenanceWindows,omitempty"`
 }

 // ExportPayload 导出整体结构，带元信息。
@@ -233,67 +233,67 @@ func (s *TaskExportService) toExported(item *model.BackupTask, targetNames, node
 		nodeName = nodeNames[item.NodeID]
 	}
 	return ExportedTask{
-		Name:                   item.Name,
-		Type:                   item.Type,
-		Enabled:                item.Enabled,
-		CronExpr:               item.CronExpr,
-		SourcePath:             item.SourcePath,
-		SourcePaths:            sourcePaths,
-		ExcludePatterns:        excludes,
-		DBHost:                 item.DBHost,
-		DBPort:                 item.DBPort,
-		DBUser:                 item.DBUser,
-		DBName:                 item.DBName,
-		DBPath:                 item.DBPath,
-		ExtraConfig:            extra,
-		StorageTargetNames:     storageNames,
-		ReplicationTargetNames: replicationNames,
-		NodeName:               nodeName,
-		DependsOnTaskNames:     dependsOnNames,
-		Tags:                   item.Tags,
-		Compression:            item.Compression,
-		Encrypt:                item.Encrypt,
-		RetentionDays:          item.RetentionDays,
-		MaxBackups:             item.MaxBackups,
-		VerifyEnabled:          item.VerifyEnabled,
-		VerifyCronExpr:         item.VerifyCronExpr,
-		VerifyMode:             item.VerifyMode,
-		SLAHoursRPO:            item.SLAHoursRPO,
+		Name:                    item.Name,
+		Type:                    item.Type,
+		Enabled:                 item.Enabled,
+		CronExpr:                item.CronExpr,
+		SourcePath:              item.SourcePath,
+		SourcePaths:             sourcePaths,
+		ExcludePatterns:         excludes,
+		DBHost:                  item.DBHost,
+		DBPort:                  item.DBPort,
+		DBUser:                  item.DBUser,
+		DBName:                  item.DBName,
+		DBPath:                  item.DBPath,
+		ExtraConfig:             extra,
+		StorageTargetNames:      storageNames,
+		ReplicationTargetNames:  replicationNames,
+		NodeName:                nodeName,
+		DependsOnTaskNames:      dependsOnNames,
+		Tags:                    item.Tags,
+		Compression:             item.Compression,
+		Encrypt:                 item.Encrypt,
+		RetentionDays:           item.RetentionDays,
+		MaxBackups:              item.MaxBackups,
+		VerifyEnabled:           item.VerifyEnabled,
+		VerifyCronExpr:          item.VerifyCronExpr,
+		VerifyMode:              item.VerifyMode,
+		SLAHoursRPO:             item.SLAHoursRPO,
 		AlertOnConsecutiveFails: item.AlertOnConsecutiveFails,
-		MaintenanceWindows:     item.MaintenanceWindows,
+		MaintenanceWindows:      item.MaintenanceWindows,
 	}
 }

 func (s *TaskExportService) toUpsertInput(t ExportedTask, targetsByName, nodesByName map[string]uint, deps []uint) BackupTaskUpsertInput {
 	return BackupTaskUpsertInput{
-		Name:                   t.Name,
-		Type:                   t.Type,
-		Enabled:                t.Enabled,
-		CronExpr:               t.CronExpr,
-		SourcePath:             t.SourcePath,
-		SourcePaths:            t.SourcePaths,
-		ExcludePatterns:        t.ExcludePatterns,
-		DBHost:                 t.DBHost,
-		DBPort:                 t.DBPort,
-		DBUser:                 t.DBUser,
-		DBName:                 t.DBName,
-		DBPath:                 t.DBPath,
-		ExtraConfig:            t.ExtraConfig,
-		StorageTargetIDs:       idsFromNames(t.StorageTargetNames, targetsByName),
-		ReplicationTargetIDs:   idsFromNames(t.ReplicationTargetNames, targetsByName),
-		NodeID:                 nodesByName[t.NodeName],
-		Tags:                   t.Tags,
-		Compression:            t.Compression,
-		Encrypt:                t.Encrypt,
-		RetentionDays:          t.RetentionDays,
-		MaxBackups:             t.MaxBackups,
-		VerifyEnabled:          t.VerifyEnabled,
-		VerifyCronExpr:         t.VerifyCronExpr,
-		VerifyMode:             t.VerifyMode,
-		SLAHoursRPO:            t.SLAHoursRPO,
+		Name:                    t.Name,
+		Type:                    t.Type,
+		Enabled:                 t.Enabled,
+		CronExpr:                t.CronExpr,
+		SourcePath:              t.SourcePath,
+		SourcePaths:             t.SourcePaths,
+		ExcludePatterns:         t.ExcludePatterns,
+		DBHost:                  t.DBHost,
+		DBPort:                  t.DBPort,
+		DBUser:                  t.DBUser,
+		DBName:                  t.DBName,
+		DBPath:                  t.DBPath,
+		ExtraConfig:             t.ExtraConfig,
+		StorageTargetIDs:        idsFromNames(t.StorageTargetNames, targetsByName),
+		ReplicationTargetIDs:    idsFromNames(t.ReplicationTargetNames, targetsByName),
+		NodeID:                  nodesByName[t.NodeName],
+		Tags:                    t.Tags,
+		Compression:             t.Compression,
+		Encrypt:                 t.Encrypt,
+		RetentionDays:           t.RetentionDays,
+		MaxBackups:              t.MaxBackups,
+		VerifyEnabled:           t.VerifyEnabled,
+		VerifyCronExpr:          t.VerifyCronExpr,
+		VerifyMode:              t.VerifyMode,
+		SLAHoursRPO:             t.SLAHoursRPO,
 		AlertOnConsecutiveFails: t.AlertOnConsecutiveFails,
-		MaintenanceWindows:     t.MaintenanceWindows,
-		DependsOnTaskIDs:       deps,
+		MaintenanceWindows:      t.MaintenanceWindows,
+		DependsOnTaskIDs:        deps,
 	}
 }

@@ -299,6 +299,20 @@ func (s *VerificationService) executeLocally(ctx context.Context, verID uint, ta
 		logger.Errorf("创建存储客户端失败：%v", err)
 		return
 	}
+	if backupRecord.BackupKind == model.BackupKindRepository {
+		logger.Infof("验证 CDC 仓库快照及全部引用块：%s", backupRecord.StoragePath)
+		report, verifyErr := backup.NewRepositoryStore(s.cipher.Key()).Verify(ctx, provider, backupRecord.StoragePath, backupRecord.Checksum)
+		if verifyErr != nil {
+			errMessage = verifyErr.Error()
+			summary = "CDC 仓库完整性校验失败"
+			logger.Errorf("验证未通过：%v", verifyErr)
+			return
+		}
+		status = model.VerificationRecordStatusSuccess
+		summary = fmt.Sprintf("CDC 仓库完整性校验通过：%d 个条目、%d 个唯一块、%d bytes", report.Entries, report.Chunks, report.Bytes)
+		logger.Infof("%s", summary)
+		return
+	}
 	fileName := backupRecord.FileName
 	if strings.TrimSpace(fileName) == "" {
 		fileName = filepath.Base(backupRecord.StoragePath)