Add takeover backend regressions and close roadmap

docs/cli.md

@@ -292,6 +292,7 @@ Text output is optimized for direct operator use. JSON output is optimized for a
 - Usage errors and fatal setup errors exit non-zero and do not emit a JSON result document.
 - `run --format json` emits a JSON result for partial destination failures, sets `ok` to `false`, includes result details and errors, and exits non-zero.
 - Warnings are included in JSON output and are printed in text output when relevant.
+- `run` summaries include separate `replace_older`, `replace_takeover`, and `force_replace` counters. Takeover destination actions include `takeover_mode` in JSON and text output.
 
 ## Diagnostics And Recovery
 

docs/config.md

@@ -416,7 +416,27 @@ 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
+
+```yaml
+takeover:
+  mode: same_pipeline
+```
+
+- `takeover.mode`: optional. Accepted values are `same_pipeline`, `same_source`, `any_managed`, and `never`; default is `same_pipeline`.
+
+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 state at the selected destination bundle path.
+- `never`: do not replace identity or source conflicts without the explicit forced replacement workflow.
+
+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
 
@@ -529,6 +549,7 @@ Defaults are applied after YAML decoding and before validation:
 - `links.primary: auto` when a `links` block is present and `primary` is omitted
 - `state.mode: single_owner`
 - `reconciliation.mode: replace`
+- `takeover.mode: same_pipeline`
 - `retention.prune.enabled: false`
 - `transfer.on_destination_same: skip`
 - `transfer.on_destination_older: replace`

docs/integrations/destination-state.md

@@ -96,9 +96,11 @@ Destination reconciliation applies when destination state is older than the sour
 - `replace`: for single-owner state, delete managed output paths recorded in `outputs` plus `.distributor.json`, require the destination bundle path to be empty afterward, write the newly planned outputs, and write state whose `outputs` are exactly that new planned set. For shared-root state, delete only the current owner's omitted outputs and preserve unrelated owners.
 - `merge`: retain managed output paths omitted from the new plan, overwrite planned paths only when they are already recorded in existing state, fail when a newly planned path exists in storage but is not recorded as managed, and write state whose `outputs` are the cumulative managed set.
 
+Takeover replacement applies when valid managed state has an identity, source, or shared-root output-owner mismatch and destination `takeover.mode` permits the current publication to take ownership. It uses bounded managed replacement behavior. For shared-root state, only taken-over output records and the current owner records are rewritten; unrelated owner records and non-conflicting outputs remain managed by their existing owners.
+
 Top-level `links.primary_url` is selected from the newly planned outputs for the current publication. Retained outputs keep their existing per-output URL metadata.
 
-If a merge publication fails after writing outputs, cleanup removes only newly created outputs from that failed attempt. Previously managed overwritten paths remain managed and are not removed by failed-attempt cleanup.
+If a same-source merge publication fails after writing outputs, cleanup removes only newly created outputs from that failed attempt. Previously managed overwritten paths remain managed and are not removed by failed-attempt cleanup. Takeover replacement does not retain omitted outputs through merge reconciliation.
 
 ## Comparison Semantics
 
@@ -106,15 +108,16 @@ If a merge publication fails after writing outputs, cleanup removes only newly c
 
 - No state and no content: publish new outputs.
 - No state and existing content: treat the destination as unmanaged.
-- Shared-root state without the current owner: publish new outputs for that owner if planned paths do not collide with other owners or unmanaged content.
+- Shared-root state without the current owner: publish new outputs for that owner if planned paths do not collide with unmanaged content or with other owners that `takeover.mode` does not permit.
 - Matching embedded source manifest: skip.
 - Same source id with older `created`: replace if policy allows.
 - Same source id with newer `created`: skip by default.
 - Same source id and same `created` with different digest: conflict.
-- Different source id, pipeline id, or destination id: conflict.
+- Different source id, pipeline id, or destination id in single-owner state: conflict unless `takeover.mode` permits managed ownership transfer.
+- Shared-root output path owned by a different owner: conflict unless `takeover.mode` permits managed ownership transfer.
 - Invalid state JSON or invalid state fields: conflict.
 
-Normal single-owner replacement deletes only managed output paths recorded in `outputs` plus `.distributor.json`. Shared-root replacement deletes only omitted outputs for the current owner. Merge publication retains omitted managed outputs. Forced replacement deletes the bounded destination bundle path.
+Normal single-owner replacement deletes only managed output paths recorded in `outputs` plus `.distributor.json`. Shared-root replacement deletes only omitted outputs for the current owner. Merge publication retains omitted managed outputs for same-source replacement. Forced replacement deletes the bounded destination bundle path.
 
 ## State Repair Semantics
 
@@ -238,7 +241,7 @@ For shared-root state:
 
 If `state.mode: shared_root` is configured and existing state is a compatible single-owner `.distributor.json` for the same pipeline id and destination id, the next successful publish writes schema version `3` shared-root state for that owner.
 
-If existing single-owner state belongs to a different pipeline or destination, publish fails as a conflict. `distributor` does not implicitly take over unrelated state or unmanaged files.
+If `state.mode: shared_root` is configured and existing single-owner state belongs to a different pipeline or destination, publish fails as a conflict. `distributor` does not implicitly convert unrelated single-owner state or take over unmanaged files during shared-root migration.
 
 ## Boundaries
 

docs/integrations/s3.md

@@ -52,7 +52,7 @@ The adapter uses these S3 operations:
 
 Writes buffer the input and set `ContentLength`. If no content type is supplied by the caller, the adapter infers a content type from the logical path.
 
-Normal replacement and failed-write cleanup delete only managed output objects plus `.distributor.json`. Merge publication retains omitted managed objects and may overwrite existing managed objects. Forced replacement deletes objects under the bounded destination bundle prefix. The backend does not manage bucket versioning, lifecycle rules, object lock, or delete markers.
+Normal replacement and failed-write cleanup delete only managed output objects plus `.distributor.json`. Same-source merge publication retains omitted managed objects and may overwrite existing managed objects. Takeover replacement does not retain omitted outputs through merge reconciliation. Forced replacement deletes objects under the bounded destination bundle prefix. The backend does not manage bucket versioning, lifecycle rules, object lock, or delete markers.
 
 ## Error Mapping
 

docs/integrations/ssh-sftp.md

@@ -54,7 +54,7 @@ The configured `path` is the backend root. All source discovery, destination pat
 
 The adapter rejects symlink ancestors for reads and writes. Reads require regular files. Writes create parent directories and prefer atomic temp-file-plus-rename writes when requested. Walk output is sorted through the shared storage walker.
 
-Managed cleanup and normal replacement delete only managed output paths plus `.distributor.json`. Merge publication retains omitted managed paths and may overwrite existing managed paths. Forced replacement deletes the bounded destination bundle path.
+Managed cleanup and normal replacement delete only managed output paths plus `.distributor.json`. Same-source merge publication retains omitted managed paths and may overwrite existing managed paths. Takeover replacement does not retain omitted outputs through merge reconciliation. Forced replacement deletes the bounded destination bundle path.
 
 ## Boundaries
 

docs/internal/app.md

@@ -20,7 +20,7 @@ User-facing command parsing stays in `internal/cli`, including `reconcile-state`
 
 ## Config Fields Used
 
-The package consumes the loaded `config.Config`: `server.http`, `secrets.directory`, pipeline ids, source and destination backend fields, validation policy, publish policy, transform policy, path mapping, links, state policy, reconciliation policy, retention policy, and transfer policy.
+The package consumes the loaded `config.Config`: `server.http`, `secrets.directory`, pipeline ids, source and destination backend fields, validation policy, publish policy, transform policy, path mapping, links, state policy, reconciliation policy, takeover policy, retention policy, and transfer policy.
 
 Config fields are validated and defaulted by `internal/config` before app workflows use them.
 
@@ -32,7 +32,7 @@ The app layer registers default transforms, including Markdown-to-HTML, and supp
 
 ## State And Manifest Behavior
 
-Run workflows discover and validate source bundles through `internal/bundle`. Destination state actions are prepared and written through `internal/publish` and `internal/state`; the app layer records report projections of those actions and results.
+Run workflows discover and validate source bundles through `internal/bundle`. Destination state actions are prepared and written through `internal/publish` and `internal/state`; the app layer records report projections of those actions and results. Run summaries count takeover replacements separately from ordinary older-state replacement and explicit forced replacement.
 
 Reconcile-state workflows load one configured pipeline/destination selector, open that destination root, parse the root `.distributor.json`, and report missing managed output records plus unmanaged storage entries. Managed output existence checks use storage `Stat`; unmanaged reporting uses bounded storage `Walk` and excludes `.distributor.json` plus all paths already recorded as managed. Apply mode removes missing managed output records from state and rewrites valid state only; dry-run reports the same repair without writing. Text output reports `changed`, `would_change`, or `unchanged`; JSON output uses the shared app envelope. It does not validate output digests, delete destination files, adopt unmanaged files, or rewrite invalid or mismatched state.
 

docs/internal/config.md

@@ -18,7 +18,7 @@ The canonical user-facing config reference is `docs/config.md`.
 
 ## Config Fields Used
 
-The package defines all user-visible config fields: `server.http`, `secrets`, `pipelines`, source and destination backend fields, validation policy, publish policy, transform policy, path mapping, links, state policy, reconciliation policy, retention policy, and transfer policy.
+The package defines all user-visible config fields: `server.http`, `secrets`, `pipelines`, source and destination backend fields, validation policy, publish policy, transform policy, path mapping, links, state policy, reconciliation policy, takeover policy, retention policy, and transfer policy.
 
 ## Adapters Used
 
@@ -26,7 +26,7 @@ No external storage adapters are used directly. The package exposes normalized c
 
 ## State And Manifest Behavior
 
-The package does not parse source manifests or destination state. It validates config values that later affect manifest validation and destination state, such as publish/transform combinations, links, state policy, reconciliation policy, retention policy, transfer policy, backend roots, S3 prefix shape, and HTTP upload source settings.
+The package does not parse source manifests or destination state. It validates config values that later affect manifest validation and destination state, such as publish/transform combinations, links, state policy, reconciliation policy, takeover policy, retention policy, transfer policy, backend roots, S3 prefix shape, and HTTP upload source settings.
 
 ## Skip And Resume Behavior
 

docs/internal/publish.md

@@ -8,9 +8,9 @@ Audience: developers and LLM coding agents changing `internal/publish`.
 
 ## Inputs And Outputs
 
-Inputs are a source bundle, source backend, destination backend, pipeline id, destination id, destination bundle path, path mapping mode, publish policy, transform policy, optional link policy, state policy, reconciliation policy, transformer resolver, transfer policy, distributor version, and force flag.
+Inputs are a source bundle, source backend, destination backend, pipeline id, destination id, destination bundle path, path mapping mode, publish policy, transform policy, optional link policy, state policy, reconciliation policy, takeover policy, transformer resolver, transfer policy, distributor version, and force flag.
 
-Output from planning is a `Plan` with action, reason, destination identity, selected outputs, state mode, owner scope, reconciliation mode, optional existing single-owner or shared-root state, optional primary URL, and force metadata. Shared-root plans also expose other-owner outputs to preserve, current-owner outputs retained by merge, current-owner outputs deleted by replace, and current-owner outputs to write. Execution writes selected source outputs, generated outputs, and `.distributor.json` for executable publish or replacement actions.
+Output from planning is a `Plan` with action, reason, destination identity, selected outputs, state mode, owner scope, reconciliation mode, takeover mode, optional existing single-owner or shared-root state, optional primary URL, and force metadata. Shared-root plans also expose other-owner outputs to preserve, taken-over outputs, current-owner outputs retained by merge, current-owner outputs deleted by replace or takeover, and current-owner outputs to write. Execution writes selected source outputs, generated outputs, and `.distributor.json` for executable publish or replacement actions.
 
 ## Boundaries
 
