diff --git a/cmd/bucket-lifecycle.go b/cmd/bucket-lifecycle.go
index b1732184b..ef0063ee2 100644
--- a/cmd/bucket-lifecycle.go
+++ b/cmd/bucket-lifecycle.go
@@ -162,7 +162,7 @@ func (n noncurrentVersionsTask) OpHash() uint64 {
 }
 
 func (j jentry) OpHash() uint64 {
-	return xxh3.HashString(j.TierName + j.ObjName)
+	return xxh3.HashString(j.Bucket + j.Object + j.TierName + j.RemoteObject + j.RemoteVersionID)
 }
 
 func (e expiryTask) OpHash() uint64 {
@@ -354,7 +354,16 @@ func (es *expiryState) Worker(input <-chan expiryOp) {
 			case noncurrentVersionsTask:
 				deleteObjectVersions(es.ctx, es.objAPI, v.bucket, v.versions, v.events)
 			case jentry:
-				transitionLogIf(es.ctx, deleteObjectFromRemoteTier(es.ctx, v.ObjName, v.VersionID, v.TierName))
+				z, ok := es.objAPI.(*erasureServerPools)
+				if !ok {
+					transitionLogIf(es.ctx, errTierOwnershipUnavailable)
+					continue
+				}
+				transitionLogIf(es.ctx, z.deleteTierTuple(es.ctx, v.Bucket, v.Object, tierRemoteTuple{
+					Tier:      v.TierName,
+					Object:    v.RemoteObject,
+					VersionID: v.RemoteVersionID,
+				}))
 			case freeVersionTask:
 				oi := v.ObjectInfo
 				traceFn := globalLifecycleSys.trace(oi)
@@ -370,19 +379,26 @@ func (es *expiryState) Worker(input <-chan expiryOp) {
 					}
 					return err
 				}
-				// Remove the remote object
-				err := deleteObjectFromRemoteTier(es.ctx, oi.TransitionedObject.Name, oi.TransitionedObject.VersionID, oi.TransitionedObject.Tier)
+				z, ok := es.objAPI.(*erasureServerPools)
+				if !ok {
+					transitionLogIf(es.ctx, errTierOwnershipUnavailable)
+					traceFn(ILMFreeVersionDelete, nil, errTierOwnershipUnavailable)
+					continue
+				}
+
+				// Check ownership, remove the remote object, and remove only the
+				// matching free record while the object lock is held.
+				err := z.deleteTierTuple(es.ctx, oi.Bucket, oi.Name, tierRemoteTuple{
+					Tier:      oi.TransitionedObject.Tier,
+					Object:    oi.TransitionedObject.Name,
+					VersionID: oi.TransitionedObject.VersionID,
+				})
 				if ignoreNotFoundErr(err) != nil {
 					transitionLogIf(es.ctx, err)
 					traceFn(ILMFreeVersionDelete, nil, err)
 					continue
 				}
 
-				// Remove this free version
-				_, err = es.objAPI.DeleteObject(es.ctx, oi.Bucket, oi.Name, ObjectOptions{
-					VersionID:        oi.VersionID,
-					InclFreeVersions: true,
-				})
 				if err == nil {
 					auditLogLifecycle(es.ctx, oi, ILMFreeVersionDelete, nil, traceFn)
 				}
@@ -661,7 +677,20 @@ func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *Ob
 	}
 
 	if !keepRemote {
-		if err := deleteObjectFromRemoteTier(ctx, oi.TransitionedObject.Name, oi.TransitionedObject.VersionID, oi.TransitionedObject.Tier); err != nil {
+		z, ok := objectAPI.(*erasureServerPools)
+		if !ok {
+			transitionLogIf(ctx, errTierOwnershipUnavailable)
+			return nil // The completed local expiry owns a durable free record.
+		}
+		if err := z.deleteTierTuple(ctx, oi.Bucket, oi.Name, tierRemoteTuple{
+			Tier:      oi.TransitionedObject.Tier,
+			Object:    oi.TransitionedObject.Name,
+			VersionID: oi.TransitionedObject.VersionID,
+		}); err != nil {
+			// The hot delete has already created a durable free record. Keep
+			// the expiry successful while the scanner retries remote cleanup;
+			// ownership or tier errors must never cause a second remote delete
+			// without that record.
 			transitionLogIf(ctx, err)
 		}
 	}
diff --git a/cmd/erasure-server-pool-decom.go b/cmd/erasure-server-pool-decom.go
index 30f1fe132..69bab607b 100644
--- a/cmd/erasure-server-pool-decom.go
+++ b/cmd/erasure-server-pool-decom.go
@@ -807,6 +807,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 			if lc == nil {
 				return lifecycle.NoneAction
 			}
+			history.loadTier(ctx, z, bucket, object)
 			versioned := vc != nil && vc.Versioned(object)
 			objInfo := fi.ToObjectInfo(bucket, object, versioned)
 
@@ -1186,7 +1187,11 @@ func (z *erasureServerPools) removeFreeVersions(ctx context.Context, bucket, obj
 		return nil
 	}
 	for _, fv := range fvs {
-		err := deleteObjectFromRemoteTier(ctx, fv.TransitionedObjName, fv.TransitionVersionID, fv.TransitionTier)
+		err := z.deleteTierTuple(ctx, bucket, object, tierRemoteTuple{
+			Tier:      fv.TransitionTier,
+			Object:    fv.TransitionedObjName,
+			VersionID: fv.TransitionVersionID,
+		})
 		if errors.Is(err, errTierNotFound) {
 			// The free version is the only record of its remote object. It
 			// stays until the tier can be reached again.
@@ -1199,25 +1204,6 @@ func (z *erasureServerPools) removeFreeVersions(ctx context.Context, bucket, obj
 				fv.TransitionedObjName, bucket, object, fv.TransitionTier, err)
 		}
 	}
-	object = encodeDirObject(object)
-	lk := z.NewNSLock(bucket, object)
-	lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
-	if err != nil {
-		return err
-	}
-	ctx = lkctx.Context()
-	defer lk.Unlock(lkctx)
-	for _, fv := range fvs {
-		_, err := z.serverPools[idx].DeleteObject(ctx, bucket, object, ObjectOptions{
-			VersionID:        fv.VersionID,
-			InclFreeVersions: true,
-			NoLock:           true,
-			NoAuditLog:       true,
-		})
-		if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
-			return err
-		}
-	}
 	return nil
 }
 
