Implement shared-root takeover policy

docs/config.md

@@ -416,7 +416,7 @@ pipelines:
             mode: sidecar
 ```
 
-Every output path in a shared root belongs to exactly one `pipeline_id` and `destination_id`. A different owner planning the same path fails as a conflict.
+Every output path in a shared root belongs to exactly one `pipeline_id` and `destination_id`. A different owner planning the same path fails as a conflict unless `takeover.mode` allows the managed output path to move to the current owner.
 
 ## Takeover Policy
 
@@ -427,14 +427,16 @@ takeover:
 
 - `takeover.mode`: optional. Accepted values are `same_pipeline`, `same_source`, `any_managed`, and `never`; default is `same_pipeline`.
 
-Takeover controls when a single-owner destination may normally replace valid distributor-managed state whose pipeline, destination, or source identity differs from the current publication.
+Takeover controls when a destination may normally replace valid distributor-managed state whose pipeline, destination, source identity, or shared-root output owner differs from the current publication.
 
 - `same_pipeline`: replace managed state owned by the same pipeline. The previous destination id and source id may differ.
 - `same_source`: replace managed state only when the existing source manifest id matches the current source id.
-- `any_managed`: replace any valid distributor-managed single-owner state at the selected destination bundle path.
+- `any_managed`: replace any valid distributor-managed state at the selected destination bundle path.
 - `never`: do not replace identity or source conflicts without the explicit forced replacement workflow.
 
-Takeover does not apply to unmanaged content, invalid destination state, same-created digest conflicts, same-source destination-newer comparisons, or shared-root output paths owned by another owner.
+For shared-root state, `same_pipeline` permits taking over output paths owned by another destination in the same pipeline, `same_source` permits taking over output paths whose owner records the same source manifest id, and `any_managed` permits taking over output paths owned by any valid shared-root owner. Unrelated owner records and non-conflicting outputs remain managed by their existing owners.
+
+Takeover does not apply to unmanaged content, invalid destination state, same-created digest conflicts, or same-source destination-newer comparisons.
 
 ## Reconciliation Policy
 

docs/internal/publish.md

@@ -30,13 +30,13 @@ The package depends on `internal/storage.Backend` for source and destination IO,
 
 Planning inspects destination state through `internal/state`, compares it with the source manifest, and maps comparison outcomes plus transfer and takeover policy into actions: `publish_new`, `replace_older`, `replace_takeover`, `force_replace`, `skip_same`, `skip_destination_newer`, `fail_conflict`, or `fail_unmanaged`.
 
-Single-owner destinations compare the whole destination state against the configured pipeline and destination ids. Valid managed identity and source conflicts can become `replace_takeover` when `takeover.mode` allows them. Shared-root destinations compare only the current owner scope, keyed by pipeline id and destination id. An absent shared-root owner is publishable for that owner unless a planned output collides with unmanaged storage content. Planned writes to a path owned by another shared-root owner fail as conflicts.
+Single-owner destinations compare the whole destination state against the configured pipeline and destination ids. Valid managed identity and source conflicts can become `replace_takeover` when `takeover.mode` allows them. Shared-root destinations compare only the current owner scope, keyed by pipeline id and destination id. An absent shared-root owner is publishable for that owner unless a planned output collides with unmanaged storage content. Planned writes to a path owned by another shared-root owner become `replace_takeover` when `takeover.mode` allows that managed output path to move to the current owner.
 
 Execution writes destination state after selected outputs are written. Destination state includes copied source output metadata, generated output metadata, output timestamps, embedded source manifest, reconciliation metadata, link metadata when configured, pipeline id, destination id, and publication timestamps.
 
 ## Skip And Resume Behavior
 
-`skip_same` and `skip_destination_newer` execute as no-ops. Replacement-mode single-owner updates remove managed output paths from existing state plus `.distributor.json`, verify the destination is empty, and write state whose outputs are exactly the new plan. Single-owner `replace_takeover` uses the same managed replacement mechanics and does not retain omitted outputs through merge reconciliation. Replacement-mode shared-root updates remove only current-owner omitted outputs and preserve unrelated owners. Merge-mode same-source updates retain omitted managed outputs, overwrite only paths already recorded as managed, reject unmanaged destination path collisions, and write cumulative output state. Failed writes trigger cleanup where practical; merge cleanup removes only newly created outputs from the failed attempt.
+`skip_same` and `skip_destination_newer` execute as no-ops. Replacement-mode single-owner updates remove managed output paths from existing state plus `.distributor.json`, verify the destination is empty, and write state whose outputs are exactly the new plan. `replace_takeover` uses managed replacement mechanics and does not retain omitted outputs through merge reconciliation. Replacement-mode shared-root updates remove only current-owner omitted outputs and preserve unrelated owners. Shared-root takeover rewrites only the taken-over output records and current owner records. Merge-mode same-source updates retain omitted managed outputs, overwrite only paths already recorded as managed, reject unmanaged destination path collisions, and write cumulative output state. Failed writes trigger cleanup where practical; same-source merge cleanup removes only newly created outputs from the failed attempt.
 
 Shared-root execution writes schema version `3` state. It preserves unrelated owner records and outputs, updates only the publishing owner metadata, preserves root `created_at`, and updates root `updated_at` after successful state writes. Compatible single-owner state for the same pipeline and destination is converted to shared-root state on successful publish.
 
@@ -46,7 +46,7 @@ Retention pruning is not part of publish execution and does not run automaticall
 
 ## Failure Behavior
 
-Planning fails for incomplete requests, invalid publish/transform policy, invalid state mode, invalid reconciliation mode, output path collisions, invalid destination state, unmanaged destination content without force, shared-root owner path conflicts, conflict outcomes not allowed by transfer policy, unresolved transforms, invalid Markdown output selection, and invalid link URL planning.
+Planning fails for incomplete requests, invalid publish/transform policy, invalid state mode, invalid reconciliation mode, output path collisions, invalid destination state, unmanaged destination content without force, shared-root owner path conflicts not allowed by `takeover.mode`, conflict outcomes not allowed by transfer policy, unresolved transforms, invalid Markdown output selection, and invalid link URL planning.
 
 Execution fails on delete, read, transform output, unmanaged merge path collision, shared-root ownership conflict, write, state validation, state serialization, or context errors. Execution refuses actions that are not executable publish or replacement actions.
 

docs/operations.md

@@ -68,13 +68,13 @@ Published destination bundle paths contain `.distributor.json`. See [Destination
 - Matching destination state: skip as already published.
 - Older destination state for the same source id: replace if transfer policy allows it.
 - Newer destination state for the same source id: skip by default.
-- Valid single-owner state with an identity or source mismatch: replace only when destination `takeover.mode` allows it.
+- Valid managed state with an identity, source, or shared-root output-owner mismatch: replace only when destination `takeover.mode` allows it.
-- Invalid destination state, identity or source mismatches not allowed by `takeover.mode`, or same-created digest mismatch: fail by default.
+- Invalid destination state, identity, source, or shared-root output-owner mismatches not allowed by `takeover.mode`, or same-created digest mismatch: fail by default.
 - Content without `.distributor.json`: fail as unmanaged content by default.
 
 When destination state is older than the source, `transfer.on_destination_older` controls whether publication may proceed and `reconciliation.mode` controls how managed outputs are updated.
 
-For single-owner takeover replacement, `reconciliation.mode: merge` does not retain omitted outputs from the previous source identity. The destination is rewritten as a managed replacement for the current source.
+For takeover replacement, `reconciliation.mode: merge` does not retain omitted outputs from the previous source identity. The destination is rewritten as a managed replacement for the current source or shared-root owner.
 
 `reconciliation.mode: replace` is the default. It deletes only managed output paths recorded in `.distributor.json` plus the state file, verifies the destination bundle path is empty, then writes the newly planned outputs and state. The new state `outputs` array is exactly the newly planned output set.
 
@@ -82,11 +82,11 @@ For single-owner takeover replacement, `reconciliation.mode: merge` does not ret
 
 For both modes, retained or overwritten paths are identified only from `.distributor.json`; unmanaged files are not adopted.
 
-For `state.mode: shared_root`, one destination root may contain outputs from multiple pipeline/destination owners. Comparisons, replacement, and merge retention are scoped to the current owner. Outputs owned by other owners are preserved. A planned output path owned by another owner fails as a conflict, and a planned path that exists in storage but is not recorded in state fails as unmanaged content by default.
+For `state.mode: shared_root`, one destination root may contain outputs from multiple pipeline/destination owners. Comparisons, replacement, and merge retention are scoped to the current owner. Outputs owned by other owners are preserved unless a planned output path is owned by another valid owner and `takeover.mode` allows moving that path to the current owner. A planned path that exists in storage but is not recorded in state fails as unmanaged content by default.
 
 If `state.mode: shared_root` is configured on a destination whose existing single-owner state belongs to the same pipeline and destination, the next successful publish converts that state file to shared-root schema. Existing single-owner state for a different pipeline or destination remains a conflict.
 
-If a write fails after some outputs were written, `distributor` attempts cleanup before returning the error. In `replace` mode, cleanup removes outputs written during that failed attempt. In `merge` mode, cleanup removes only newly created outputs from that failed attempt; overwritten managed outputs are left in place because they previously belonged to the managed set. Operators should still inspect the destination after a failed write before retrying.
+If a write fails after some outputs were written, `distributor` attempts cleanup before returning the error. In `replace` mode and takeover replacement, cleanup removes outputs written during that failed attempt. In same-source `merge` mode, cleanup removes only newly created outputs from that failed attempt; overwritten managed outputs are left in place because they previously belonged to the managed set. Operators should still inspect the destination after a failed write before retrying.
 
 Fan-out destinations are independent. If one destination fails after planning or execution begins, later destinations are still attempted. The command exits non-zero if any destination failed.
 
@@ -153,13 +153,13 @@ For single-owner state, the state owner must match the selected pipeline and des
 
 `run --dry-run` loads config, resolves credentials, discovers source bundles, opens destinations, inspects destination state, builds publish plans, and prints actions. It does not write outputs, `.distributor.json`, or SSH `known_hosts` entries. For reconciliation, dry runs report the same high-level action labels as execution; inspect the configured destination's `reconciliation.mode` to determine whether `replace_older` will replace the managed set or merge into it.
 
-For shared-root destinations, dry runs are owner-scoped. A `replace_older` action replaces or merges only the current owner according to `reconciliation.mode`; unrelated owners remain managed by the shared-root state. Paths owned by another owner still fail as conflicts.
+For shared-root destinations, dry runs are owner-scoped. A `replace_older` action replaces or merges only the current owner according to `reconciliation.mode`; unrelated owners remain managed by the shared-root state. Paths owned by another owner fail as conflicts unless `takeover.mode` allows ownership transfer.
 
 Review these action labels before publishing:
 
 - `publish_new`: destination state is absent, or a shared-root owner is absent and planned paths are publishable.
 - `replace_older`: destination state is older than the source.
-- `replace_takeover`: single-owner destination state is valid managed state and `takeover.mode` allows replacement across an identity or source mismatch.
+- `replace_takeover`: destination state is valid managed state and `takeover.mode` allows replacement across an identity, source, or shared-root output-owner mismatch.
 - `skip_same`: destination state already matches the source.
 - `skip_destination_newer`: destination state is newer than the source and is skipped.
 - `force_replace`: destructive replacement selected because `--force` is present and policy permits it.
@@ -286,7 +286,7 @@ Use these recovery boundaries:
 - For source validation failures, regenerate the source bundle and manifest together.
 - For an empty or missing destination, rerun after fixing config or storage access.
 - For unmanaged destination content, move unrelated files aside or use a different destination path before publishing.
-- For shared-root ownership conflicts, change one owner so it writes a different destination path, or use a separate destination root.
+- For shared-root ownership conflicts, change one owner so it writes a different destination path, use a separate destination root, or configure `takeover.mode` when the current owner should take over valid managed output paths.
 - For missing managed output files recorded in state, run `reconcile-state --dry-run`, then apply `reconcile-state` if the missing files should no longer be considered managed.
 - For configured retention cleanup, run `prune --dry-run`, then apply `prune --apply` after reviewing the managed output list.
 - For failed writes, inspect the destination bundle path, remove only confirmed partial outputs if needed, then rerun `--dry-run`. In merge mode, retained outputs may be intentional managed outputs from the prior state.

docs/troubleshooting.md

@@ -205,7 +205,7 @@ Reference: [Operations](operations.md#forced-replacement-workflow).
 
 Symptom: `fail_conflict`, `destination source id differs`, `same id and created time but different digest`, `pipeline id ... does not match`, or `destination id ... does not match`.
 
-Likely cause: `.distributor.json` belongs to a different pipeline, destination, source id, or same-created source with different content. For single-owner state, identity and source mismatches can publish as `replace_takeover` only when destination `takeover.mode` allows them.
+Likely cause: `.distributor.json` belongs to a different pipeline, destination, source id, shared-root output owner, or same-created source with different content. Valid managed identity, source, and shared-root output-owner mismatches can publish as `replace_takeover` only when destination `takeover.mode` allows them.
 
 Diagnostic:
 
@@ -214,7 +214,7 @@ cat <destination-path>/.distributor.json
 go run ./cmd/distributor inspect <source-root>
 ```
 