@@ -20,7 +20,7 @@ External destination state semantics are documented in `docs/integrations/destin
 
 ## Config Fields Used
 
-The package consumes already-defaulted config values for destination `publish`, `transform`, `links`, `state`, `reconciliation`, `transfer`, and path mapping mode. It uses `config.ValidatePublishTransformPolicy` for publish/transform consistency.
+The package consumes already-defaulted config values for destination `publish`, `transform`, `links`, `state`, `reconciliation`, `takeover`, `transfer`, and path mapping mode. It uses `config.ValidatePublishTransformPolicy` for publish/transform consistency.
 
 ## Adapters Used
 
@@ -28,15 +28,15 @@ The package depends on `internal/storage.Backend` for source and destination IO,
 
 ## State And Manifest Behavior
 
-Planning inspects destination state through `internal/state`, compares it with the source manifest, and maps comparison outcomes plus transfer policy into actions: `publish_new`, `replace_older`, `force_replace`, `skip_same`, `skip_destination_newer`, `fail_conflict`, or `fail_unmanaged`.
+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. 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. Replacement-mode shared-root updates remove only current-owner omitted outputs and preserve unrelated owners. Merge-mode 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.
 
@@ -65,7 +65,7 @@ Execution fails on delete, read, transform output, unmanaged merge path collisio
 - Replacement reconciliation deletes only managed paths recorded in existing state plus `.distributor.json` for single-owner state, and only current-owner omitted outputs for shared-root state.
 - Merge reconciliation never adopts unmanaged content.
 - Merge state output records are cumulative for the single owner.
-- Shared-root planning is owner-scoped and preserves unrelated owner outputs.
+- Shared-root planning is owner-scoped, preserves unrelated owner outputs, and records taken-over managed output paths separately from unrelated owners.
 - Shared-root execution writes owner-scoped changes without deleting unrelated owners.
 - Forced replacement deletes only within the supplied destination bundle path.
 - Destination state is written after selected outputs are written.

docs/internal/state.md

@@ -44,7 +44,7 @@ Shared-root helper projections preserve output `created_at` for existing managed
 
 ## Skip And Resume Behavior
 
-Comparison is pure. It returns outcomes for absent state, unmanaged content, invalid state, pipeline/destination mismatch, same source manifest, older destination, newer destination, same-created digest conflict, and different source id conflict. It does not decide whether to skip, replace, force, or fail; publish planning maps outcomes to actions.
+Comparison is pure. It returns outcomes for absent state, unmanaged content, invalid state, pipeline/destination mismatch, same source manifest, older destination, newer destination, same-created digest conflict, and different source id conflict. Shared-root helpers expose structured output ownership conflicts. State code does not decide whether to skip, replace, take over, force, or fail; publish planning maps outcomes and conflicts to actions.
 
 Shared-root comparison is owner-scoped. It compares only the owner keyed by the current pipeline id and destination id, treats an absent owner as absent destination state for that owner, and can compare compatible single-owner state for the current owner without converting unrelated single-owner state.
 
@@ -74,7 +74,7 @@ Parsing rejects invalid JSON, trailing data, missing required fields, invalid ti
 - Newly written single-owner state uses schema version `2`.
 - Schema version `1` state remains readable as replacement-mode single-owner state.
 - Schema version `3` shared-root state is parsed and validated without converting unrelated single-owner state.
-- Shared-root owner updates preserve unrelated owners and reject planned path collisions with other owners.
+- Shared-root owner updates preserve unrelated owners. Publish planning removes taken-over output records before owner replacement when takeover policy allows a managed path to move owners.
 - Missing-output repair helpers preserve unrelated owner records and outputs.
 - Prune planning uses output `updated_at` and preserves unrelated shared-root owners.
 - Generated outputs always record a transform id.

docs/operations.md

@@ -67,23 +67,26 @@ Published destination bundle paths contain `.distributor.json`. See [Destination
 - No destination state and no destination content: publish new outputs.
 - Matching destination state: skip as already published.
 - Older destination state for the same source id: replace if transfer policy allows it.
-- Newer destination state: skip by default.
-- Invalid destination state, identity mismatch, different source id, or same-created digest mismatch: fail by default.
+- Newer destination state for the same source id: skip by default.
+- 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, 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 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.
 
 `reconciliation.mode: merge` retains prior managed outputs that are omitted from the new plan. It overwrites planned paths only when those paths are already recorded in existing state as managed. If a newly planned path already exists in storage but is not recorded in state, publication fails as an unmanaged path collision. The new state `outputs` array is the cumulative managed output set.
 
 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.
 
@@ -150,12 +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.
+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`: 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.
@@ -163,7 +167,7 @@ Review these action labels before publishing:
 
 Fixed destinations add fixed-path warnings during dry runs, including the selected source bundle and replacement warnings when the destination root would be replaced.
 
-JSON output includes warnings, pipeline summaries, destination action records, output records, URLs when configured, final counters, and partial failure details. Fatal setup failures such as unreadable config or invalid secrets do not produce a JSON result document.
+Text and JSON summaries count `publish_new`, `replace_older`, `replace_takeover`, `force_replace`, skipped, and failed destinations separately. Takeover action records include the configured takeover mode and the conflict reason. JSON output includes warnings, pipeline summaries, destination action records, output records, URLs when configured, final counters, and partial failure details. Fatal setup failures such as unreadable config or invalid secrets do not produce a JSON result document.
 
 ## Forced Replacement Workflow
 
@@ -267,7 +271,7 @@ S3 execution uses the AWS SDK for Go v2. See [S3-Compatible Storage Integration]
 
 When explicit S3 credential variable names are configured, both must resolve to non-empty values through the process environment or `secrets.directory`. When omitted, the AWS SDK default credential chain is used as-is.
 
-Normal single-owner replacement and failed-write cleanup delete only managed output objects recorded in `.distributor.json` plus the state object. Shared-root replacement deletes only current-owner omitted output objects and rewrites the shared state object. Merge publication retains omitted managed objects and may overwrite existing managed objects. Forced replacement deletes objects under the bounded destination bundle prefix. Distributor does not manage bucket versioning or delete markers.
+Normal single-owner replacement and failed-write cleanup delete only managed output objects recorded in `.distributor.json` plus the state object. Shared-root replacement deletes only current-owner omitted output objects and rewrites the shared state object. Same-source merge publication retains omitted managed objects and may overwrite existing managed objects. Takeover replacement does not retain omitted outputs through merge reconciliation. Forced replacement deletes objects under the bounded destination bundle prefix. Distributor does not manage bucket versioning or delete markers.
 
 ## Secrets Operation
 
@@ -282,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/policy/architecture.md

@@ -171,7 +171,7 @@ Destination comparison rules are based on `.distributor.json`:
 - Existing state has the same source id and an older source `created`: replace, subject to destructive-operation safety rules.
 - Existing state has the same source id and a newer source `created`: skip because the destination is newer than the source.
 - Existing state has the same source id and same `created` but different digest: fail as a conflict.
-- Existing state has a different source id: fail as a conflict.
+- Existing valid managed state has a different source id, pipeline id, destination id, or shared-root output owner: replace only when destination `takeover.mode` permits that ownership transfer; otherwise fail as a conflict.
 
 For older destination state, destination `reconciliation.mode` controls output updates. `replace` rewrites the managed output set to match the new plan. `merge` retains omitted managed outputs, overwrites only existing managed paths, and rejects planned paths that collide with unmanaged storage content.
 
@@ -264,6 +264,7 @@ Pipeline configuration should express:
 - per-destination transform policy;
 - per-destination public link policy;
 - validation behavior;
+- per-destination takeover behavior;
 - destination conflict/replacement behavior.
 
 ## Modules and Registries

docs/policy/development.md

@@ -18,7 +18,7 @@ Use it with `docs/policy/architecture.md` and `docs/policy/documentation.md`.
 - `internal/adapters/ssh`: SSH/SFTP backend.
 - `internal/adapters/s3`: S3-compatible object storage backend.
 - `internal/storage/fake`: in-memory backend for tests.
-- `internal/publish`: destination inspection, output planning, reconciliation, execution, managed cleanup, and explicit forced replacement.
+- `internal/publish`: destination inspection, output planning, takeover planning, reconciliation, execution, managed cleanup, and explicit forced replacement.
 - `internal/transform`: transform interface and registry.
 - `internal/transform/markdown`: Markdown-to-HTML transform.
 - `internal/notify`: notification interface and current no-op notifier.

docs/roadmap/implementation.md

@@ -1,14 +1,14 @@
 # Managed Destination Takeover Implementation Roadmap
 
-This is the active staged implementation plan for
-`docs/roadmap/takeover.md`. The feature roadmap defines the target policy and
-end-state semantics; this document defines the implementation sequence for an
-LLM coding agent to follow stage by stage.
+This is the completed staged implementation plan for the takeover feature. The
+detailed feature roadmap was removed after implementation; current behavior is
+documented outside `docs/roadmap/`. This document records the implementation
+sequence used by LLM coding agents stage by stage.
 
-Future behavior must remain under `docs/roadmap/` until each stage is
-implemented. Preparatory internal stages should not update user-facing current
-docs. Current-behavior docs should be updated when takeover behavior is wired
-for operator-facing use.
+Future behavior must remain under `docs/roadmap/` until implemented.
+Preparatory internal stages should not update user-facing current docs.
+Current-behavior docs should be updated when behavior is wired for
+operator-facing use.
 
 ## Current Baseline
 
@@ -55,7 +55,8 @@ changing publish behavior.
 
 Source roadmap reference:
 
-- `docs/roadmap/takeover.md`: Configuration, Policy Semantics, Safety Rules.
+- Completed takeover feature roadmap: Configuration, Policy Semantics, Safety
+  Rules.
 
 Implementation scope:
 
@@ -96,7 +97,7 @@ takeover eligibility without parsing human-readable reason strings.
 
 Source roadmap reference:
 
-- `docs/roadmap/takeover.md`: Policy Semantics, Relationship To Existing
+- Completed takeover feature roadmap: Policy Semantics, Relationship To Existing
   Policies.
 
 Implementation scope:
@@ -140,7 +141,7 @@ Goal: implement `takeover.mode` for single-owner destination state.
 
 Source roadmap reference:
 
-- `docs/roadmap/takeover.md`: Policy Semantics, Publish Planning, Destination
+- Completed takeover feature roadmap: Policy Semantics, Publish Planning, Destination
   State Results, Safety Rules.
 
 Implementation scope:
@@ -207,7 +208,7 @@ conflicts.
 
 Source roadmap reference:
 
-- `docs/roadmap/takeover.md`: Policy Semantics, Publish Planning, Destination
+- Completed takeover feature roadmap: Policy Semantics, Publish Planning, Destination
   State Results, Safety Rules.
 
 Implementation scope:
@@ -262,7 +263,7 @@ feature.
 
 Source roadmap reference:
 
-- `docs/roadmap/takeover.md`: Documentation Impact, Publish Planning,
+- Completed takeover feature roadmap: Documentation Impact, Publish Planning,
   Relationship To Existing Policies.
 
 Implementation scope:
@@ -313,7 +314,7 @@ after implementation.
 
 Source roadmap reference:
 
-- `docs/roadmap/takeover.md`: Goals, Non-Goals, Safety Rules.
+- Completed takeover feature roadmap: Goals, Non-Goals, Safety Rules.
 
 Implementation scope:
 
@@ -328,9 +329,8 @@ Implementation scope:
 - Add tests for archive-style `preserve_relative` destinations using
   `takeover.mode: same_source` where strict source identity is desired.
 - Run final consistency searches.
-- Once behavior and docs are complete, either remove `docs/roadmap/takeover.md`
-  or reduce it to future-only material according to the documentation policy.
-  If no future takeover work remains, remove the roadmap file.
+- Remove the completed detailed takeover roadmap when no future takeover work
+  remains.
 
 Tests:
 

docs/roadmap/takeover.md

@@ -1,229 +0,0 @@
-# Managed Destination Takeover Roadmap
-
-This roadmap defines planned replacement policy for distributor-managed
-destination content when the existing destination state does not describe the
-same source bundle as the current publication.
-
-Current behavior is intentionally conservative. A destination may be replaced
-automatically when its existing state describes the same source id and an older
-source creation time. If the existing state belongs to a different source id,
-pipeline id, destination id, or shared-root owner, `distributor` reports a
-conflict unless an explicit forced replacement workflow is selected where
-supported.
-
-That strict default was useful for the initial implementation, but it is too
-rigid for producers that intentionally publish newer generated content to the
-same destination path from different producer tasks or bundle ids. `distributor`
-is primarily a distribution tool: by default, a valid producer request should
-publish the requested bundle unless doing so would claim unmanaged content or
-cross an ownership boundary that is likely to surprise the operator.
-
-## Goals
-
-- Add a destination-level policy that defines when normal, non-force publication
-  may replace existing distributor-managed content whose source or owner
-  identity differs from the current source.
-- Default the policy to `same_pipeline`, so a pipeline may distribute the latest
-  valid bundle to its configured destination paths without requiring every
-  producer task to reuse one source manifest id.
-- Preserve strong safety boundaries for unmanaged content, invalid destination
-  state, and cross-pipeline ownership by default.
-- Keep reconciliation policy focused on how managed output sets are updated
-  after replacement is allowed.
-- Keep transfer policy focused on same-source comparison outcomes and explicit
-  forced replacement behavior.
-- Preserve clear dry-run and JSON output so takeover behavior is visible before
-  files are changed.
-
-## Non-Goals
-
-- Do not allow automatic adoption of unmanaged destination files.
-- Do not allow invalid `.distributor.json` state to be overwritten without the
-  existing explicit force workflow.
-- Do not make producers select destination ownership or takeover policy through
-  source manifests or HTTP upload requests.
-- Do not replace `state.mode`, `reconciliation.mode`, `path_mapping.mode`, or
-  `transfer` policy.
-- Do not introduce broad synchronization behavior outside configured
-  destination bundle paths.
-
-## Configuration
-
-Add destination-level takeover policy:
-
-```yaml
-destinations:
-  - id: latest
-    backend: local
-    path: /srv/reports/weather/latest/tomorrow
-    path_mapping:
-      mode: fixed
-    takeover:
-      mode: same_pipeline
-```
-
-Accepted `takeover.mode` values:
-
-- `same_pipeline`: default. Normal publication may replace valid
-  distributor-managed state or managed output paths owned by the same pipeline,
-  even when the previous source id or destination id differs.
-- `same_source`: normal publication may replace only when the destination state
-  has the same source manifest id as the current source. This is closest to the
-  current strict managed replacement behavior.
-- `any_managed`: normal publication may replace any valid distributor-managed
-  state or managed output path in the selected destination bundle path,
-  regardless of pipeline id, destination id, or source id.
-- `never`: normal publication never performs takeover replacement for identity
-  or source conflicts. Same-source older/newer behavior remains governed by
-  `transfer`, and explicit forced replacement remains governed by `--force` and
-  `transfer.on_conflict`.
-
-The field is destination-local because different destinations from the same
-pipeline can have different ownership expectations. Archive destinations often
-want stricter behavior, while latest-style fixed destinations often want
-`same_pipeline` or, after explicit operator review, `any_managed`.
-
-## Policy Semantics
-
-Takeover policy applies only after destination state has been parsed and
-validated as distributor-managed state.
-
-For single-owner state:
-
-- `same_pipeline` permits replacement when the existing state `pipeline_id`
-  matches the current pipeline id. The previous `destination_id` and source id
-  may differ.
-- `same_source` permits replacement only when the existing source manifest id
-  matches the current source manifest id.
-- `any_managed` permits replacement of any valid single-owner destination state.
-- `never` does not permit takeover replacement.
-
-For shared-root state:
-
-- Takeover is evaluated per managed output path that collides with the current
-  plan and per current owner record when one already exists.
-- `same_pipeline` permits the current destination owner to take over output
-  paths owned by another destination under the same pipeline.
-- `same_source` permits takeover only when the colliding output owner records
-  the same source manifest id as the current source.
-- `any_managed` permits takeover of colliding managed output paths owned by any
-  owner in the shared root.
-- `never` preserves the current owner-scoped conflict behavior.
-
-Takeover never applies to:
-
-- destination content with no valid `.distributor.json`;
-- invalid destination state;
-- planned paths that exist in storage but are not recorded as managed;
-- source digest mismatches or invalid source manifests;
-- destination newer/same-created digest comparisons for the same source id,
-  except where existing transfer policy already permits replacement.
-
-## Relationship To Existing Policies
-
-`takeover.mode` answers this question:
-
-> May this destination normally replace valid managed content whose owner or
-> source identity differs from the current publication?
-
-`reconciliation.mode` continues to answer:
-
-> Once replacement is allowed, should omitted managed outputs be removed
-> (`replace`) or retained (`merge`)?
-
-`transfer` continues to answer:
-
-> What should happen for same-source comparisons, destination-newer state, and
-> explicit forced conflict replacement?
-
-`--force` remains an operator workflow for exceptional cases. Takeover policy is
-not a replacement for forced replacement of unmanaged content or invalid state.
-
-## Publish Planning
-
-Add a new planned action for successful takeover replacement, tentatively named
-`replace_takeover`.
-
-`replace_takeover` should execute through the same bounded managed replacement
-machinery as other managed replacements, with one additional source-identity
-safety rule: when takeover crosses source id or owner identity, omitted outputs
-from the previous source must not be retained merely because the destination is
-configured with `reconciliation.mode: merge`. Retaining omitted outputs is valid
-only when those outputs still belong to the same source identity being
-published. This keeps destination state from claiming that outputs derived from
-an old source belong to the new source manifest.
-
-- single-owner `replace` deletes only managed outputs recorded in existing
-  state plus `.distributor.json`, then writes the new outputs and state;
-- single-owner `merge` retains omitted managed outputs only for same-source
-  updates; cross-source takeover behaves as a managed replacement for the
-  affected owner;
-- shared-root `replace` deletes only output records taken over by the current
-  owner or omitted from the current owner according to the shared-root plan;
-- shared-root `merge` preserves unrelated non-conflicting managed outputs and
-  updates ownership for paths explicitly taken over, but does not retain omitted
-  outputs from a different source under the taking-over owner.
-
-Text output, JSON output, dry-run output, and summary counters should identify
-takeover replacements separately from `replace_older` and `force_replace`.
-
-## Destination State Results
-
-After a successful takeover, destination state must describe the current
-pipeline, destination, source manifest, outputs, links, reconciliation policy,
-and state mode.
-
-For single-owner state, the destination bundle path has one current owner after
-publication: the publishing pipeline and destination.
-
-For shared-root state, only the affected owner records and output records are
-changed. Unrelated owners and non-conflicting outputs remain recorded.
-
-No source manifest schema change is required. No destination state schema change
-is expected unless implementation discovers that takeover metadata must be
-persisted for recovery or audit. The preferred initial design is to make
-takeover a planning decision, not a new persisted state concept.
-
-## Documentation Impact
-
-When this feature is implemented, current-behavior docs should explain the new
-policy where operators and maintainers already look for publication safety,
-state comparison, and configuration behavior:
-
-- `docs/config.md`: document `takeover.mode`, default `same_pipeline`, accepted
-  values, examples for archive and latest destinations, and interaction with
-  `transfer`, `reconciliation`, and `--force`.
-- `docs/operations.md`: explain takeover dry-runs, normal replacement safety,
-  shared-root behavior, and recovery guidance.
-- `docs/troubleshooting.md`: update conflict guidance for
-  `destination source id differs from source`, pipeline/destination mismatch,
-  and shared-root ownership conflicts.
-- `docs/integrations/destination-state.md`: describe how takeover affects
-  comparison and state rewriting without changing the source manifest contract.
-- `docs/internal/publish.md`, `docs/internal/state.md`, and
-  `docs/internal/config.md`: document package responsibilities and invariants.
-- `docs/policy/architecture.md`: clarify that the default posture is to
-  distribute valid producer bundles through configured destinations while
-  protecting unmanaged content and cross-pipeline ownership by default.
-- `docs/policy/development.md`: add `takeover.mode` to the list of config
-  changes that must be validated and documented when publish safety changes.
-
-## Safety Rules
-
-- Takeover must be dry-runnable.
-- Takeover must operate only inside the resolved destination bundle path.
-- Takeover must never delete parent paths, sibling paths, or unmanaged files.
-- Takeover must never claim storage content that is not recorded in valid
-  destination state.
-- `any_managed` should be documented as an intentional cross-pipeline ownership
-  policy for tightly controlled destinations, not as the recommended default for
-  archives.
-- Archive-style destinations should generally use `same_source` if each bundle
-  id represents a distinct immutable artifact.
-
-## Implementation Reference
-
-The staged implementation plan for this feature lives in
-`docs/roadmap/implementation.md`. This document remains the feature contract and
-policy reference; implementation sequencing, package-by-package work, and test
-commands belong in the implementation roadmap.

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.
+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. To replace the existing state, 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:
 
@@ -354,7 +354,7 @@ Diagnostic:
 find <destination-path> -maxdepth 2 -print
 ```
 
-Safe fix: inspect the destination bundle path printed in the error. `distributor` attempts to remove outputs from the failed attempt, but operators should verify the destination before retrying. In merge mode, previously managed retained or overwritten outputs may remain intentionally. Rerun `--dry-run` before publishing again.
+Safe fix: inspect the destination bundle path printed in the error. `distributor` attempts to remove outputs from the failed attempt, but operators should verify the destination before retrying. In same-source merge mode, previously managed retained or overwritten outputs may remain intentionally. Rerun `--dry-run` before publishing again.
 
 Reference: [Operations](operations.md#destination-state-and-retry-behavior).
 

internal/app/run_destination.go

@@ -71,6 +71,7 @@ func processDestinationSelection(ctx context.Context, request runDestinationRequ
 		Links:                 request.destination.Links,
 		State:                 request.destination.State,
 		Reconciliation:        request.destination.Reconciliation,
+		Takeover:              request.destination.Takeover,
 		Transformers:          request.transforms,
 		Transfer:              request.destination.Transfer,
 		DistributorVersion:    Version,

internal/app/run_notify.go

@@ -6,7 +6,7 @@ import (
 )
 
 func shouldNotify(action publish.Action) bool {
-	return action == publish.ActionPublishNew || action == publish.ActionReplaceOlder || action == publish.ActionForceReplace
+	return action == publish.ActionPublishNew || action == publish.ActionReplaceOlder || action == publish.ActionReplaceTakeover || action == publish.ActionForceReplace
 }
 
 func notifyEvent(plan publish.Plan) notify.Event {

internal/app/run_output.go

@@ -60,7 +60,7 @@ func writeRunActionLine(w io.Writer, action RunActionRecord) {
 		fmt.Fprintf(w, "  - bundle=%s destination=%s backend=%s%s action=error reason=%q\n", action.BundlePath, destinationID, action.Backend, pathMappingRecordSummary(action), action.Reason)
 		return
 	}
-	fmt.Fprintf(w, "  - bundle=%s destination=%s backend=%s%s action=%s outputs=%s reason=%q\n", action.BundlePath, action.DestinationID, action.Backend, pathMappingRecordSummary(action), action.Action, outputRecordSummary(action.Outputs), action.Reason)
+	fmt.Fprintf(w, "  - bundle=%s destination=%s backend=%s%s action=%s%s outputs=%s reason=%q\n", action.BundlePath, action.DestinationID, action.Backend, pathMappingRecordSummary(action), action.Action, takeoverModeRecordSummary(action), outputRecordSummary(action.Outputs), action.Reason)
 }
 
 func pathMappingRecordSummary(action RunActionRecord) string {
@@ -70,6 +70,13 @@ func pathMappingRecordSummary(action RunActionRecord) string {
 	return fmt.Sprintf(" path_mapping=fixed target=%s", action.DestinationPath)
 }
 
+func takeoverModeRecordSummary(action RunActionRecord) string {
+	if action.TakeoverMode == "" {
+		return ""
+	}
+	return fmt.Sprintf(" takeover_mode=%s", action.TakeoverMode)
+}
+
 func outputRecordSummary(outputs []RunOutputRecord) string {
 	if len(outputs) == 0 {
 		return "none"
@@ -137,6 +144,7 @@ type RunActionRecord struct {
 	DestinationPath string            `json:"destination_path"`
 	PathMapping     string            `json:"path_mapping,omitempty"`
 	Action          string            `json:"action"`
+	TakeoverMode    string            `json:"takeover_mode,omitempty"`
 	PrimaryURL      string            `json:"primary_url,omitempty"`
 	Reason          string            `json:"reason,omitempty"`
 	Outputs         []RunOutputRecord `json:"outputs"`
@@ -181,12 +189,20 @@ func runActionFromPlan(backend string, plan publish.Plan, planErr error) RunActi
 		DestinationPath: storage.DisplayPath(plan.DestinationBundlePath),
 		PathMapping:     plan.PathMapping,
 		Action:          string(plan.Action),
+		TakeoverMode:    takeoverModeForAction(plan),
 		PrimaryURL:      plan.PrimaryURL,
 		Reason:          plan.Reason,
 		Outputs:         runOutputsFromPlan(plan.Outputs),
 	}
 }
 
+func takeoverModeForAction(plan publish.Plan) string {
+	if plan.Action != publish.ActionReplaceTakeover {
+		return ""
+	}
+	return plan.TakeoverMode
+}
+
 func errorAction(pipelineID, destinationID, backend, bundlePath string, err error) RunActionRecord {
 	return RunActionRecord{
 		PipelineID:      pipelineID,

internal/app/run_selection.go

@@ -63,10 +63,13 @@ func fixedPathSelectionWarning(pipelineID, destinationID string, selections []de
 }
 
 func isDestructiveFixedPathAction(action publish.Action) bool {
-	return action == publish.ActionReplaceOlder || action == publish.ActionForceReplace
+	return action == publish.ActionReplaceOlder || action == publish.ActionReplaceTakeover || action == publish.ActionForceReplace
 }
 
 func fixedPathReplacementWarning(plan publish.Plan) OutputWarning {
+	if plan.Action == publish.ActionReplaceTakeover {
+		return OutputWarning{Message: fmt.Sprintf("pipeline=%s destination=%s path_mapping=fixed action=%s takeover_mode=%s replaces destination root for selected_bundle=%s reason=%q", plan.PipelineID, plan.DestinationID, plan.Action, plan.TakeoverMode, storage.DisplayPath(plan.BundlePath), plan.Reason)}
+	}
 	return OutputWarning{Message: fmt.Sprintf("pipeline=%s destination=%s path_mapping=fixed action=%s replaces destination root for selected_bundle=%s", plan.PipelineID, plan.DestinationID, plan.Action, storage.DisplayPath(plan.BundlePath))}
 }
 

internal/app/run_summary.go

@@ -11,6 +11,7 @@ type runSummary struct {
 	planned         int
 	publishNew      int
 	replaceOlder    int
+	replaceTakeover int
 	forceReplace    int
 	skipped         int
 	failures        int
@@ -24,6 +25,8 @@ func (s *runSummary) recordPlan(action publish.Action) {
 		s.publishNew++
 	case publish.ActionReplaceOlder:
 		s.replaceOlder++
+	case publish.ActionReplaceTakeover:
+		s.replaceTakeover++
 	case publish.ActionForceReplace:
 		s.forceReplace++
 	case publish.ActionSkipSame, publish.ActionSkipDestinationNewer:
@@ -44,6 +47,7 @@ type RunSummaryCounters struct {
 	Planned         int    `json:"planned"`
 	PublishNew      int    `json:"publish_new"`
 	ReplaceOlder    int    `json:"replace_older"`
+	ReplaceTakeover int    `json:"replace_takeover"`
 	ForceReplace    int    `json:"force_replace"`
 	Skipped         int    `json:"skipped"`
 	Failed          int    `json:"failed"`
@@ -52,7 +56,7 @@ type RunSummaryCounters struct {
 }
 
 func (s RunSummaryCounters) Line() string {
-	return fmt.Sprintf("Final status: %s planned=%d publish_new=%d replace_older=%d force_replace=%d skipped=%d failed=%d dry_run=%t fixed_path=%d", s.Status, s.Planned, s.PublishNew, s.ReplaceOlder, s.ForceReplace, s.Skipped, s.Failed, s.DryRun, s.FixedPath)
+	return fmt.Sprintf("Final status: %s planned=%d publish_new=%d replace_older=%d replace_takeover=%d force_replace=%d skipped=%d failed=%d dry_run=%t fixed_path=%d", s.Status, s.Planned, s.PublishNew, s.ReplaceOlder, s.ReplaceTakeover, s.ForceReplace, s.Skipped, s.Failed, s.DryRun, s.FixedPath)
 }
 
 func (s runSummary) Result() RunSummaryCounters {
@@ -65,6 +69,7 @@ func (s runSummary) Result() RunSummaryCounters {
 		Planned:         s.planned,
 		PublishNew:      s.publishNew,
 		ReplaceOlder:    s.replaceOlder,
+		ReplaceTakeover: s.replaceTakeover,
 		ForceReplace:    s.forceReplace,
 		Skipped:         s.skipped,
 		Failed:          s.failures,

internal/app/run_test.go

@@ -43,7 +43,7 @@ func TestRunDryRunPrintsConfigSummary(t *testing.T) {
 		"Configured pipelines: 1",
 		"- pipeline=reports source=local bundles=1 destinations=archive",
 		"bundle=. destination=archive backend=local action=publish_new outputs=report.md,summary.txt",
-		"Final status: ok planned=1 publish_new=1 replace_older=0 force_replace=0 skipped=0 failed=0 dry_run=true",
+		"Final status: ok planned=1 publish_new=1 replace_older=0 replace_takeover=0 force_replace=0 skipped=0 failed=0 dry_run=true",
 	} {
 		if !strings.Contains(output, want) {
 			t.Fatalf("Run() output = %q, want substring %q", output, want)
@@ -592,8 +592,9 @@ func TestRunFixedPathDryRunWarnsForReplacement(t *testing.T) {
 	}
 	output := stdout.String()
 	for _, want := range []string{
-		"Warning: pipeline=reports destination=archive path_mapping=fixed action=replace_older replaces destination root for selected_bundle=new",
-		"bundle=new destination=archive backend=local path_mapping=fixed target=. action=replace_older",
+		"Warning: pipeline=reports destination=archive path_mapping=fixed action=replace_takeover takeover_mode=same_pipeline replaces destination root for selected_bundle=new reason=\"destination source id differs from source\"",
+		"bundle=new destination=archive backend=local path_mapping=fixed target=. action=replace_takeover takeover_mode=same_pipeline outputs=report.md,summary.txt reason=\"destination source id differs from source\"",
+		"replace_takeover=1",
 	} {
 		if !strings.Contains(output, want) {
 			t.Fatalf("stdout = %q, want substring %q", output, want)
@@ -602,6 +603,61 @@ func TestRunFixedPathDryRunWarnsForReplacement(t *testing.T) {
 	testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nOld.\n")
 }
 
+func TestRunJSONIncludesTakeoverActionAndSummary(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	writeSourceBundle(t, sourceRoot, "old", testBundleOptions{
+		ID:      "reports.old",
+		Created: testutil.DefaultCreated,
+		Files: []testFile{
+			{Path: "report.md", Data: "# Report\nOld.\n"},
+			{Path: "summary.txt", Data: "Old summary\n"},
+		},
+	})
+	configPath := testutil.WriteLocalConfigWithPathMapping(t, sourceRoot, destinationRoot, config.PathMappingFixed)
+	if err := Run(context.Background(), RunOptions{ConfigPath: configPath}); err != nil {
+		t.Fatalf("first Run() error = %v", err)
+	}
+	writeSourceBundle(t, sourceRoot, "new", testBundleOptions{
+		ID:      "reports.new",
+		Created: testutil.DefaultCreated.Add(time.Hour),
+		Files: []testFile{
+			{Path: "report.md", Data: "# Report\nNew.\n"},
+			{Path: "summary.txt", Data: "New summary\n"},
+		},
+	})
+
+	var stdout bytes.Buffer
+	err := Run(context.Background(), RunOptions{
+		ConfigPath:   configPath,
+		DryRun:       true,
+		Stdout:       &stdout,
+		OutputFormat: OutputFormatJSON,
+	})
+	if err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	result := decodeAppResult(t, stdout.String())
+	actions, ok := result["actions"].([]any)
+	if !ok || len(actions) != 1 {
+		t.Fatalf("actions = %#v, want one action", result["actions"])
+	}
+	action, ok := actions[0].(map[string]any)
+	if !ok {
+		t.Fatalf("action = %#v, want object", actions[0])
+	}
+	if action["action"] != "replace_takeover" || action["takeover_mode"] != "same_pipeline" || action["reason"] != "destination source id differs from source" {
+		t.Fatalf("action = %#v, want takeover action metadata", action)
+	}
+	summary, ok := result["summary"].(map[string]any)
+	if !ok {
+		t.Fatalf("summary = %#v, want object", result["summary"])
+	}
+	if summary["replace_takeover"] != float64(1) || summary["replace_older"] != float64(0) || summary["force_replace"] != float64(0) {
+		t.Fatalf("summary = %#v, want takeover counter only", summary)
+	}
+}
+
 func TestRunFixedPathReplacesOlderManagedState(t *testing.T) {
 	sourceRoot := t.TempDir()
 	destinationRoot := t.TempDir()
@@ -628,9 +684,13 @@ func TestRunFixedPathReplacesOlderManagedState(t *testing.T) {
 		},
 	})
 
-	if err := Run(context.Background(), RunOptions{ConfigPath: configPath}); err != nil {
+	var stdout bytes.Buffer
+	if err := Run(context.Background(), RunOptions{ConfigPath: configPath, Stdout: &stdout}); err != nil {
 		t.Fatalf("second Run() error = %v", err)
 	}
+	if !strings.Contains(stdout.String(), "action=replace_takeover takeover_mode=same_pipeline") {
+		t.Fatalf("stdout = %q, want same-pipeline takeover", stdout.String())
+	}
 	testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nNew.\n")
 	destinationState := readStateFile(t, filepath.Join(destinationRoot, storage.StateFileName))
 	if destinationState.Source.Manifest.ID != "reports.new" {
@@ -638,11 +698,72 @@ func TestRunFixedPathReplacesOlderManagedState(t *testing.T) {
 	}
 }
 
+func TestRunPreserveRelativeSameSourceTakeoverAllowsOwnerMismatch(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	sourceManifest := writeSourceBundle(t, sourceRoot, "daily/report", testBundleOptions{
+		ID:    "reports.same",
+		Files: []testFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
+	})
+	testutil.WriteDestinationState(t, destinationRoot, "daily/report", sourceManifest, testutil.DestinationStateOptions{
+		PipelineID: "other",
+	})
+	if err := os.WriteFile(filepath.Join(destinationRoot, "daily", "report", "report.md"), []byte("# Report\nOld.\n"), 0o600); err != nil {
+		t.Fatalf("write old report: %v", err)
+	}
+
+	var stdout bytes.Buffer
+	err := Run(context.Background(), RunOptions{
+		ConfigPath: writeSameSourcePreserveRelativeConfig(t, sourceRoot, destinationRoot),
+		Stdout:     &stdout,
+	})
+	if err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	if !strings.Contains(stdout.String(), "action=replace_takeover takeover_mode=same_source") {
+		t.Fatalf("stdout = %q, want same-source takeover", stdout.String())
+	}
+	testutil.AssertFile(t, filepath.Join(destinationRoot, "daily", "report", "report.md"), "# Report\nNew.\n")
+	destinationState := readStateFile(t, filepath.Join(destinationRoot, "daily", "report", storage.StateFileName))
+	if destinationState.PipelineID != "reports" || destinationState.Source.Manifest.ID != "reports.same" {
+		t.Fatalf("state owner/source = %s/%s source=%s, want reports/archive reports.same", destinationState.PipelineID, destinationState.DestinationID, destinationState.Source.Manifest.ID)
+	}
+}
+
+func TestRunPreserveRelativeSameSourceRefusesDifferentSource(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	sourceManifest := writeSourceBundle(t, sourceRoot, "daily/report", testBundleOptions{
+		ID:    "reports.same",
+		Files: []testFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
+	})
+	destinationManifest := sourceManifest
+	destinationManifest.ID = "reports.other"
+	testutil.WriteDestinationState(t, destinationRoot, "daily/report", destinationManifest, testutil.DestinationStateOptions{
+		PipelineID: "other",
+	})
+	if err := os.WriteFile(filepath.Join(destinationRoot, "daily", "report", "report.md"), []byte("# Report\nOld.\n"), 0o600); err != nil {
+		t.Fatalf("write old report: %v", err)
+	}
+
+	err := Run(context.Background(), RunOptions{
+		ConfigPath: writeSameSourcePreserveRelativeConfig(t, sourceRoot, destinationRoot),
+	})
+	if err == nil || !strings.Contains(err.Error(), "fail_conflict") {
+		t.Fatalf("Run() error = %v, want fail_conflict", err)
+	}
+	testutil.AssertFile(t, filepath.Join(destinationRoot, "daily", "report", "report.md"), "# Report\nOld.\n")
+	destinationState := readStateFile(t, filepath.Join(destinationRoot, "daily", "report", storage.StateFileName))
+	if destinationState.PipelineID != "other" || destinationState.Source.Manifest.ID != "reports.other" {
+		t.Fatalf("state owner/source = %s/%s source=%s, want unchanged other/archive reports.other", destinationState.PipelineID, destinationState.DestinationID, destinationState.Source.Manifest.ID)
+	}
+}
+
 func TestRunFixedPathSkipsWhenDestinationStateIsNewer(t *testing.T) {
 	sourceRoot := t.TempDir()
 	destinationRoot := t.TempDir()
 	newer := testutil.ValidManifest(testutil.BundleOptions{
-		ID:      "reports.newer",
+		ID:      "reports.same",
 		Created: testutil.DefaultCreated.Add(time.Hour),
 	})
 	writeDestinationState(t, destinationRoot, "", newer)
@@ -650,7 +771,7 @@ func TestRunFixedPathSkipsWhenDestinationStateIsNewer(t *testing.T) {
 		t.Fatalf("write existing report: %v", err)
 	}
 	writeSourceBundle(t, sourceRoot, "older", testBundleOptions{
-		ID:      "reports.older",
+		ID:      "reports.same",
 		Created: testutil.DefaultCreated,
 	})
 
@@ -721,14 +842,6 @@ func TestRunFixedPathRemoteBackendsUseBackendRoots(t *testing.T) {
 			{Path: "summary.txt", Data: "Old summary\n"},
 		},
 	})
-	writeSourceBundle(t, localSourceRoot, "new", testBundleOptions{
-		ID:      "reports.new",
-		Created: testutil.DefaultCreated.Add(time.Hour),
-		Files: []testFile{
-			{Path: "report.md", Data: "# Report\nNew.\n"},
-			{Path: "summary.txt", Data: "New summary\n"},
-		},
-	})
 	s3Destination := fake.New()
 	sshDestination := fake.New()
 	cfg := config.Config{Pipelines: []config.Pipeline{{
@@ -760,6 +873,30 @@ func TestRunFixedPathRemoteBackendsUseBackendRoots(t *testing.T) {
 	if err := runConfigWithBackendFactory(context.Background(), cfg, RunOptions{}, provider); err != nil {
 		t.Fatalf("Run() error = %v", err)
 	}
+	testutil.AssertFakeFile(t, s3Destination, "report.md", "# Report\nOld.\n")
+	testutil.AssertFakeFile(t, sshDestination, "summary.txt", "Old summary\n")
+
+	writeSourceBundle(t, localSourceRoot, "new", testBundleOptions{
+		ID:      "reports.new",
+		Created: testutil.DefaultCreated.Add(time.Hour),
+		Files: []testFile{
+			{Path: "report.md", Data: "# Report\nNew.\n"},
+			{Path: "summary.txt", Data: "New summary\n"},
+		},
+	})
+	var stdout bytes.Buffer
+	if err := runConfigWithBackendFactory(context.Background(), cfg, RunOptions{Stdout: &stdout}, provider); err != nil {
+		t.Fatalf("second Run() error = %v", err)
+	}
+	for _, want := range []string{
+		"destination=object-latest backend=s3 path_mapping=fixed target=. action=replace_takeover takeover_mode=same_pipeline",
+		"destination=ssh-latest backend=ssh path_mapping=fixed target=. action=replace_takeover takeover_mode=same_pipeline",
+		"replace_takeover=2",
+	} {
+		if !strings.Contains(stdout.String(), want) {
+			t.Fatalf("stdout = %q, want substring %q", stdout.String(), want)
+		}
+	}
 	testutil.AssertFakeFile(t, s3Destination, "report.md", "# Report\nNew.\n")
 	testutil.AssertFakeFile(t, s3Destination, "summary.txt", "New summary\n")
 	testutil.AssertFakeMissing(t, s3Destination, "new/report.md")
@@ -1221,7 +1358,7 @@ func TestRunContinuesAfterDestinationFailure(t *testing.T) {
 	for _, want := range []string{
 		"destination=archive-one backend=local action=error",
 		"destination=archive-two backend=local action=publish_new",
-		"Final status: failed planned=1 publish_new=1 replace_older=0 force_replace=0 skipped=0 failed=1 dry_run=false",
+		"Final status: failed planned=1 publish_new=1 replace_older=0 replace_takeover=0 force_replace=0 skipped=0 failed=1 dry_run=false",
 	} {
 		if !strings.Contains(output, want) {
 			t.Fatalf("stdout = %q, want substring %q", output, want)
@@ -1526,14 +1663,27 @@ func TestRunSkipsNewerDestination(t *testing.T) {
 	testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "newer\n")
 }
 
-func TestRunFailsOnConflict(t *testing.T) {
+func TestRunTakeoverNeverFailsOnConflict(t *testing.T) {
 	sourceRoot := t.TempDir()
 	destinationRoot := t.TempDir()
 	manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
 	manifest.ID = "other.source"
 	writeDestinationState(t, destinationRoot, "", manifest)
+	configPath := writeConfigFile(t, `
+pipelines:
+  - id: reports
+    source:
+      backend: local
+      path: `+sourceRoot+`
+    destinations:
+      - id: archive
+        backend: local
+        path: `+destinationRoot+`
+        takeover:
+          mode: never
+`)
 
-	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot)})
+	err := Run(context.Background(), RunOptions{ConfigPath: configPath})
 	if err == nil || !strings.Contains(err.Error(), "fail_conflict") {
 		t.Fatalf("Run() error = %v, want fail_conflict", err)
 	}
@@ -1649,7 +1799,7 @@ func TestRunExercisesRemoteBackendShapesThroughCommonPath(t *testing.T) {
 		"pipeline=local-to-ssh source=local",
 		"destination=ssh-archive backend=ssh action=publish_new",
 		"pipeline=ssh-to-local source=ssh",
-		"Final status: ok planned=4 publish_new=4 replace_older=0 force_replace=0 skipped=0 failed=0 dry_run=true",
+		"Final status: ok planned=4 publish_new=4 replace_older=0 replace_takeover=0 force_replace=0 skipped=0 failed=0 dry_run=true",
 	} {
 		if !strings.Contains(dryRunOutput.String(), want) {
 			t.Fatalf("dry-run output = %q, want substring %q", dryRunOutput.String(), want)
@@ -1823,6 +1973,25 @@ pipelines:
 `)
 }
 
