Add destination takeover config policy

internal/config/config.go

@@ -57,6 +57,7 @@ type Destination struct {
 	Links          *Links               `yaml:"links"`
 	State          StatePolicy          `yaml:"state"`
 	Reconciliation ReconciliationPolicy `yaml:"reconciliation"`
+	Takeover       TakeoverPolicy       `yaml:"takeover"`
 	Retention      RetentionPolicy      `yaml:"retention"`
 	Transfer       TransferPolicy       `yaml:"transfer"`
 }
@@ -126,6 +127,10 @@ type ReconciliationPolicy struct {
 	Mode string `yaml:"mode"`
 }
 
+type TakeoverPolicy struct {
+	Mode string `yaml:"mode"`
+}
+
 type StatePolicy struct {
 	Mode string `yaml:"mode"`
 }

internal/config/defaults.go

@@ -47,6 +47,13 @@ const (
 	ReconciliationModeMerge   = "merge"
 )
 
+const (
+	TakeoverModeSamePipeline = "same_pipeline"
+	TakeoverModeSameSource   = "same_source"
+	TakeoverModeAnyManaged   = "any_managed"
+	TakeoverModeNever        = "never"
+)
+
 const (
 	StateModeSingleOwner = "single_owner"
 	StateModeSharedRoot  = "shared_root"
@@ -95,6 +102,9 @@ func ApplyDefaults(cfg *Config) {
 			if destination.Reconciliation.Mode == "" {
 				destination.Reconciliation.Mode = ReconciliationModeReplace
 			}
+			if destination.Takeover.Mode == "" {
+				destination.Takeover.Mode = TakeoverModeSamePipeline
+			}
 			if destination.Transfer.OnDestinationSame == "" {
 				destination.Transfer.OnDestinationSame = TransferActionSkip
 			}

internal/config/load_test.go

@@ -39,6 +39,9 @@ pipelines:
 	if got, want := destination.State.Mode, StateModeSingleOwner; got != want {
 		t.Fatalf("state mode default = %q, want %q", got, want)
 	}
+	if got, want := destination.Takeover.Mode, TakeoverModeSamePipeline; got != want {
+		t.Fatalf("takeover mode default = %q, want %q", got, want)
+	}
 	if destination.Retention.Prune.Enabled {
 		t.Fatal("retention.prune.enabled default = true, want false")
 	}
@@ -231,6 +234,50 @@ pipelines:
 	}
 }
 
+func TestLoadFileAcceptsExplicitTakeoverModes(t *testing.T) {
+	cfg := loadConfig(t, `
+pipelines:
+  - id: reports
+    source:
+      backend: local
+      path: /source
+    destinations:
+      - id: same-pipeline
+        backend: local
+        path: /same-pipeline
+        takeover:
+          mode: same_pipeline
+      - id: same-source
+        backend: local
+        path: /same-source
+        takeover:
+          mode: same_source
+      - id: any-managed
+        backend: local
+        path: /any-managed
+        takeover:
+          mode: any_managed
+      - id: never
+        backend: local
+        path: /never
+        takeover:
+          mode: never
+`)
+
+	destinations := cfg.Pipelines[0].Destinations
+	wants := []string{
+		TakeoverModeSamePipeline,
+		TakeoverModeSameSource,
+		TakeoverModeAnyManaged,
+		TakeoverModeNever,
+	}
+	for index, want := range wants {
+		if got := destinations[index].Takeover.Mode; got != want {
+			t.Fatalf("destinations[%d].takeover.mode = %q, want %q", index, got, want)
+		}
+	}
+}
+
 func TestLoadFileAcceptsRetentionPruneConfig(t *testing.T) {
 	cfg := loadConfig(t, `
 pipelines:
@@ -868,6 +915,38 @@ pipelines:
 `, "on_destination_older must be replace or fail")
 }
 
+func TestLoadFileRejectsInvalidTakeoverMode(t *testing.T) {
+	assertLoadError(t, `
+pipelines:
+  - id: reports
+    source:
+      backend: local
+      path: /source
+    destinations:
+      - id: archive
+        backend: local
+        path: /archive
+        takeover:
+          mode: unmanaged
+`, "takeover.mode must be same_pipeline, same_source, any_managed, or never")
+}
+
+func TestLoadFileRejectsUnknownTakeoverFields(t *testing.T) {
+	assertLoadError(t, `
+pipelines:
+  - id: reports
+    source:
+      backend: local
+      path: /source
+    destinations:
+      - id: archive
+        backend: local
+        path: /archive
+        takeover:
+          surprise: true
+`, "field surprise not found")
+}
+
 func TestLoadFileRejectsInvalidValidationAction(t *testing.T) {
 	assertLoadError(t, `
 pipelines:

internal/config/validate.go

@@ -76,6 +76,7 @@ func Validate(cfg Config) error {
 			errs = validateLinks(errs, destinationContext+".links", destination.Links)
 			errs = validateStatePolicy(errs, destinationContext+".state", destination.State)
 			errs = validateReconciliationPolicy(errs, destinationContext+".reconciliation", destination.Reconciliation)
+			errs = validateTakeoverPolicy(errs, destinationContext+".takeover", destination.Takeover)
 			errs = validateRetentionPolicy(errs, destinationContext+".retention", destination.Retention)
 			errs = validateTransferPolicy(errs, destinationContext+".transfer", destination.Transfer)
 		}
@@ -375,6 +376,15 @@ func validateReconciliationPolicy(errs ValidationErrors, context string, policy
 	return errs
 }
 
+func validateTakeoverPolicy(errs ValidationErrors, context string, policy TakeoverPolicy) ValidationErrors {
+	switch policy.Mode {
+	case TakeoverModeSamePipeline, TakeoverModeSameSource, TakeoverModeAnyManaged, TakeoverModeNever:
+	default:
+		errs = append(errs, context+".mode must be "+TakeoverModeSamePipeline+", "+TakeoverModeSameSource+", "+TakeoverModeAnyManaged+", or "+TakeoverModeNever)
+	}
+	return errs
+}
+
 func validateRetentionPolicy(errs ValidationErrors, context string, policy RetentionPolicy) ValidationErrors {
 	prune := policy.Prune
 	if !prune.Enabled {

internal/config/validate_test.go

@@ -142,6 +142,64 @@ func TestValidateReconciliationPolicy(t *testing.T) {
 	}
 }
 
+func TestValidateTakeoverPolicy(t *testing.T) {
+	tests := []struct {
+		name    string
+		mode    string
+		wantErr bool
+	}{
+		{name: "same pipeline", mode: TakeoverModeSamePipeline},
+		{name: "same source", mode: TakeoverModeSameSource},
+		{name: "any managed", mode: TakeoverModeAnyManaged},
+		{name: "never", mode: TakeoverModeNever},
+		{name: "invalid", mode: "unmanaged", wantErr: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg := Config{Pipelines: []Pipeline{{
+				ID:     "reports",
+				Source: Backend{Backend: BackendLocal, Path: "/source"},
+				Destinations: []Destination{{
+					ID:       "archive",
+					Backend:  BackendLocal,
+					Path:     "/destination",
+					Takeover: TakeoverPolicy{Mode: tt.mode},
+				}},
+			}}}
+			ApplyDefaults(&cfg)
+			err := Validate(cfg)
+			if tt.wantErr && err == nil {
+				t.Fatal("Validate() error = nil, want error")
+			}
+			if !tt.wantErr && err != nil {
+				t.Fatalf("Validate() error = %v", err)
+			}
+		})
+	}
+}
+
+func TestValidateTakeoverPolicyReportsFieldContext(t *testing.T) {
+	cfg := Config{Pipelines: []Pipeline{{
+		ID:     "reports",
+		Source: Backend{Backend: BackendLocal, Path: "/source"},
+		Destinations: []Destination{{
+			ID:       "archive",
+			Backend:  BackendLocal,
+			Path:     "/destination",
+			Takeover: TakeoverPolicy{Mode: "unmanaged"},
+		}},
+	}}}
+	ApplyDefaults(&cfg)
+	err := Validate(cfg)
+	if err == nil {
+		t.Fatal("Validate() error = nil, want error")
+	}
+	want := "pipelines[0].destinations[0].takeover.mode must be same_pipeline, same_source, any_managed, or never"
+	if !strings.Contains(err.Error(), want) {
+		t.Fatalf("Validate() error = %q, want %q", err, want)
+	}
+}
+
 func TestValidateStatePolicy(t *testing.T) {
 	tests := []struct {
 		name    string