diff --git a/cmd/erasure-server-pool-delete-all.go b/cmd/erasure-server-pool-delete-all.go
index 47b8ba229..670c98d0d 100644
--- a/cmd/erasure-server-pool-delete-all.go
+++ b/cmd/erasure-server-pool-delete-all.go
@@ -300,11 +300,9 @@ func (z *erasureServerPools) deleteAllVersions(ctx context.Context, bucket, obje
 	// Check every write before deleting any: the accepted revisions, then
 	// every revision on every drive that the purge will delete on.
 	ids := make(map[string]bool, len(history))
-	held := make([]bool, len(z.serverPools))
 	for _, v := range history {
 		ids[v.id] = true
 		for _, c := range v.copies {
-			held[c.pool] = true
 			if err := deleteAllRefuses(ctx, bucket, object, c.fi, lr.LockEnabled); err != nil {
 				return fail(err)
 			}
@@ -324,9 +322,6 @@ func (z *erasureServerPools) deleteAllVersions(ctx context.Context, bucket, obje
 	}
 	drives := make([][]StorageAPI, len(z.serverPools))
 	for pool := range z.serverPools {
-		if !held[pool] {
-			continue
-		}
 		if drives[pool], err = z.serverPools[pool].getHashedSet(object).deleteAllDrives(ctx, bucket, object, ids, lr.LockEnabled, writes); err != nil {
 			return fail(err)
 		}
@@ -432,7 +427,7 @@ func (er erasureObjects) deleteAllDrives(ctx context.Context, bucket, object str
 		return nil, err
 	}
 	for i := range disks {
-		if errors.Is(errs[i], errFileNotFound) {
+		if errors.Is(errs[i], errFileNotFound) || errors.Is(errs[i], errVolumeNotFound) {
 			continue
 		}
 		if errs[i] != nil {
@@ -447,6 +442,11 @@ func (er erasureObjects) deleteAllDrives(ctx context.Context, bucket, object str
 			return nil, errDeleteAllHistoryUnavailable
 		}
 		for _, fi := range versions {
+			// A minority-drive or other-pool revision can be hidden by the
+			// logical history merge. It still owns its remote object.
+			if tierRefusesDeleteAll(fi.TierFreeVersion(), fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName) {
+				return nil, errDeleteAllTiered
+			}
 			if fi.TierFreeVersion() || !ids[deleteAllVersionID(fi.VersionID)] {
 				continue
 			}
diff --git a/cmd/erasure-server-pool-lifecycle.go b/cmd/erasure-server-pool-lifecycle.go
index ab76c53c2..d3b652b97 100644
--- a/cmd/erasure-server-pool-lifecycle.go
+++ b/cmd/erasure-server-pool-lifecycle.go
@@ -780,6 +780,16 @@ func (z *erasureServerPools) migrationExpiryApplies(ctx context.Context, oi Obje
 		opts.VersionID = oi.VersionID
 	}
 	if evt.Action.DeleteAll() {
+		// A whole-key purge cannot handle any live tier revision, including
+		// one hidden by a duplicate-version merge or a minority drive. Move
+		// these versions instead, retaining their remote cleanup ownership.
+		if h == nil {
+			h = &migrationHistory{}
+		}
+		h.loadTier(ctx, z, oi.Bucket, oi.Name)
+		if h.tierBlocked {
+			return false
+		}
 		opts.DeletePrefix = true
 	}
 	opts.Transition.ExpireRestored = evt.Action.DeleteRestored()
@@ -807,6 +817,20 @@ var migrationHistoryLoaded func(object string)
 // migrationHistory is the history of one key in all pools, read once by a
 // decommission or rebalance walk for the prechecks of its versions.
 type migrationHistory struct {
-	loaded bool
-	all    lifecycleAllPools
+	tierLoaded  bool
+	tierBlocked bool
+	loaded      bool
+	all         lifecycleAllPools
+}
+
+// Load before moving the first version: moving a hot version can overwrite a
+// minority tier revision in the destination, turning it into a free record.
+// That must not change the whole-key expiry decision halfway through a walk.
+func (h *migrationHistory) loadTier(ctx context.Context, z *erasureServerPools, bucket, object string) {
+	if h.tierLoaded {
+		return
+	}
+	tier, err := z.readTierHistory(ctx, bucket, encodeDirObject(object))
+	h.tierBlocked = err != nil || len(tier.liveTuples) != 0
+	h.tierLoaded = true
 }
diff --git a/cmd/erasure-server-pool-rebalance.go b/cmd/erasure-server-pool-rebalance.go
index 49e10e584..1a46b4be7 100644
--- a/cmd/erasure-server-pool-rebalance.go
+++ b/cmd/erasure-server-pool-rebalance.go
@@ -594,6 +594,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
 			if lc == nil {
 				return lifecycle.NoneAction
 			}
+			history.loadTier(ctx, z, bucket, object)
 			versioned := vc != nil && vc.Versioned(object)
 			objInfo := fi.ToObjectInfo(bucket, object, versioned)
 
diff --git a/cmd/erasure-server-pool-tier-ownership.go b/cmd/erasure-server-pool-tier-ownership.go
new file mode 100644
index 000000000..4facddaf4
--- /dev/null
+++ b/cmd/erasure-server-pool-tier-ownership.go
@@ -0,0 +1,208 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"slices"
+)
+
+// tierRemoteTuple is the complete identity of an object in a remote tier.
+// The tier name is part of the identity: configuring a new backend under a
+// different name must not make an old free record point at it.
+type tierRemoteTuple struct {
+	Tier      string
+	Object    string
+	VersionID string
+}
+
+func (t tierRemoteTuple) equal(other tierRemoteTuple) bool {
+	return t == other
+}
+
+// tierFreeRecord retains the exact drive and free record that owns a remote
+// tuple. Cleanup must not delete the same version ID on another drive where
+// it may name a different tuple.
+type tierFreeRecord struct {
+	disk  StorageAPI
+	info  FileInfo
+	Tuple tierRemoteTuple
+}
+
+type tierHistory struct {
+	liveTuples []tierRemoteTuple
+	free       []tierFreeRecord
+}
+
+var (
+	errTierOwnershipUnavailable = errors.New("tier ownership history is unavailable")
+	errTierStillReferenced      = errors.New("remote tier object still has a live reference")
+)
+
+func tierOwnershipUnavailable(bucket, object string, err error) error {
+	return fmt.Errorf("%w for %s/%s: %v: %w", errTierOwnershipUnavailable, bucket, decodeDirObject(object), err, InsufficientReadQuorum{
+		Bucket: bucket,
+		Object: decodeDirObject(object),
+		Err:    errErasureReadQuorum,
+		Type:   RQInconsistentMeta,
+	})
+}
+
+// tierOwnership reads every drive in every pool and retains the transition
+// evidence of every metadata revision. It deliberately does not use
+// objectVersions or lifecycleEventsAllPools: those APIs merge duplicate
+// versions and omit free versions, which is correct for reads but unsafe for
+// deleting a remote tuple.
+//
+// The caller must hold the encoded object's writer lock. A matching live
+// reference or an unreadable drive prevents remote deletion. Matching free
+// records are returned so the caller can remove only the records that name
+// this exact remote tuple.
+func (z *erasureServerPools) tierOwnership(ctx context.Context, bucket, object string, tuple tierRemoteTuple) (live bool, free []tierFreeRecord, err error) {
+	history, err := z.readTierHistory(ctx, bucket, object)
+	if err != nil {
+		return false, nil, err
+	}
+	for _, record := range history.free {
+		if record.Tuple.equal(tuple) {
+			free = append(free, record)
+		}
+	}
+	for _, liveTuple := range history.liveTuples {
+		// Incomplete transition fields cannot prove that this is a different
+		// remote object. Keep the cleanup record until they are resolved.
+		if (liveTuple.Tier == "" || liveTuple.Tier == tuple.Tier) &&
+			(liveTuple.Object == "" || liveTuple.Object == tuple.Object) &&
+			(liveTuple.VersionID == "" || tuple.VersionID == "" || liveTuple.VersionID == tuple.VersionID) {
+			live = true
+			break
+		}
+	}
+	return live, free, nil
+}
+
+// readTierHistory is the full per-drive metadata scan used by both remote
+// cleanup and delete-all. It keeps every live tuple so a hidden older version
+// or a minority-drive revision cannot be discarded by a winner merge.
+func (z *erasureServerPools) readTierHistory(ctx context.Context, bucket, object string) (tierHistory, error) {
+	var history tierHistory
+	if z == nil || len(z.serverPools) == 0 {
+		return history, tierOwnershipUnavailable(bucket, object, errTierOwnershipUnavailable)
+	}
+
+	for _, p := range z.serverPools {
+		set := p.getHashedSet(object)
+		disks := slices.Clone(set.getDisks())
+		if len(disks) == 0 {
+			return history, tierOwnershipUnavailable(bucket, object, errTierOwnershipUnavailable)
+		}
+		raws, errs := readAllRawFileInfo(ctx, disks, bucket, object, false)
+		if err := ctx.Err(); err != nil {
+			return history, err
+		}
+		for drive := range disks {
+			err := errs[drive]
+			switch {
+			case err == nil:
+			case errors.Is(err, errFileNotFound), errors.Is(err, errVolumeNotFound):
+				continue
+			default:
+				return history, tierOwnershipUnavailable(bucket, object, err)
+			}
+
+			var xl xlMetaV2
+			if err := xl.LoadOrConvert(raws[drive].Buf); err != nil {
+				return history, tierOwnershipUnavailable(bucket, object, err)
+			}
+			versions, err := xl.ListVersions(bucket, object, false)
+			if err != nil {
+				return history, tierOwnershipUnavailable(bucket, object, err)
+			}
+			for _, fi := range versions {
+				got := tierRemoteTuple{
+					Tier:      fi.TransitionTier,
+					Object:    fi.TransitionedObjName,
+					VersionID: fi.TransitionVersionID,
+				}
+				if fi.TierFreeVersion() {
+					history.free = append(history.free, tierFreeRecord{
+						disk:  disks[drive],
+						info:  fi,
+						Tuple: got,
+					})
+					continue
+				}
+				if tierRefusesDeleteAll(false, fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName) {
+					history.liveTuples = append(history.liveTuples, got)
+				}
+			}
+		}
+	}
+	return history, nil
+}
+
+func deleteTierRemote(ctx context.Context, tuple tierRemoteTuple) error {
+	err := deleteObjectFromRemoteTier(ctx, tuple.Object, tuple.VersionID, tuple.Tier)
+	if err == nil || isErrObjectNotFound(err) || isErrVersionNotFound(err) {
+		return nil
+	}
+	return err
+}
+
+// deleteTierTuple performs the only safe remote-tier cleanup sequence. The
+// object lock covers the fresh ownership read, the remote delete and local
+// free-record cleanup. A failed remote delete leaves every matching record as
+// the durable retry point. A remote not-found is successful and still allows
+// record cleanup.
+func (z *erasureServerPools) deleteTierTuple(ctx context.Context, bucket, object string, tuple tierRemoteTuple) error {
+	object = encodeDirObject(object)
+	lk := z.NewObjectNSLock(bucket, object)
+	lkctx, err := lk.GetLock(ctx, globalDeleteOperationTimeout)
+	if err != nil {
+		return err
+	}
+	ctx = lkctx.Context()
+	defer lk.Unlock(lkctx)
+	return z.deleteTierTupleLocked(ctx, bucket, object, tuple)
+}
+
+// deleteTierTupleLocked is the same sequence for callers that already hold
+// the object's writer lock, such as a delete-all purge.
+func (z *erasureServerPools) deleteTierTupleLocked(ctx context.Context, bucket, object string, tuple tierRemoteTuple) error {
+	live, records, err := z.tierOwnership(ctx, bucket, object, tuple)
+	if err != nil {
+		return err
+	}
+	if live {
+		return errTierStillReferenced
+	}
+	if len(records) == 0 {
+		// An old queued sweeper is not evidence of ownership. In particular,
+		// it must not delete a remote object after the free record is gone.
+		return nil
+	}
+	if tuple.Tier == "" || tuple.Object == "" {
+		return tierOwnershipUnavailable(bucket, object, errTierOwnershipUnavailable)
+	}
+	if err := deleteTierRemote(ctx, tuple); err != nil {
+		return err
+	}
+
+	// Only the captured drives may lose these records. The storage operation
+	// also checks the free marker and tuple, so a stale cleanup cannot remove
+	// a different revision with the same local version ID.
+	for _, record := range records {
+		fi := FileInfo{VersionID: record.info.VersionID,
+			TransitionTier: tuple.Tier, TransitionedObjName: tuple.Object,
+			TransitionVersionID: tuple.VersionID}
+		fi.SetTierFreeVersion()
+		err := record.disk.DeleteVersion(ctx, bucket, object, fi, false, DeleteOptions{})
+		if err != nil && !errors.Is(err, errFileNotFound) && !errors.Is(err, errFileVersionNotFound) {
+			return err
+		}
+	}
+	return nil
+}
diff --git a/cmd/erasure-server-pool-tier-ownership_test.go b/cmd/erasure-server-pool-tier-ownership_test.go
new file mode 100644
index 000000000..62415025f
--- /dev/null
+++ b/cmd/erasure-server-pool-tier-ownership_test.go
@@ -0,0 +1,218 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"errors"
+	"slices"
+	"testing"
+
+	"github.com/minio/minio/internal/bucket/lifecycle"
+)
+
+func rewriteTierTestMeta(t *testing.T, disk StorageAPI, bucket, key string, change func(*xlMetaV2)) {
+	t.Helper()
+	raw, err := disk.ReadAll(t.Context(), bucket, pathJoin(key, xlStorageFormatFile))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var xl xlMetaV2
+	if err = xl.LoadOrConvert(raw); err != nil {
+		t.Fatal(err)
+	}
+	change(&xl)
+	raw, err = xl.AppendTo(nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = disk.WriteAll(t.Context(), bucket, pathJoin(key, xlStorageFormatFile), raw); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestTierCleanupRequiresDurableOwnership(t *testing.T) {
+	for _, kind := range []string{"absent", "older-live", "minority-live", "unknown", "remote-failure", "same-id-other-tuple"} {
+		t.Run(kind, func(t *testing.T) {
+			f := newFreeVersionTest(t, "")
+			const key = "shared"
+			oi := f.tiered(key, "remote", daysAgo(4))
+			tuple := tierRemoteTuple{Tier: "WARM", Object: f.remote(key, "remote")}
+			live := slices.Clone(f.xlMeta(0, key))
+			f.deleteVersion(oi)
+			disks := f.z.serverPools[0].getHashedSet(key).getDisks()
+			other := f.z.serverPools[1].getHashedSet(key).getDisks()
+			switch kind {
+			case "absent":
+				// A stale task with a different remote identity has no owning free record.
+				tuple.Object += "-stale"
+				f.tier.objects[tuple.Object] = []byte("must survive")
+			case "older-live", "minority-live":
+				for i, disk := range other {
+					if kind == "minority-live" && i > 0 {
+						break
+					}
+					if err := disk.WriteAll(t.Context(), f.bucket, pathJoin(key, xlStorageFormatFile), live); err != nil {
+						t.Fatal(err)
+					}
+				}
+				// Hide the live remote reference beneath a newer hot version.
+				putConsistencyObject(t, f.z, f.bucket, key, 1, "hot", ObjectOptions{Versioned: true})
+			case "unknown":
+				if err := other[0].WriteAll(t.Context(), f.bucket, pathJoin(key, xlStorageFormatFile), []byte("corrupt")); err != nil {
+					t.Fatal(err)
+				}
+			case "remote-failure":
+				f.tier.removeErr = errors.New("remote unavailable")
+			case "same-id-other-tuple":
+				rewriteTierTestMeta(t, disks[len(disks)-1], f.bucket, key, func(xl *xlMetaV2) {
+					ver, err := xl.getIdx(0)
+					if err != nil {
+						t.Fatal(err)
+					}
+					ver.DeleteMarker.MetaSys[metaTierObjName] = []byte("another-remote")
+					if err := xl.setIdx(0, *ver); err != nil {
+						t.Fatal(err)
+					}
+				})
+				f.tier.objects["another-remote"] = []byte("other")
+			}
+			err := f.z.deleteTierTuple(t.Context(), f.bucket, key, tuple)
+			switch kind {
+			case "older-live", "minority-live":
+				if !errors.Is(err, errTierStillReferenced) {
+					t.Fatalf("cleanup: %v", err)
+				}
+			case "unknown":
+				if !errors.Is(err, errTierOwnershipUnavailable) {
+					t.Fatalf("cleanup: %v", err)
+				}
+			case "remote-failure":
+				if !errors.Is(err, f.tier.removeErr) {
+					t.Fatalf("cleanup: %v", err)
+				}
+			default:
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+			if kind != "same-id-other-tuple" {
+				if len(f.tier.removed) != 0 {
+					t.Fatalf("deleted without sole durable ownership: %v", f.tier.removed)
+				}
+				if got := f.stored(0, key); len(got) != 1 || got[0] != "free" {
+					t.Fatalf("lost retry record: %v", got)
+				}
+			} else {
+				raw, err := disks[len(disks)-1].ReadAll(t.Context(), f.bucket, pathJoin(key, xlStorageFormatFile))
+				if err != nil {
+					t.Fatal(err)
+				}
+				versions, err := getFileInfoVersions(raw, f.bucket, key, true)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if len(versions.Versions) != 1 || versions.Versions[0].TransitionedObjName != "another-remote" {
+					t.Fatalf("removed different tuple: %+v", versions)
+				}
+				if len(f.tier.removed) != 1 || f.tier.removed[0] != tuple.Object {
+					t.Fatalf("remote deletions: %v", f.tier.removed)
+				}
+			}
+			if kind == "remote-failure" {
+				f.tier.removeErr = nil
+				if err := f.z.deleteTierTuple(t.Context(), f.bucket, key, tuple); err != nil {
+					t.Fatal(err)
+				}
+				if got := f.stored(0, key); len(got) != 0 {
+					t.Fatalf("retry left %v", got)
+				}
+				if err := f.z.deleteTierTuple(t.Context(), f.bucket, key, tuple); err != nil {
+					t.Fatal(err)
+				}
+				if len(f.tier.removed) != 1 {
+					t.Fatalf("stale retry deleted again: %v", f.tier.removed)
+				}
+			}
+		})
+	}
+}
+
+func TestTierSweeperPreservesOtherPoolOwner(t *testing.T) {
+	f := newFreeVersionTest(t, "")
+	const key = "overwrite"
+	oi := f.tiered(key, "remote", daysAgo(4))
+	queue := f.expiryQueue()
+	sweeper := newObjSweeper(f.bucket, key).WithVersion(oi.VersionID).WithVersioning(true, false)
+	sweeper.SetTransitionState(TransitionedObject{Status: lifecycle.TransitionComplete, Tier: "WARM", Name: f.remote(key, "remote")})
+	// Overwrite in the other pool while the draining pool retains its owner.
+	putConsistencyObject(t, f.z, f.bucket, key, 1, "replacement", ObjectOptions{Versioned: true})
+	sweeper.Sweep()
+	close(queue)
+	globalExpiryState.Worker(queue)
+	if len(f.tier.removed) != 0 {
+		t.Fatalf("sweeper deleted a live remote: %v", f.tier.removed)
+	}
+	if got := migrationTestBody(t, f.z.GetObjectNInfo, f.bucket, key, oi.VersionID); got != key+" remote" {
+		t.Fatalf("old version body: %q", got)
+	}
+}
+
+func TestDecommissionMovesTieredDeleteAll(t *testing.T) {
+	for _, kind := range []string{"complete", "pending", "minority-other-pool"} {
+		t.Run(kind, func(t *testing.T) {
+			f := newFreeVersionTest(t, purgeRules)
+			const key = "ea/tiered"
+			oi := f.tiered(key, "v1", daysAgo(4))
+			if kind == "pending" {
+				for _, disk := range f.z.serverPools[0].getHashedSet(key).getDisks() {
+					rewriteTierTestMeta(t, disk, f.bucket, key, func(xl *xlMetaV2) {
+						i, ver, err := xl.findVersionStr(oi.VersionID)
+						if err != nil {
+							t.Fatal(err)
+						}
+						ver.ObjectV2.MetaSys[metaTierStatus] = []byte(lifecycle.TransitionPending)
+						if err := xl.setIdx(i, *ver); err != nil {
+							t.Fatal(err)
+						}
+					})
+				}
+			}
+			if kind == "minority-other-pool" {
+				raw := slices.Clone(f.xlMeta(0, key))
+				// A newer hot copy wins the logical merge, while one other-pool disk
+				// still holds the older tier revision with the same version ID.
+				putConsistencyObject(t, f.z, f.bucket, key, 0, "hot-v1", ObjectOptions{Versioned: true, VersionID: oi.VersionID, MTime: daysAgo(3)})
+				disk := f.z.serverPools[1].getHashedSet(key).getDisks()[0]
+				if err := disk.WriteAll(t.Context(), f.bucket, pathJoin(key, xlStorageFormatFile), raw); err != nil {
+					t.Fatal(err)
+				}
+			}
+			f.put(key, "v2", daysAgo(2))
+			// Verify the precheck uses per-drive evidence, including minority data.
+			var history migrationHistory
+			latest, err := f.z.GetObjectInfo(t.Context(), f.bucket, key, ObjectOptions{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			if f.z.migrationExpiryApplies(t.Context(), latest, lifecycle.Event{Action: lifecycle.DeleteAllVersionsAction}, &history) {
+				t.Fatal("tiered purge selected instead of migration")
+			}
+			if kind == "pending" {
+				return
+			} // Pending data is tested at the precheck; a real transition owns its data move.
+			if state := f.decommission(); !state.Complete || state.Failed {
+				t.Fatalf("decommission: %+v", state)
+			}
+			if got := f.stored(0, key); len(got) != 0 {
+				t.Fatalf("source retained %v", got)
+			}
+			if got := f.stored(1, key); len(got) < 2 || !slices.Equal(got[:2], []string{"v2", "v1"}) {
+				t.Fatalf("destination versions: %v", got)
+			}
+			if slices.Contains(f.tier.removed, f.remote(key, "v1")) {
+				t.Fatal("deleted live tier object during migration")
+			}
+		})
+	}
+}
diff --git a/cmd/tier-sweeper.go b/cmd/tier-sweeper.go
index 189c06663..21c9a9858 100644
--- a/cmd/tier-sweeper.go
+++ b/cmd/tier-sweeper.go
@@ -121,9 +121,11 @@ func (os *objSweeper) shouldRemoveRemoteObject() (jentry, bool) {
 	}
 	if delTier {
 		return jentry{
-			ObjName:   os.RemoteObject,
-			VersionID: os.TransitionVersionID,
-			TierName:  os.TransitionTier,
+			Bucket:          os.Bucket,
+			Object:          os.Object,
+			RemoteObject:    os.RemoteObject,
+			RemoteVersionID: os.TransitionVersionID,
+			TierName:        os.TransitionTier,
 		}, true
 	}
 	return jentry{}, false
@@ -137,9 +139,11 @@ func (os *objSweeper) Sweep() {
 }
 
 type jentry struct {
-	ObjName   string
-	VersionID string
-	TierName  string
+	Bucket          string
+	Object          string
+	RemoteObject    string
+	RemoteVersionID string
+	TierName        string
 }
 
 func deleteObjectFromRemoteTier(ctx context.Context, objName, rvID, tierName string) error {
diff --git a/cmd/xl-storage-format-v2.go b/cmd/xl-storage-format-v2.go
index 652cab453..118ec428f 100644
--- a/cmd/xl-storage-format-v2.go
+++ b/cmd/xl-storage-format-v2.go
@@ -1374,6 +1374,19 @@ func (x *xlMetaV2) DeleteVersion(fi FileInfo) (string, error) {
 		}
 	}
 
+	if fi.TierFreeVersion() {
+		_, ver, err := x.findVersion(uv)
+		if err != nil {
+			return "", err
+		}
+		if ver.Type != DeleteType || !ver.DeleteMarker.FreeVersion() ||
+			string(ver.DeleteMarker.MetaSys[metaTierName]) != fi.TransitionTier ||
+			string(ver.DeleteMarker.MetaSys[metaTierObjName]) != fi.TransitionedObjName ||
+			string(ver.DeleteMarker.MetaSys[metaTierVersionID]) != fi.TransitionVersionID {
+			return "", errFileVersionNotFound
+		}
+	}
+
 	if fi.ExpireRestored {
 		i, ver, err := x.findVersion(uv)
 		if err != nil {