+func writeSameSourcePreserveRelativeConfig(t *testing.T, sourceRoot, destinationRoot string) string {
+	t.Helper()
+	return writeConfigFile(t, `
+pipelines:
+  - id: reports
+    source:
+      backend: local
+      path: `+sourceRoot+`
+    destinations:
+      - id: archive
+        backend: local
+        path: `+destinationRoot+`
+        path_mapping:
+          mode: preserve_relative
+        takeover:
+          mode: same_source
+`)
+}
+
 func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestination string) string {
 	t.Helper()
 	return testutil.WriteFanoutLocalConfig(t, sourceRoot, firstDestination, secondDestination)

internal/cli/root_test.go

@@ -635,7 +635,7 @@ func TestExecuteRunDryRun(t *testing.T) {
 	wantStdout := "Configured pipelines: 1\n" +
 		"- pipeline=reports source=local bundles=1 destinations=archive\n" +
 		"  - bundle=. destination=archive backend=local action=publish_new outputs=report.md,summary.txt reason=\"destination state is absent\"\n" +
-		"Final status: ok planned=1 publish_new=1 replace_older=0 force_replace=0 skipped=0 failed=0 dry_run=true fixed_path=0\n"
+		"Final status: ok planned=1 publish_new=1 replace_older=0 replace_takeover=0 force_replace=0 skipped=0 failed=0 dry_run=true fixed_path=0\n"
 	if got := stdout.String(); got != wantStdout {
 		t.Fatalf("stdout = %q, want %q", got, wantStdout)
 	}

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

internal/publish/execute.go

@@ -16,7 +16,7 @@ func Execute(ctx context.Context, req Request, plan Plan) error {
 	switch plan.Action {
 	case ActionSkipSame, ActionSkipDestinationNewer:
 		return nil
-	case ActionPublishNew, ActionReplaceOlder, ActionForceReplace:
+	case ActionPublishNew, ActionReplaceOlder, ActionReplaceTakeover, ActionForceReplace:
 		if usesSharedRootState(req, plan) {
 			return executeSharedRoot(ctx, req, plan)
 		}
@@ -24,11 +24,11 @@ func Execute(ctx context.Context, req Request, plan Plan) error {
 		return fmt.Errorf("cannot execute action %s: %s", plan.Action, plan.Reason)
 	}
 
-	if plan.Action == ActionReplaceOlder {
+	if plan.Action == ActionReplaceOlder || plan.Action == ActionReplaceTakeover {
 		if plan.ExistingState == nil {
 			return fmt.Errorf("replace requires existing destination state")
 		}
-		if plan.Reconciliation.Mode == config.ReconciliationModeReplace {
+		if plan.Reconciliation.Mode == config.ReconciliationModeReplace || plan.Action == ActionReplaceTakeover {
 			if err := req.DestinationBackend.DeleteManagedBundle(ctx, req.DestinationBundlePath, state.ManagedOutputPaths(*plan.ExistingState), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil {
 				return err
 			}
@@ -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/execute_test.go

@@ -189,7 +189,6 @@ func TestExecuteFixedPathSupportsReconciliationModes(t *testing.T) {
 			})
 			destinationBackend := fake.New()
 			older := sourceBundle.Manifest
-			older.ID = "older.source"
 			older.Created = older.Created.Add(-time.Hour)
 			testutil.WriteFakeDestinationState(t, destinationBackend, "", older, testutil.DestinationStateOptions{})
 

internal/publish/force_test.go

@@ -166,6 +166,7 @@ func forceRequest(sourceBackend, destinationBackend *fake.Backend, sourceBundle
 		DestinationBackend:    destinationBackend,
 		DestinationBundlePath: sourceBundle.RootRelativePath,
 		Publish:               config.PublishPolicy{Source: true},
+		Takeover:              config.TakeoverPolicy{Mode: config.TakeoverModeNever},
 		Transfer:              transfer,
 		DistributorVersion:    "test",
 	}

internal/publish/plan.go

@@ -21,6 +21,7 @@ const (
 	ActionFailConflict         Action = "fail_conflict"
 	ActionFailUnmanaged        Action = "fail_unmanaged"
 	ActionForceReplace         Action = "force_replace"
+	ActionReplaceTakeover      Action = "replace_takeover"
 )
 
 type Request struct {
@@ -36,6 +37,7 @@ type Request struct {
 	Links                 *config.Links
 	State                 config.StatePolicy
 	Reconciliation        config.ReconciliationPolicy
+	Takeover              config.TakeoverPolicy
 	Transformers          TransformerResolver
 	Transfer              config.TransferPolicy
 	DistributorVersion    string
@@ -60,10 +62,12 @@ type Plan struct {
 	StateMode             string
 	OwnerScope            state.OwnerScope
 	Reconciliation        config.ReconciliationPolicy
+	TakeoverMode          string
 	Outputs               []Output
 	ExistingState         *state.DistributorState
 	ExistingSharedRoot    *state.SharedRootState
 	OtherOwnerOutputs     []state.SharedRootOutputFile
+	TakenOverOwnerOutputs []state.SharedRootOutputFile
 	RetainedOwnerOutputs  []state.SharedRootOutputFile
 	OwnerOutputsToDelete  []state.SharedRootOutputFile
 	OwnerOutputsToWrite   []Output
@@ -96,10 +100,14 @@ func Build(ctx context.Context, req Request) (Plan, error) {
 	if err != nil {
 		return Plan{}, err
 	}
-	comparison := compareDestination(req, status)
-	action, reason := actionForComparison(comparison, req.Transfer, req.Force)
 	reconciliation := normalizeReconciliation(req.Reconciliation)
 	stateMode := normalizeState(req.State).Mode
+	comparison := compareDestination(req, status)
+	action, reason := actionForComparison(comparison, req.Transfer, req.Force)
+	if takeoverActionAllowed(req, status, comparison, stateMode, action) {
+		action = ActionReplaceTakeover
+		reason = comparison.Reason
+	}
 	plan := Plan{
 		PipelineID:            req.PipelineID,
 		DestinationID:         req.DestinationID,
@@ -114,24 +122,28 @@ func Build(ctx context.Context, req Request) (Plan, error) {
 		StateMode:             stateMode,
 		OwnerScope:            state.CurrentOwnerScope(req.PipelineID, req.DestinationID),
 		Reconciliation:        reconciliation,
+		TakeoverMode:          normalizeTakeover(req.Takeover).Mode,
 		Outputs:               outputs,
 		ExistingState:         status.State,
 		ExistingSharedRoot:    status.SharedRoot,
 	}
 	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 {
-		return plan, fmt.Errorf("%s: %s", action, reason)
+	if plan.Action == ActionFailConflict || plan.Action == ActionFailUnmanaged {
+		return plan, fmt.Errorf("%s: %s", plan.Action, plan.Reason)
 	}
 	return plan, nil
 }
@@ -162,6 +174,11 @@ func validateRequest(req Request) error {
 	default:
 		return fmt.Errorf("state.mode must be %s or %s", config.StateModeSingleOwner, config.StateModeSharedRoot)
 	}
+	switch normalizeTakeover(req.Takeover).Mode {
+	case config.TakeoverModeSamePipeline, config.TakeoverModeSameSource, config.TakeoverModeAnyManaged, config.TakeoverModeNever:
+	default:
+		return fmt.Errorf("takeover.mode must be %s, %s, %s, or %s", config.TakeoverModeSamePipeline, config.TakeoverModeSameSource, config.TakeoverModeAnyManaged, config.TakeoverModeNever)
+	}
 	return nil
 }
 
@@ -179,6 +196,13 @@ func normalizeState(policy config.StatePolicy) config.StatePolicy {
 	return policy
 }
 
+func normalizeTakeover(policy config.TakeoverPolicy) config.TakeoverPolicy {
+	if policy.Mode == "" {
+		policy.Mode = config.TakeoverModeSamePipeline
+	}
+	return policy
+}
+
 func compareDestination(req Request, status state.DestinationStatus) state.Comparison {
 	if normalizeState(req.State).Mode == config.StateModeSharedRoot {
 		scope := state.CurrentOwnerScope(req.PipelineID, req.DestinationID)
@@ -194,21 +218,19 @@ func compareDestination(req Request, status state.DestinationStatus) state.Compa
 			return state.Comparison{Outcome: state.OutcomeDestinationOlder, Reason: "fixed destination source is older than selected source"}
 		}
 		if destinationManifest.Created.After(req.SourceBundle.Manifest.Created) {
-			return state.Comparison{Outcome: state.OutcomeDestinationNewer, Reason: "fixed destination source is newer than selected source"}
+			return state.Comparison{
+				Outcome: state.OutcomeDestinationNewer,
+				Reason:  "fixed destination source is newer than selected source",
+				Detail: state.ComparisonDetail{
+					Kind:                state.ComparisonDetailDestinationNewer,
+					CurrentSourceID:     req.SourceBundle.Manifest.ID,
+					DestinationSourceID: destinationManifest.ID,
+				},
+			}
 		}
 		return comparison
 	}
 	comparison := state.Compare(req.SourceBundle.Manifest, req.PipelineID, req.DestinationID, status)
-	if req.PathMapping != config.PathMappingFixed || comparison.Outcome != state.OutcomeDifferentSourceConflict || status.State == nil {
-		return comparison
-	}
-	destinationManifest := status.State.Source.Manifest
-	if destinationManifest.Created.Before(req.SourceBundle.Manifest.Created) {
-		return state.Comparison{Outcome: state.OutcomeDestinationOlder, Reason: "fixed destination source is older than selected source"}
-	}
-	if destinationManifest.Created.After(req.SourceBundle.Manifest.Created) {
-		return state.Comparison{Outcome: state.OutcomeDestinationNewer, Reason: "fixed destination source is newer than selected source"}
-	}
 	return comparison
 }
 
@@ -226,6 +248,7 @@ type sharedRootPlanDetails struct {
 	Action                Action
 	Reason                string
 	OtherOwnerOutputs     []state.SharedRootOutputFile
+	TakenOverOwnerOutputs []state.SharedRootOutputFile
 	RetainedOwnerOutputs  []state.SharedRootOutputFile
 	OwnerOutputsToDelete  []state.SharedRootOutputFile
 	OwnerOutputsToWrite   []Output
@@ -244,19 +267,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 {
-		reason := fmt.Sprintf("destination output path %s is owned by %s/%s", conflict.Path, conflict.Owner.PipelineID, conflict.Owner.DestinationID)
+	conflicts := sharedRootPathOwnershipConflicts(status, scope, plannedPaths)
+	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 {
@@ -266,11 +299,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)
 		}
 	}
@@ -280,7 +313,7 @@ func planSharedRootOwner(ctx context.Context, req Request, status state.Destinat
 
 func isWriteAction(action Action) bool {
 	switch action {
-	case ActionPublishNew, ActionReplaceOlder, ActionForceReplace:
+	case ActionPublishNew, ActionReplaceOlder, ActionReplaceTakeover, ActionForceReplace:
 		return true
 	default:
 		return false
@@ -295,11 +328,76 @@ func outputPaths(outputs []Output) []string {
 	return paths
 }
 
-func sharedRootPathOwnershipConflict(status state.DestinationStatus, scope state.OwnerScope, paths []string) (state.PathOwnershipConflict, bool) {
-	if status.SharedRoot != nil {
-		return status.SharedRoot.PathOwnershipConflict(scope, paths)
+func sharedRootOwnershipConflictReason(conflict state.PathOwnershipConflict) string {
+	return fmt.Sprintf("destination output path %s is owned by %s/%s", conflict.Path, conflict.Owner.PipelineID, conflict.Owner.DestinationID)
+}
+
+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 {
@@ -333,18 +431,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))
@@ -423,3 +537,32 @@ func actionForComparison(comparison state.Comparison, transfer config.TransferPo
 		return ActionFailConflict, "unsupported comparison outcome"
 	}
 }
+
+func takeoverActionAllowed(req Request, status state.DestinationStatus, comparison state.Comparison, stateMode string, action Action) bool {
+	if stateMode != config.StateModeSingleOwner || status.State == nil {
+		return false
+	}
+	if action == ActionForceReplace {
+		return false
+	}
+	switch comparison.Detail.Kind {
+	case state.ComparisonDetailPipelineIDMismatch,
+		state.ComparisonDetailDestinationIDMismatch,
+		state.ComparisonDetailDifferentSourceID:
+	default:
+		return false
+	}
+	takeover := normalizeTakeover(req.Takeover)
+	switch takeover.Mode {
+	case config.TakeoverModeSamePipeline:
+		return status.State.PipelineID == req.PipelineID
+	case config.TakeoverModeSameSource:
+		return status.State.Source.Manifest.ID == req.SourceBundle.Manifest.ID
+	case config.TakeoverModeAnyManaged:
+		return true
+	case config.TakeoverModeNever:
+		return false
+	default:
+		return false
+	}
+}

internal/publish/plan_test.go

@@ -0,0 +1,36 @@
+package publish
+
+import (
+	"testing"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+	"gitea.maximumdirect.net/eric/distributor/internal/state"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
+	"gitea.maximumdirect.net/eric/distributor/internal/testutil"
+)
+
+func TestCompareDestinationFixedPathPreservesDifferentSourceConflict(t *testing.T) {
+	sourceBackend := fake.New()
+	sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{})
+	destinationState := testutil.DestinationState(sourceBundle.Manifest, testutil.DestinationStateOptions{})
+	destinationState.Source.Manifest.ID = "latest.previous"
+
+	comparison := compareDestination(Request{
+		PipelineID:            "reports",
+		DestinationID:         "archive",
+		SourceBundle:          sourceBundle,
+		DestinationBundlePath: "",
+		PathMapping:           config.PathMappingFixed,
+		State:                 config.StatePolicy{Mode: config.StateModeSingleOwner},
+	}, state.DestinationStatus{State: &destinationState, HasContents: true})
+
+	if comparison.Outcome != state.OutcomeDifferentSourceConflict {
+		t.Fatalf("comparison outcome = %s, want %s", comparison.Outcome, state.OutcomeDifferentSourceConflict)
+	}
+	if comparison.Detail.Kind != state.ComparisonDetailDifferentSourceID {
+		t.Fatalf("detail kind = %q, want %q", comparison.Detail.Kind, state.ComparisonDetailDifferentSourceID)
+	}
+	if comparison.Detail.CurrentSourceID != sourceBundle.Manifest.ID || comparison.Detail.DestinationSourceID != "latest.previous" {
+		t.Fatalf("detail = %#v, want source ids", comparison.Detail)
+	}
+}

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)

internal/publish/takeover_test.go

@@ -0,0 +1,215 @@
+package publish
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
+	"gitea.maximumdirect.net/eric/distributor/internal/testutil"
+)
+
+func TestBuildPlansSingleOwnerTakeoverByPolicy(t *testing.T) {
+	tests := []struct {
+		name        string
+		takeover    config.TakeoverPolicy
+		mutateState func(*bundle.Manifest, *testutil.DestinationStateOptions)
+		wantAction  Action
+		wantErr     string
+	}{
+		{
+			name:     "default same pipeline different source",
+			takeover: config.TakeoverPolicy{},
+			mutateState: func(manifest *bundle.Manifest, opts *testutil.DestinationStateOptions) {
+				manifest.ID = "other.source"
+			},
+			wantAction: ActionReplaceTakeover,
+		},
+		{
+			name:     "same pipeline different newer source",
+			takeover: config.TakeoverPolicy{Mode: config.TakeoverModeSamePipeline},
+			mutateState: func(manifest *bundle.Manifest, opts *testutil.DestinationStateOptions) {
+				manifest.ID = "other.source"
+				manifest.Created = manifest.Created.AddDate(0, 0, 1)
+			},
+			wantAction: ActionReplaceTakeover,
+		},
+		{
+			name:     "default same pipeline different destination",
+			takeover: config.TakeoverPolicy{},
+			mutateState: func(manifest *bundle.Manifest, opts *testutil.DestinationStateOptions) {
+				opts.DestinationID = "web"
+			},
+			wantAction: ActionReplaceTakeover,
+		},
+		{
+			name:     "same pipeline refuses different pipeline",
+			takeover: config.TakeoverPolicy{Mode: config.TakeoverModeSamePipeline},
+			mutateState: func(manifest *bundle.Manifest, opts *testutil.DestinationStateOptions) {
+				opts.PipelineID = "other"
+			},
+			wantErr: "fail_conflict",
+		},
+		{
+			name:     "same source allows different pipeline",
+			takeover: config.TakeoverPolicy{Mode: config.TakeoverModeSameSource},
+			mutateState: func(manifest *bundle.Manifest, opts *testutil.DestinationStateOptions) {
+				opts.PipelineID = "other"
+			},
+			wantAction: ActionReplaceTakeover,
+		},
+		{
+			name:     "same source refuses different source",
+			takeover: config.TakeoverPolicy{Mode: config.TakeoverModeSameSource},
+			mutateState: func(manifest *bundle.Manifest, opts *testutil.DestinationStateOptions) {
+				manifest.ID = "other.source"
+			},
+			wantErr: "fail_conflict",
+		},
+		{
+			name:     "any managed allows different pipeline",
+			takeover: config.TakeoverPolicy{Mode: config.TakeoverModeAnyManaged},
+			mutateState: func(manifest *bundle.Manifest, opts *testutil.DestinationStateOptions) {
+				manifest.ID = "other.source"
+				opts.PipelineID = "other"
+			},
+			wantAction: ActionReplaceTakeover,
+		},
+		{
+			name:     "never refuses different source",
+			takeover: config.TakeoverPolicy{Mode: config.TakeoverModeNever},
+			mutateState: func(manifest *bundle.Manifest, opts *testutil.DestinationStateOptions) {
+				manifest.ID = "other.source"
+			},
+			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{})
+			destinationBackend := fake.New()
+			destinationManifest := sourceBundle.Manifest
+			destinationManifest.Files = append([]bundle.ManifestFile(nil), sourceBundle.Manifest.Files...)
+			opts := testutil.DestinationStateOptions{}
+			tt.mutateState(&destinationManifest, &opts)
+			testutil.WriteFakeDestinationState(t, destinationBackend, "bundle", destinationManifest, opts)
+
+			req := takeoverRequest(sourceBackend, destinationBackend, sourceBundle, tt.takeover, config.ReconciliationModeReplace)
+			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)
+				}
+				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)
+			}
+		})
+	}
+}
+
+func TestBuildDoesNotTakeOverInvalidOrUnmanagedDestination(t *testing.T) {
+	tests := []struct {
+		name    string
+		prepare func(t *testing.T, backend *fake.Backend)
+		wantErr string
+	}{
+		{
+			name: "invalid state",
+			prepare: func(t *testing.T, backend *fake.Backend) {
+				t.Helper()
+				statePath, err := storage.StatePath("bundle")
+				if err != nil {
+					t.Fatalf("state path: %v", err)
+				}
+				testutil.WriteFakeFile(t, backend, statePath, "{invalid")
+			},
+			wantErr: "fail_conflict",
+		},
+		{
+			name: "unmanaged content",
+			prepare: func(t *testing.T, backend *fake.Backend) {
+				t.Helper()
+				testutil.WriteFakeFile(t, backend, "bundle/old.txt", "old")
+			},
+			wantErr: "fail_unmanaged",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			sourceBackend := fake.New()
+			sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{})
+			destinationBackend := fake.New()
+			tt.prepare(t, destinationBackend)
+
+			req := takeoverRequest(sourceBackend, destinationBackend, sourceBundle, config.TakeoverPolicy{Mode: config.TakeoverModeAnyManaged}, config.ReconciliationModeReplace)
+			_, err := Build(context.Background(), req)
+			if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
+				t.Fatalf("Build() error = %v, want %q", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestExecuteTakeoverMergeDoesNotRetainOldSourceOutputs(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()
+	oldManifest := sourceBundle.Manifest
+	oldManifest.ID = "old.source"
+	oldManifest.Files = []bundle.ManifestFile{
+		{Path: "report.md", SHA256: bundle.FileDigest([]byte("old\n")), Size: int64(len("old\n"))},
+		{Path: "summary.txt", SHA256: bundle.FileDigest([]byte("old summary\n")), Size: int64(len("old summary\n"))},
+	}
+	oldManifest.Digest = bundle.BundleDigest(oldManifest.Files)
+	testutil.WriteFakeDestinationState(t, destinationBackend, "bundle", oldManifest, testutil.DestinationStateOptions{})
+
+	req := takeoverRequest(sourceBackend, destinationBackend, sourceBundle, config.TakeoverPolicy{Mode: config.TakeoverModeSamePipeline}, config.ReconciliationModeMerge)
+	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.AssertFakeMissing(t, destinationBackend, "bundle/summary.txt")
+	destinationState := readFakeState(t, destinationBackend, "bundle")
+	if got, want := len(destinationState.Outputs), 1; got != want {
+		t.Fatalf("state output count = %d, want %d", got, want)
+	}
+	if got, want := destinationState.Source.Manifest.ID, sourceBundle.Manifest.ID; got != want {
+		t.Fatalf("state source id = %q, want %q", got, want)
+	}
+}
+
+func takeoverRequest(sourceBackend, destinationBackend *fake.Backend, sourceBundle bundle.Bundle, takeover config.TakeoverPolicy, reconciliationMode string) Request {
+	return Request{
+		PipelineID:            "reports",
+		DestinationID:         "archive",
+		SourceBundle:          sourceBundle,
+		SourceBackend:         sourceBackend,
+		DestinationBackend:    destinationBackend,
+		DestinationBundlePath: sourceBundle.RootRelativePath,
+		Publish:               config.PublishPolicy{Source: true},
+		Reconciliation:        config.ReconciliationPolicy{Mode: reconciliationMode},
+		Takeover:              takeover,
+		Transfer:              defaultTransfer(),
+		DistributorVersion:    "test",
+	}
+}

internal/state/compare.go

@@ -30,37 +30,75 @@ type DestinationStatus struct {
 type Comparison struct {
 	Outcome Outcome
 	Reason  string
+	Detail  ComparisonDetail
+}
+
+type ComparisonDetailKind string
+
+const (
+	ComparisonDetailNone                      ComparisonDetailKind = ""
+	ComparisonDetailInvalidState              ComparisonDetailKind = "invalid_state"
+	ComparisonDetailUnmanagedContent          ComparisonDetailKind = "unmanaged_content"
+	ComparisonDetailPipelineIDMismatch        ComparisonDetailKind = "pipeline_id_mismatch"
+	ComparisonDetailDestinationIDMismatch     ComparisonDetailKind = "destination_id_mismatch"
+	ComparisonDetailDifferentSourceID         ComparisonDetailKind = "different_source_id"
+	ComparisonDetailSameCreatedDigestConflict ComparisonDetailKind = "same_created_digest_conflict"
+	ComparisonDetailDestinationNewer          ComparisonDetailKind = "destination_newer"
+	ComparisonDetailSharedRootOwnerAbsent     ComparisonDetailKind = "shared_root_owner_absent"
+	ComparisonDetailSharedRootOutputOwner     ComparisonDetailKind = "shared_root_output_owner_conflict"
+)
+
+type ComparisonDetail struct {
+	Kind                     ComparisonDetailKind
+	CurrentPipelineID        string
+	CurrentDestinationID     string
+	DestinationPipelineID    string
+	DestinationDestinationID string
+	CurrentSourceID          string
+	DestinationSourceID      string
+	CurrentSourceDigest      string
+	DestinationSourceDigest  string
+	Path                     string
+	CurrentOwner             OwnerScope
+	ConflictingOwner         OwnerScope
 }
 
 func CompareSharedRootOwner(source bundle.Manifest, scope OwnerScope, status DestinationStatus) Comparison {
 	if status.StateErr != nil {
-		return Comparison{Outcome: OutcomeInvalidState, Reason: status.StateErr.Error()}
+		return Comparison{Outcome: OutcomeInvalidState, Reason: status.StateErr.Error(), Detail: ComparisonDetail{Kind: ComparisonDetailInvalidState}}
 	}
 	if status.SharedRoot != nil {
 		if err := ValidateSharedRoot(*status.SharedRoot); err != nil {
-			return Comparison{Outcome: OutcomeInvalidState, Reason: err.Error()}
+			return Comparison{Outcome: OutcomeInvalidState, Reason: err.Error(), Detail: ComparisonDetail{Kind: ComparisonDetailInvalidState}}
 		}
 		owner, ok := status.SharedRoot.Owner(scope)
 		if !ok {
-			return Comparison{Outcome: OutcomeDestinationAbsent, Reason: fmt.Sprintf("destination owner %s/%s is absent", scope.PipelineID, scope.DestinationID)}
+			return Comparison{
+				Outcome: OutcomeDestinationAbsent,
+				Reason:  fmt.Sprintf("destination owner %s/%s is absent", scope.PipelineID, scope.DestinationID),
+				Detail: ComparisonDetail{
+					Kind:         ComparisonDetailSharedRootOwnerAbsent,
+					CurrentOwner: scope,
+				},
+			}
 		}
 		return compareManifests(source, owner.Source.Manifest)
 	}
 	if status.State != nil {
 		destinationState := *status.State
 		if err := Validate(destinationState); err != nil {
-			return Comparison{Outcome: OutcomeInvalidState, Reason: err.Error()}
+			return Comparison{Outcome: OutcomeInvalidState, Reason: err.Error(), Detail: ComparisonDetail{Kind: ComparisonDetailInvalidState}}
 		}
 		if destinationState.PipelineID != scope.PipelineID {
-			return Comparison{Outcome: OutcomeIdentityMismatch, Reason: fmt.Sprintf("pipeline id %q does not match %q", destinationState.PipelineID, scope.PipelineID)}
+			return pipelineIDMismatchComparison(destinationState.PipelineID, scope.PipelineID)
 		}
 		if destinationState.DestinationID != scope.DestinationID {
-			return Comparison{Outcome: OutcomeIdentityMismatch, Reason: fmt.Sprintf("destination id %q does not match %q", destinationState.DestinationID, scope.DestinationID)}
+			return destinationIDMismatchComparison(destinationState.DestinationID, scope.DestinationID)
 		}
 		return compareManifests(source, destinationState.Source.Manifest)
 	}
 	if status.HasContents {
-		return Comparison{Outcome: OutcomeDestinationUnmanaged, Reason: "destination has content but no distributor state"}
+		return Comparison{Outcome: OutcomeDestinationUnmanaged, Reason: "destination has content but no distributor state", Detail: ComparisonDetail{Kind: ComparisonDetailUnmanagedContent}}
 	}
 	return Comparison{Outcome: OutcomeDestinationAbsent, Reason: "destination state is absent"}
 }
@@ -70,45 +108,95 @@ func compareManifests(source, destination bundle.Manifest) Comparison {
 		return Comparison{Outcome: OutcomeSameSource, Reason: "destination source manifest matches source"}
 	}
 	if destination.ID != source.ID {
-		return Comparison{Outcome: OutcomeDifferentSourceConflict, Reason: "destination source id differs from source"}
+		return Comparison{
+			Outcome: OutcomeDifferentSourceConflict,
+			Reason:  "destination source id differs from source",
+			Detail: ComparisonDetail{
+				Kind:                ComparisonDetailDifferentSourceID,
+				CurrentSourceID:     source.ID,
+				DestinationSourceID: destination.ID,
+			},
+		}
 	}
 	if destination.Created.Before(source.Created) {
 		return Comparison{Outcome: OutcomeDestinationOlder, Reason: "destination source is older than source"}
 	}
 	if destination.Created.After(source.Created) {
-		return Comparison{Outcome: OutcomeDestinationNewer, Reason: "destination source is newer than source"}
+		return Comparison{
+			Outcome: OutcomeDestinationNewer,
+			Reason:  "destination source is newer than source",
+			Detail: ComparisonDetail{
+				Kind:                ComparisonDetailDestinationNewer,
+				CurrentSourceID:     source.ID,
+				DestinationSourceID: destination.ID,
+			},
+		}
 	}
 	if destination.Digest != source.Digest {
-		return Comparison{Outcome: OutcomeSameCreatedConflict, Reason: "destination source has same id and created time but different digest"}
+		return Comparison{
+			Outcome: OutcomeSameCreatedConflict,
+			Reason:  "destination source has same id and created time but different digest",
+			Detail: ComparisonDetail{
+				Kind:                    ComparisonDetailSameCreatedDigestConflict,
+				CurrentSourceID:         source.ID,
+				DestinationSourceID:     destination.ID,
+				CurrentSourceDigest:     source.Digest,
+				DestinationSourceDigest: destination.Digest,
+			},
 		}
-	return Comparison{Outcome: OutcomeInvalidState, Reason: "destination source differs from source without a supported comparison outcome"}
+	}
+	return Comparison{Outcome: OutcomeInvalidState, Reason: "destination source differs from source without a supported comparison outcome", Detail: ComparisonDetail{Kind: ComparisonDetailInvalidState}}
 }
 
 func Compare(source bundle.Manifest, pipelineID, destinationID string, status DestinationStatus) Comparison {
 	if status.StateErr != nil {
-		return Comparison{Outcome: OutcomeInvalidState, Reason: status.StateErr.Error()}
+		return Comparison{Outcome: OutcomeInvalidState, Reason: status.StateErr.Error(), Detail: ComparisonDetail{Kind: ComparisonDetailInvalidState}}
 	}
 	if status.State == nil {
 		if status.HasContents {
-			return Comparison{Outcome: OutcomeDestinationUnmanaged, Reason: "destination has content but no distributor state"}
+			return Comparison{Outcome: OutcomeDestinationUnmanaged, Reason: "destination has content but no distributor state", Detail: ComparisonDetail{Kind: ComparisonDetailUnmanagedContent}}
 		}
 		return Comparison{Outcome: OutcomeDestinationAbsent, Reason: "destination state is absent"}
 	}
 
 	destinationState := *status.State
 	if err := Validate(destinationState); err != nil {
-		return Comparison{Outcome: OutcomeInvalidState, Reason: err.Error()}
+		return Comparison{Outcome: OutcomeInvalidState, Reason: err.Error(), Detail: ComparisonDetail{Kind: ComparisonDetailInvalidState}}
 	}
 	if destinationState.PipelineID != pipelineID {
-		return Comparison{Outcome: OutcomeIdentityMismatch, Reason: fmt.Sprintf("pipeline id %q does not match %q", destinationState.PipelineID, pipelineID)}
+		return pipelineIDMismatchComparison(destinationState.PipelineID, pipelineID)
 	}
 	if destinationState.DestinationID != destinationID {
-		return Comparison{Outcome: OutcomeIdentityMismatch, Reason: fmt.Sprintf("destination id %q does not match %q", destinationState.DestinationID, destinationID)}
+		return destinationIDMismatchComparison(destinationState.DestinationID, destinationID)
 	}
 
 	return compareManifests(source, destinationState.Source.Manifest)
 }
 
+func pipelineIDMismatchComparison(destinationPipelineID, currentPipelineID string) Comparison {
+	return Comparison{
+		Outcome: OutcomeIdentityMismatch,
+		Reason:  fmt.Sprintf("pipeline id %q does not match %q", destinationPipelineID, currentPipelineID),
+		Detail: ComparisonDetail{
+			Kind:                  ComparisonDetailPipelineIDMismatch,
+			CurrentPipelineID:     currentPipelineID,
+			DestinationPipelineID: destinationPipelineID,
+		},
+	}
+}
+
+func destinationIDMismatchComparison(destinationDestinationID, currentDestinationID string) Comparison {
+	return Comparison{
+		Outcome: OutcomeIdentityMismatch,
+		Reason:  fmt.Sprintf("destination id %q does not match %q", destinationDestinationID, currentDestinationID),
+		Detail: ComparisonDetail{
+			Kind:                     ComparisonDetailDestinationIDMismatch,
+			CurrentDestinationID:     currentDestinationID,
+			DestinationDestinationID: destinationDestinationID,
+		},
+	}
+}
+
 func manifestsEqual(a, b bundle.Manifest) bool {
 	if a.SchemaVersion != b.SchemaVersion ||
 		a.ID != b.ID ||

internal/state/compare_test.go

@@ -97,6 +97,93 @@ func TestCompareOutcomes(t *testing.T) {
 	}
 }
 
+func TestCompareReportsStructuredDetails(t *testing.T) {
+	source := validManifest(t)
+	tests := []struct {
+		name       string
+		status     DestinationStatus
+		wantKind   ComparisonDetailKind
+		assertions func(t *testing.T, detail ComparisonDetail)
+	}{
+		{
+			name:     "pipeline mismatch",
+			status:   DestinationStatus{State: withState(t, source, func(s *DistributorState) { s.PipelineID = "other" })},
+			wantKind: ComparisonDetailPipelineIDMismatch,
+			assertions: func(t *testing.T, detail ComparisonDetail) {
+				t.Helper()
+				if detail.DestinationPipelineID != "other" || detail.CurrentPipelineID != "reports" {
+					t.Fatalf("detail = %#v, want pipeline ids", detail)
+				}
+			},
+		},
+		{
+			name:     "destination mismatch",
+			status:   DestinationStatus{State: withState(t, source, func(s *DistributorState) { s.DestinationID = "other" })},
+			wantKind: ComparisonDetailDestinationIDMismatch,
+			assertions: func(t *testing.T, detail ComparisonDetail) {
+				t.Helper()
+				if detail.DestinationDestinationID != "other" || detail.CurrentDestinationID != "archive" {
+					t.Fatalf("detail = %#v, want destination ids", detail)
+				}
+			},
+		},
+		{
+			name:     "different source id",
+			status:   DestinationStatus{State: withState(t, source, func(s *DistributorState) { s.Source.Manifest.ID = "other.source" })},
+			wantKind: ComparisonDetailDifferentSourceID,
+			assertions: func(t *testing.T, detail ComparisonDetail) {
+				t.Helper()
+				if detail.DestinationSourceID != "other.source" || detail.CurrentSourceID != source.ID {
+					t.Fatalf("detail = %#v, want source ids", detail)
+				}
+			},
+		},
+		{
+			name: "same created digest conflict",
+			status: DestinationStatus{State: withState(t, source, func(s *DistributorState) {
+				s.Source.Manifest.Files[0].SHA256 = "sha256:3333333333333333333333333333333333333333333333333333333333333333"
+				s.Source.Manifest.Digest = bundle.BundleDigest(s.Source.Manifest.Files)
+			})},
+			wantKind: ComparisonDetailSameCreatedDigestConflict,
+			assertions: func(t *testing.T, detail ComparisonDetail) {
+				t.Helper()
+				if detail.CurrentSourceDigest == "" || detail.DestinationSourceDigest == "" || detail.CurrentSourceDigest == detail.DestinationSourceDigest {
+					t.Fatalf("detail = %#v, want different source digests", detail)
+				}
+			},
+		},
+		{
+			name: "destination newer",
+			status: DestinationStatus{State: withState(t, source, func(s *DistributorState) {
+				s.Source.Manifest.Created = source.Created.Add(time.Hour)
+			})},
+			wantKind: ComparisonDetailDestinationNewer,
+		},
+		{
+			name:     "invalid state",
+			status:   DestinationStatus{StateErr: errors.New("invalid json")},
+			wantKind: ComparisonDetailInvalidState,
+		},
+		{
+			name:     "unmanaged content",
+			status:   DestinationStatus{HasContents: true},
+			wantKind: ComparisonDetailUnmanagedContent,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := Compare(source, "reports", "archive", tt.status)
+			if got.Detail.Kind != tt.wantKind {
+				t.Fatalf("Compare() detail kind = %q, want %q; comparison=%#v", got.Detail.Kind, tt.wantKind, got)
+			}
+			if tt.assertions != nil {
+				tt.assertions(t, got.Detail)
+			}
+		})
+	}
+}
+
 func withState(t *testing.T, source bundle.Manifest, mutate func(*DistributorState)) *DistributorState {
 	t.Helper()
 	stateManifest := source

internal/state/outputs.go

@@ -191,7 +191,17 @@ func (s SharedRootState) PathOwnershipConflict(scope OwnerScope, paths []string)
 	for _, path := range paths {
 		owner, exists := s.OutputOwner(path)
 		if exists && owner != scope {
-			return PathOwnershipConflict{Path: path, Owner: owner}, true
+			return PathOwnershipConflict{
+				Path:         path,
+				Owner:        owner,
+				CurrentOwner: scope,
+				Detail: ComparisonDetail{
+					Kind:             ComparisonDetailSharedRootOutputOwner,
+					Path:             path,
+					CurrentOwner:     scope,
+					ConflictingOwner: owner,
+				},
+			}, true
 		}
 	}
 	return PathOwnershipConflict{}, false

internal/state/shared_root.go

@@ -59,6 +59,8 @@ type SharedRootOutputFile struct {
 type PathOwnershipConflict struct {
 	Path         string
 	Owner        OwnerScope
+	CurrentOwner OwnerScope
+	Detail       ComparisonDetail
 }
 
 type rawSharedRootState struct {

internal/state/shared_root_test.go

@@ -130,6 +130,12 @@ func TestSharedRootOutputHelpers(t *testing.T) {
 	if !ok || conflict.Owner != html {
 		t.Fatalf("conflict = %#v ok=%t, want html owner conflict", conflict, ok)
 	}
+	if conflict.Detail.Kind != ComparisonDetailSharedRootOutputOwner {
+		t.Fatalf("conflict detail kind = %q, want %q", conflict.Detail.Kind, ComparisonDetailSharedRootOutputOwner)
+	}
+	if conflict.Detail.Path != "report.html" || conflict.Detail.CurrentOwner != archive || conflict.Detail.ConflictingOwner != html {
+		t.Fatalf("conflict detail = %#v, want path and owners", conflict.Detail)
+	}
 }
 
 func TestRemoveMissingSharedRootOwnerOutputs(t *testing.T) {
@@ -241,6 +247,12 @@ func TestCompareSharedRootOwnerScopesCurrentOwner(t *testing.T) {
 	if missing.Outcome != OutcomeDestinationAbsent {
 		t.Fatalf("missing owner comparison = %#v, want destination absent", missing)
 	}
+	if missing.Detail.Kind != ComparisonDetailSharedRootOwnerAbsent {
+		t.Fatalf("missing owner detail kind = %q, want %q", missing.Detail.Kind, ComparisonDetailSharedRootOwnerAbsent)
+	}
+	if missing.Detail.CurrentOwner != CurrentOwnerScope("missing", "archive") {
+		t.Fatalf("missing owner detail = %#v, want missing/archive", missing.Detail)
+	}
 }
 
 func TestCompareSharedRootOwnerAcceptsMatchingSingleOwnerState(t *testing.T) {