-Safe fix: verify the source and destination are intended to match. Use a separate destination path for unrelated content. For normal single-owner managed replacement, configure destination `takeover.mode` to match the intended ownership boundary. To force exceptional replacement, configure `transfer.on_conflict: replace`, preview with `--dry-run --force`, then publish with `--force`.
+Safe fix: verify the source and destination are intended to match. Use a separate destination path for unrelated content. For normal managed replacement, configure destination `takeover.mode` to match the intended ownership boundary. To force exceptional replacement, configure `transfer.on_conflict: replace`, preview with `--dry-run --force`, then publish with `--force`.
 
 Reference: [Operations](operations.md#destination-state-and-retry-behavior).
 
@@ -312,7 +312,7 @@ Reference: [Configuration](config.md#publish-and-transform-policy).
 
 Symptom: `fail_conflict` with a reason like `destination output path ... is owned by <pipeline>/<destination>`.
 
-Likely cause: a `state.mode: shared_root` destination planned an output path already recorded in `.distributor.json` for another pipeline/destination owner.
+Likely cause: a `state.mode: shared_root` destination planned an output path already recorded in `.distributor.json` for another pipeline/destination owner, and `takeover.mode` does not allow that managed path to move to the current owner.
 
 Diagnostic:
 

internal/publish/execute.go

@@ -148,7 +148,7 @@ func executeSharedRoot(ctx context.Context, req Request, plan Plan) error {
 			return err
 		}
 	}
-	if plan.Action == ActionReplaceOlder && plan.Reconciliation.Mode == config.ReconciliationModeReplace {
+	if plan.Action == ActionReplaceTakeover || (plan.Action == ActionReplaceOlder && plan.Reconciliation.Mode == config.ReconciliationModeReplace) {
 		if err := req.DestinationBackend.DeleteManagedOutputs(ctx, req.DestinationBundlePath, sharedRootOutputPaths(plan.OwnerOutputsToDelete), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil {
 			return err
 		}
@@ -158,7 +158,7 @@ func executeSharedRoot(ctx context.Context, req Request, plan Plan) error {
 	newOutputs := make([]Output, 0, len(plan.Outputs))
 	cleanup := func() {
 		outputs := writtenOutputs
-		if plan.Reconciliation.Mode == config.ReconciliationModeMerge {
+		if plan.Action == ActionReplaceOlder && plan.Reconciliation.Mode == config.ReconciliationModeMerge {
 			outputs = newOutputs
 		}
 		_ = req.DestinationBackend.DeleteManagedOutputs(ctx, req.DestinationBundlePath, ManagedOutputPaths(outputs), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true})
@@ -257,6 +257,11 @@ func outputManagedBySharedRootPlan(output Output, plan Plan) bool {
 			return true
 		}
 	}
+	for _, existing := range plan.TakenOverOwnerOutputs {
+		if existing.Path == output.DestinationPath {
+			return true
+		}
+	}
 	return false
 }
 
@@ -283,6 +288,7 @@ func sharedRootStateForPlan(req Request, plan Plan, now time.Time) (state.Shared
 		scope = state.CurrentOwnerScope(req.PipelineID, req.DestinationID)
 	}
 	base := sharedRootBaseState(req, plan, now)
+	base = removeTakenOverSharedRootOutputs(base, plan.TakenOverOwnerOutputs)
 	owner := state.OwnerRecord{
 		Scope:          scope,
 		Reconciliation: state.ReconciliationPolicy{Mode: plan.Reconciliation.Mode},
@@ -319,6 +325,22 @@ func sharedRootBaseState(req Request, plan Plan, now time.Time) state.SharedRoot
 	return newSharedRootState(req, now)
 }
 
+func removeTakenOverSharedRootOutputs(sharedRoot state.SharedRootState, takenOver []state.SharedRootOutputFile) state.SharedRootState {
+	if len(takenOver) == 0 {
+		return sharedRoot
+	}
+	paths := sharedRootOutputPathSet(takenOver)
+	next := sharedRoot
+	next.Outputs = make([]state.SharedRootOutputFile, 0, len(sharedRoot.Outputs))
+	for _, output := range sharedRoot.Outputs {
+		if _, remove := paths[output.Path]; remove {
+			continue
+		}
+		next.Outputs = append(next.Outputs, output)
+	}
+	return next
+}
+
 func newSharedRootState(req Request, now time.Time) state.SharedRootState {
 	return state.SharedRootState{
 		SchemaVersion:      state.SharedRootSchemaVersion,

internal/publish/plan.go

@@ -66,6 +66,7 @@ type Plan struct {
 	ExistingState         *state.DistributorState
 	ExistingSharedRoot    *state.SharedRootState
 	OtherOwnerOutputs     []state.SharedRootOutputFile
+	TakenOverOwnerOutputs []state.SharedRootOutputFile
 	RetainedOwnerOutputs  []state.SharedRootOutputFile
 	OwnerOutputsToDelete  []state.SharedRootOutputFile
 	OwnerOutputsToWrite   []Output
@@ -126,18 +127,21 @@ func Build(ctx context.Context, req Request) (Plan, error) {
 	}
 	if stateMode == config.StateModeSharedRoot {
 		sharedDetails, err := planSharedRootOwner(ctx, req, status, action, reconciliation, outputs)
+		plan.Action = sharedDetails.Action
+		if sharedDetails.Reason != "" {
+			plan.Reason = sharedDetails.Reason
+		}
 		plan.OtherOwnerOutputs = sharedDetails.OtherOwnerOutputs
+		plan.TakenOverOwnerOutputs = sharedDetails.TakenOverOwnerOutputs
 		plan.RetainedOwnerOutputs = sharedDetails.RetainedOwnerOutputs
 		plan.OwnerOutputsToDelete = sharedDetails.OwnerOutputsToDelete
 		plan.OwnerOutputsToWrite = sharedDetails.OwnerOutputsToWrite
 		if err != nil {
-			plan.Action = sharedDetails.Action
-			plan.Reason = sharedDetails.Reason
 			return plan, err
 		}
 	}
-	if action == ActionFailConflict || action == ActionFailUnmanaged {
+	if plan.Action == ActionFailConflict || plan.Action == ActionFailUnmanaged {
-		return plan, fmt.Errorf("%s: %s", action, reason)
+		return plan, fmt.Errorf("%s: %s", plan.Action, plan.Reason)
 	}
 	return plan, nil
 }
@@ -242,6 +246,7 @@ type sharedRootPlanDetails struct {
 	Action                Action
 	Reason                string
 	OtherOwnerOutputs     []state.SharedRootOutputFile
+	TakenOverOwnerOutputs []state.SharedRootOutputFile
 	RetainedOwnerOutputs  []state.SharedRootOutputFile
 	OwnerOutputsToDelete  []state.SharedRootOutputFile
 	OwnerOutputsToWrite   []Output
@@ -260,19 +265,29 @@ func planSharedRootOwner(ctx context.Context, req Request, status state.Destinat
 		details.OwnerOutputsToWrite = append([]Output(nil), outputs...)
 		return details, nil
 	}
-	if conflict, ok := sharedRootPathOwnershipConflict(status, scope, plannedPaths); ok {
+	conflicts := sharedRootPathOwnershipConflicts(status, scope, plannedPaths)
-		reason := fmt.Sprintf("destination output path %s is owned by %s/%s", conflict.Path, conflict.Owner.PipelineID, conflict.Owner.DestinationID)
+	if len(conflicts) > 0 {
+		for _, conflict := range conflicts {
+			if sharedRootTakeoverAllowed(req, status, conflict) {
+				continue
+			}
+			reason := sharedRootOwnershipConflictReason(conflict)
 			details.Action = ActionFailConflict
 			details.Reason = reason
 			return details, fmt.Errorf("%s: %s", ActionFailConflict, reason)
 		}
+		details.Action = ActionReplaceTakeover
+		details.Reason = sharedRootOwnershipConflictReason(conflicts[0])
+		details.TakenOverOwnerOutputs = sharedRootConflictOutputs(status.SharedRoot, conflicts)
+	}
 	if err := rejectSharedRootUnmanagedCollisions(ctx, req.DestinationBackend, req.DestinationBundlePath, status, scope, plannedPaths); err != nil {
 		details.Action = ActionFailUnmanaged
 		details.Reason = err.Error()
 		return details, fmt.Errorf("%s: %s", ActionFailUnmanaged, err)
 	}
 
-	details.OtherOwnerOutputs = otherOwnerOutputs(status, scope)
+	takenOverPaths := sharedRootOutputPathSet(details.TakenOverOwnerOutputs)
+	details.OtherOwnerOutputs = otherOwnerOutputsExcept(status, scope, takenOverPaths)
 	ownerOutputs := currentOwnerOutputs(status, scope)
 	planned := make(map[string]struct{}, len(plannedPaths))
 	for _, path := range plannedPaths {
@@ -282,11 +297,11 @@ func planSharedRootOwner(ctx context.Context, req Request, status state.Destinat
 		if _, exists := planned[output.Path]; exists {
 			continue
 		}
-		if action == ActionReplaceOlder && reconciliation.Mode == config.ReconciliationModeReplace {
+		if details.Action == ActionReplaceTakeover || (details.Action == ActionReplaceOlder && reconciliation.Mode == config.ReconciliationModeReplace) {
 			details.OwnerOutputsToDelete = append(details.OwnerOutputsToDelete, output)
 			continue
 		}
-		if action == ActionReplaceOlder && reconciliation.Mode == config.ReconciliationModeMerge {
+		if details.Action == ActionReplaceOlder && reconciliation.Mode == config.ReconciliationModeMerge {
 			details.RetainedOwnerOutputs = append(details.RetainedOwnerOutputs, output)
 		}
 	}
@@ -311,11 +326,76 @@ func outputPaths(outputs []Output) []string {
 	return paths
 }
 
-func sharedRootPathOwnershipConflict(status state.DestinationStatus, scope state.OwnerScope, paths []string) (state.PathOwnershipConflict, bool) {
+func sharedRootOwnershipConflictReason(conflict state.PathOwnershipConflict) string {
-	if status.SharedRoot != nil {
+	return fmt.Sprintf("destination output path %s is owned by %s/%s", conflict.Path, conflict.Owner.PipelineID, conflict.Owner.DestinationID)
-		return status.SharedRoot.PathOwnershipConflict(scope, paths)
+}
+
+func sharedRootPathOwnershipConflicts(status state.DestinationStatus, scope state.OwnerScope, paths []string) []state.PathOwnershipConflict {
+	if status.SharedRoot == nil {
+		return nil
+	}
+	conflicts := make([]state.PathOwnershipConflict, 0)
+	seen := make(map[string]struct{}, len(paths))
+	for _, path := range paths {
+		if _, exists := seen[path]; exists {
+			continue
+		}
+		seen[path] = struct{}{}
+		owner, exists := status.SharedRoot.OutputOwner(path)
+		if !exists || owner == scope {
+			continue
+		}
+		conflicts = append(conflicts, state.PathOwnershipConflict{
+			Path:         path,
+			Owner:        owner,
+			CurrentOwner: scope,
+			Detail: state.ComparisonDetail{
+				Kind:             state.ComparisonDetailSharedRootOutputOwner,
+				Path:             path,
+				CurrentOwner:     scope,
+				ConflictingOwner: owner,
+			},
+		})
+	}
+	return conflicts
+}
+
+func sharedRootConflictOutputs(sharedRoot *state.SharedRootState, conflicts []state.PathOwnershipConflict) []state.SharedRootOutputFile {
+	if sharedRoot == nil || len(conflicts) == 0 {
+		return nil
+	}
+	paths := make(map[string]struct{}, len(conflicts))
+	for _, conflict := range conflicts {
+		paths[conflict.Path] = struct{}{}
+	}
+	outputs := make([]state.SharedRootOutputFile, 0, len(conflicts))
+	for _, output := range sharedRoot.Outputs {
+		if _, exists := paths[output.Path]; exists {
+			outputs = append(outputs, output)
+		}
+	}
+	return outputs
+}
+
+func sharedRootTakeoverAllowed(req Request, status state.DestinationStatus, conflict state.PathOwnershipConflict) bool {
+	if status.SharedRoot == nil {
+		return false
+	}
+	takeover := normalizeTakeover(req.Takeover)
+	switch takeover.Mode {
+	case config.TakeoverModeSamePipeline:
+		return conflict.Owner.PipelineID == req.PipelineID
+	case config.TakeoverModeSameSource:
+		owner, ok := status.SharedRoot.Owner(conflict.Owner)
+		return ok && owner.Source.Manifest.ID == req.SourceBundle.Manifest.ID
+	case config.TakeoverModeAnyManaged:
+		_, ok := status.SharedRoot.Owner(conflict.Owner)
+		return ok
+	case config.TakeoverModeNever:
+		return false
+	default:
+		return false
 	}
-	return state.PathOwnershipConflict{}, false
 }
 
 func rejectSharedRootUnmanagedCollisions(ctx context.Context, backend storage.Backend, bundlePath string, status state.DestinationStatus, scope state.OwnerScope, paths []string) error {
@@ -349,18 +429,34 @@ func pathManagedBySharedRootStatus(status state.DestinationStatus, scope state.O
 }
 
 func otherOwnerOutputs(status state.DestinationStatus, scope state.OwnerScope) []state.SharedRootOutputFile {
+	return otherOwnerOutputsExcept(status, scope, nil)
+}
+
+func otherOwnerOutputsExcept(status state.DestinationStatus, scope state.OwnerScope, exclude map[string]struct{}) []state.SharedRootOutputFile {
 	if status.SharedRoot == nil {
 		return nil
 	}
 	outputs := make([]state.SharedRootOutputFile, 0, len(status.SharedRoot.Outputs))
 	for _, output := range status.SharedRoot.Outputs {
-		if output.Owner != scope {
+		if output.Owner == scope {
+			continue
+		}
+		if _, skip := exclude[output.Path]; skip {
+			continue
+		}
 		outputs = append(outputs, output)
 	}
-	}
 	return outputs
 }
 
+func sharedRootOutputPathSet(outputs []state.SharedRootOutputFile) map[string]struct{} {
+	paths := make(map[string]struct{}, len(outputs))
+	for _, output := range outputs {
+		paths[output.Path] = struct{}{}
+	}
+	return paths
+}
+
 func currentOwnerOutputs(status state.DestinationStatus, scope state.OwnerScope) []state.SharedRootOutputFile {
 	if status.SharedRoot != nil {
 		outputs := make([]state.SharedRootOutputFile, 0, len(status.SharedRoot.Outputs))

internal/publish/shared_root_test.go

@@ -113,6 +113,94 @@ func TestBuildSharedRootRejectsOtherOwnerPathConflict(t *testing.T) {
 	}
 }
 
+func TestBuildSharedRootPlansOutputTakeoverByPolicy(t *testing.T) {
+	tests := []struct {
+		name       string
+		takeover   config.TakeoverPolicy
+		ownerScope state.OwnerScope
+		sameSource bool
+		wantAction Action
+		wantErr    string
+	}{
+		{
+			name:       "default same pipeline allows different destination",
+			takeover:   config.TakeoverPolicy{},
+			ownerScope: state.CurrentOwnerScope("reports", "web"),
+			wantAction: ActionReplaceTakeover,
+		},
+		{
+			name:       "same pipeline refuses different pipeline",
+			takeover:   config.TakeoverPolicy{Mode: config.TakeoverModeSamePipeline},
+			ownerScope: state.CurrentOwnerScope("other", "archive"),
+			wantErr:    "fail_conflict",
+		},
+		{
+			name:       "same source allows different pipeline",
+			takeover:   config.TakeoverPolicy{Mode: config.TakeoverModeSameSource},
+			ownerScope: state.CurrentOwnerScope("other", "archive"),
+			sameSource: true,
+			wantAction: ActionReplaceTakeover,
+		},
+		{
+			name:       "same source refuses different source",
+			takeover:   config.TakeoverPolicy{Mode: config.TakeoverModeSameSource},
+			ownerScope: state.CurrentOwnerScope("reports", "web"),
+			wantErr:    "fail_conflict",
+		},
+		{
+			name:       "any managed allows different pipeline",
+			takeover:   config.TakeoverPolicy{Mode: config.TakeoverModeAnyManaged},
+			ownerScope: state.CurrentOwnerScope("other", "archive"),
+			wantAction: ActionReplaceTakeover,
+		},
+		{
+			name:       "never refuses same pipeline",
+			takeover:   config.TakeoverPolicy{Mode: config.TakeoverModeNever},
+			ownerScope: state.CurrentOwnerScope("reports", "web"),
+			wantErr:    "fail_conflict",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			sourceBackend := fake.New()
+			sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{
+				Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\n"}},
+			})
+			destinationBackend := fake.New()
+			sharedRoot := sharedRootStateWithOwners(t, sourceBundle.Manifest, false)
+			ownerManifest := sharedRoot.Owners[0].Source.Manifest
+			if tt.sameSource {
+				ownerManifest = sourceBundle.Manifest
+			}
+			setSharedRootOwnerOutput(t, &sharedRoot, 0, tt.ownerScope, ownerManifest, "report.md")
+			writeFakeSharedRootState(t, destinationBackend, "bundle", sharedRoot)
+
+			req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace)
+			req.Takeover = tt.takeover
+			plan, err := Build(context.Background(), req)
+			if tt.wantErr != "" {
+				if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
+					t.Fatalf("Build() error = %v, want %q", err, tt.wantErr)
+				}
+				if plan.Action != ActionFailConflict {
+					t.Fatalf("plan action = %s, want fail_conflict", plan.Action)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("Build() error = %v", err)
+			}
+			if plan.Action != tt.wantAction {
+				t.Fatalf("plan action = %s, want %s", plan.Action, tt.wantAction)
+			}
+			if got, want := sharedRootOutputPathList(plan.TakenOverOwnerOutputs), "report.md"; got != want {
+				t.Fatalf("taken over outputs = %q, want %q", got, want)
+			}
+		})
+	}
+}
+
 func TestBuildSharedRootRejectsUnmanagedPathCollision(t *testing.T) {
 	sourceBackend := fake.New()
 	sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{
@@ -203,6 +291,120 @@ func TestExecuteSharedRootReplaceDeletesOnlyCurrentOwnerOmittedOutputs(t *testin
 	}
 }
 
+func TestExecuteSharedRootTakeoverReassignsPathAndPreservesUnrelatedOutputs(t *testing.T) {
+	sourceBackend := fake.New()
+	sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{
+		Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
+	})
+	destinationBackend := fake.New()
+	sharedRoot := sharedRootStateWithOwners(t, sourceBundle.Manifest, false)
+	previousScope := state.CurrentOwnerScope("reports", "web")
+	setSharedRootOwnerOutput(t, &sharedRoot, 0, previousScope, sharedRoot.Owners[0].Source.Manifest, "report.md")
+	keepOutput := sharedRoot.Outputs[0]
+	keepOutput.Path = "web/keep.md"
+	keepOutput.SourcePath = "web/keep.md"
+	sharedRoot.Outputs = append(sharedRoot.Outputs, keepOutput)
+	writeFakeSharedRootState(t, destinationBackend, "bundle", sharedRoot)
+
+	req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace)
+	plan, err := Build(context.Background(), req)
+	if err != nil {
+		t.Fatalf("Build() error = %v", err)
+	}
+	if plan.Action != ActionReplaceTakeover {
+		t.Fatalf("plan action = %s, want %s", plan.Action, ActionReplaceTakeover)
+	}
+	if err := Execute(context.Background(), req, plan); err != nil {
+		t.Fatalf("Execute() error = %v", err)
+	}
+
+	testutil.AssertFakeFile(t, destinationBackend, "bundle/report.md", "# Report\nNew.\n")
+	testutil.AssertFakeFile(t, destinationBackend, "bundle/web/keep.md", "old")
+	destinationState := readFakeSharedRootState(t, destinationBackend, "bundle")
+	reportOutput, ok := findSharedRootOutputForTest(destinationState.Outputs, "report.md")
+	if !ok {
+		t.Fatal("report.md missing from shared-root outputs")
+	}
+	if reportOutput.Owner != state.CurrentOwnerScope("reports", "archive") {
+		t.Fatalf("report.md owner = %#v, want reports/archive", reportOutput.Owner)
+	}
+	keep, ok := findSharedRootOutputForTest(destinationState.Outputs, "web/keep.md")
+	if !ok {
+		t.Fatal("web/keep.md missing from shared-root outputs")
+	}
+	if keep.Owner != previousScope {
+		t.Fatalf("web/keep.md owner = %#v, want reports/web", keep.Owner)
+	}
+}
+
+func TestExecuteSharedRootTakeoverMergeDoesNotRetainOldSourceOutputs(t *testing.T) {
+	sourceBackend := fake.New()
+	sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{
+		Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
+	})
+	destinationBackend := fake.New()
+	sharedRoot := sharedRootStateWithOwners(t, sourceBundle.Manifest, false)
+	previousScope := state.CurrentOwnerScope("reports", "web")
+	setSharedRootOwnerOutput(t, &sharedRoot, 0, previousScope, sharedRoot.Owners[0].Source.Manifest, "report.md")
+
+	createdAt := sharedRoot.CreatedAt
+	oldManifest := sourceBundle.Manifest
+	oldManifest.ID = "old.source"
+	oldManifest.Created = oldManifest.Created.Add(-time.Hour)
+	oldManifest.Files = []bundle.ManifestFile{{
+		Path:   "old.md",
+		SHA256: bundle.FileDigest([]byte("old\n")),
+		Size:   int64(len("old\n")),
+	}}
+	oldManifest.Digest = bundle.BundleDigest(oldManifest.Files)
+	currentScope := state.CurrentOwnerScope("reports", "archive")
+	sharedRoot.Owners = append(sharedRoot.Owners, state.OwnerRecord{
+		Scope:          currentScope,
+		Reconciliation: state.ReconciliationPolicy{Mode: config.ReconciliationModeMerge},
+		Source:         state.SourceState{Manifest: oldManifest},
+	})
+	sharedRoot.Outputs = append(sharedRoot.Outputs, state.SharedRootOutputFile{
+		Path:          "old.md",
+		Kind:          state.OutputKindSource,
+		SourcePath:    "old.md",
+		SHA256:        oldManifest.Files[0].SHA256,
+		Size:          oldManifest.Files[0].Size,
+		Owner:         currentScope,
+		SourceID:      oldManifest.ID,
+		SourceDigest:  oldManifest.Digest,
+		SourceCreated: oldManifest.Created,
+		CreatedAt:     createdAt,
+		UpdatedAt:     createdAt,
+	})
+	writeFakeSharedRootState(t, destinationBackend, "bundle", sharedRoot)
+
+	req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeMerge)
+	req.PathMapping = config.PathMappingFixed
+	plan, err := Build(context.Background(), req)
+	if err != nil {
+		t.Fatalf("Build() error = %v", err)
+	}
+	if plan.Action != ActionReplaceTakeover {
+		t.Fatalf("plan action = %s, want %s", plan.Action, ActionReplaceTakeover)
+	}
+	if got, want := sharedRootOutputPathList(plan.OwnerOutputsToDelete), "old.md"; got != want {
+		t.Fatalf("owner outputs to delete = %q, want %q", got, want)
+	}
+	if len(plan.RetainedOwnerOutputs) != 0 {
+		t.Fatalf("retained owner outputs = %#v, want none", plan.RetainedOwnerOutputs)
+	}
+	if err := Execute(context.Background(), req, plan); err != nil {
+		t.Fatalf("Execute() error = %v", err)
+	}
+
+	testutil.AssertFakeFile(t, destinationBackend, "bundle/report.md", "# Report\nNew.\n")
+	testutil.AssertFakeMissing(t, destinationBackend, "bundle/old.md")
+	destinationState := readFakeSharedRootState(t, destinationBackend, "bundle")
+	if got, want := strings.Join(destinationState.AllManagedOutputPaths(), ","), "report.md"; got != want {
+		t.Fatalf("managed paths = %q, want %q", got, want)
+	}
+}
+
 func TestExecuteSharedRootMergeRetainsCurrentOwnerOmittedOutputs(t *testing.T) {
 	sourceBackend := fake.New()
 	sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{
@@ -322,6 +524,26 @@ func findSharedRootOutputForTest(outputs []state.SharedRootOutputFile, path stri
 	return state.SharedRootOutputFile{}, false
 }
 
+func setSharedRootOwnerOutput(t *testing.T, sharedRoot *state.SharedRootState, index int, scope state.OwnerScope, manifest bundle.Manifest, path string) {
+	t.Helper()
+	sharedRoot.Owners[index].Scope = scope
+	sharedRoot.Owners[index].Source = state.SourceState{Manifest: manifest}
+	sourcePath := path
+	if len(manifest.Files) > 0 {
+		sourcePath = manifest.Files[0].Path
+	}
+	sharedRoot.Outputs[index].Path = path
+	sharedRoot.Outputs[index].SourcePath = sourcePath
+	sharedRoot.Outputs[index].Owner = scope
+	sharedRoot.Outputs[index].SourceID = manifest.ID
+	sharedRoot.Outputs[index].SourceDigest = manifest.Digest
+	sharedRoot.Outputs[index].SourceCreated = manifest.Created
+	if len(manifest.Files) > 0 {
+		sharedRoot.Outputs[index].SHA256 = manifest.Files[0].SHA256
+		sharedRoot.Outputs[index].Size = manifest.Files[0].Size
+	}
+}
+
 func sharedRootStateWithOwners(t *testing.T, current bundle.Manifest, includeCurrent bool) state.SharedRootState {
 	t.Helper()
 	createdAt := time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC)