diff --git a/engine/schema/src/main/java/com/cloud/storage/dao/VolumeDao.java b/engine/schema/src/main/java/com/cloud/storage/dao/VolumeDao.java
index 4cd9a8e23bfc..8db88a18a41a 100644
--- a/engine/schema/src/main/java/com/cloud/storage/dao/VolumeDao.java
+++ b/engine/schema/src/main/java/com/cloud/storage/dao/VolumeDao.java
@@ -68,6 +68,13 @@ public interface VolumeDao extends GenericDao<VolumeVO, Long>, StateDao<Volume.S
 
     void detachVolume(long volumeId);
 
+    /**
+     * Clears the attachment of the volume only if it is still attached to the given VM.
+     *
+     * @return true if the attachment was cleared; false if the volume is no longer attached to vmId
+     */
+    boolean detachVolume(long volumeId, long vmId);
+
     boolean isAnyVolumeActivelyUsingTemplateOnPool(long templateId, long poolId);
 
     List<VolumeVO> findCreatedByInstance(long id);
diff --git a/engine/schema/src/main/java/com/cloud/storage/dao/VolumeDaoImpl.java b/engine/schema/src/main/java/com/cloud/storage/dao/VolumeDaoImpl.java
index 1d5ff5e93402..bce04c345950 100644
--- a/engine/schema/src/main/java/com/cloud/storage/dao/VolumeDaoImpl.java
+++ b/engine/schema/src/main/java/com/cloud/storage/dao/VolumeDaoImpl.java
@@ -347,6 +347,23 @@ public void detachVolume(long volumeId) {
         update(volumeId, volume);
     }
 
+    @Override
+    public boolean detachVolume(long volumeId, long vmId) {
+        VolumeVO existing = findById(volumeId);
+        VolumeVO volume = createForUpdate(volumeId);
+        volume.setInstanceId(null);
+        volume.setDeviceId(null);
+        volume.setUpdated(new Date());
+        volume.setAttached(null);
+        if (existing != null && existing.getVolumeType() == Type.ROOT) {
+            volume.setVolumeType(Type.DATADISK);
+        }
+        SearchCriteria<VolumeVO> sc = AllFieldsSearch.create();
+        sc.setParameters("id", volumeId);
+        sc.setParameters("instanceId", vmId);
+        return update(volume, sc) > 0;
+    }
+
     @Override
     @DB
     public HypervisorType getHypervisorType(long volumeId) {
diff --git a/engine/schema/src/test/java/com/cloud/storage/dao/VolumeDaoImplTest.java b/engine/schema/src/test/java/com/cloud/storage/dao/VolumeDaoImplTest.java
index 6f153727ab76..0a5af6dc5098 100644
--- a/engine/schema/src/test/java/com/cloud/storage/dao/VolumeDaoImplTest.java
+++ b/engine/schema/src/test/java/com/cloud/storage/dao/VolumeDaoImplTest.java
@@ -35,6 +35,7 @@
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
@@ -43,6 +44,8 @@
 
 import com.cloud.storage.Volume;
 import com.cloud.storage.VolumeVO;
+import com.cloud.utils.Pair;
+import com.cloud.utils.db.Attribute;
 import com.cloud.utils.db.Filter;
 import com.cloud.utils.db.SearchBuilder;
 import com.cloud.utils.db.SearchCriteria;
@@ -205,4 +208,69 @@ public void testSearchRemovedByVms() {
                 Mockito.any(SearchCriteria.class), Mockito.any(Filter.class), Mockito.eq(null),
                 Mockito.eq(false));
     }
+
+    private static final long DETACHED_VOLUME_ID = 1L;
+    private static final long DETACHED_FROM_VM_ID = 2L;
+
+    private void mockVolumeForConditionalDetach(Volume.Type existingType, int updatedRows) {
+        VolumeVO existing = Mockito.mock(VolumeVO.class);
+        Mockito.doReturn(existingType).when(existing).getVolumeType();
+        Mockito.doReturn(existing).when(volumeDao).findById(DETACHED_VOLUME_ID);
+        Mockito.doReturn(updatedRows).when(volumeDao).update(Mockito.any(VolumeVO.class), Mockito.<SearchCriteria<VolumeVO>>any());
+    }
+
+    @SuppressWarnings("unchecked")
+    private SearchCriteria<VolumeVO> captureConditionalDetachCriteria() {
+        ArgumentCaptor<SearchCriteria<VolumeVO>> criteria = ArgumentCaptor.forClass(SearchCriteria.class);
+        Mockito.verify(volumeDao).update(Mockito.any(VolumeVO.class), criteria.capture());
+        return criteria.getValue();
+    }
+
+    private List<Object> boundValuesOf(SearchCriteria<VolumeVO> criteria) {
+        List<Object> values = new ArrayList<>();
+        for (Pair<Attribute, Object> bound : criteria.getValues()) {
+            values.add(bound.second());
+        }
+        return values;
+    }
+
+    @Test
+    public void detachVolumeTestRestrictsTheUpdateToTheVmTheVolumeIsStillAttachedTo() {
+        mockVolumeForConditionalDetach(Volume.Type.DATADISK, 1);
+
+        volumeDao.detachVolume(DETACHED_VOLUME_ID, DETACHED_FROM_VM_ID);
+
+        SearchCriteria<VolumeVO> criteria = captureConditionalDetachCriteria();
+        Assert.assertTrue("the update must be restricted by instance_id, otherwise a stale detach clears a newer attachment; where clause was: "
+                + criteria.getWhereClause(), criteria.getWhereClause().contains("instance_id"));
+        List<Object> bound = boundValuesOf(criteria);
+        Assert.assertEquals("expected the volume id and the vm id to be bound, got " + bound, 2, bound.size());
+        Assert.assertTrue("the vm id must be bound, got " + bound, bound.contains(DETACHED_FROM_VM_ID));
+        Assert.assertTrue("the volume id must be bound, got " + bound, bound.contains(DETACHED_VOLUME_ID));
+    }
+
+    @Test
+    public void detachVolumeTestClearsAttachmentWhenStillAttachedToVm() {
+        mockVolumeForConditionalDetach(Volume.Type.DATADISK, 1);
+
+        Assert.assertTrue(volumeDao.detachVolume(DETACHED_VOLUME_ID, DETACHED_FROM_VM_ID));
+    }
+
+    @Test
+    public void detachVolumeTestKeepsAttachmentWhenAttachedToAnotherVm() {
+        mockVolumeForConditionalDetach(Volume.Type.DATADISK, 0);
+
+        Assert.assertFalse(volumeDao.detachVolume(DETACHED_VOLUME_ID, DETACHED_FROM_VM_ID));
+    }
+
+    @Test
+    public void detachVolumeTestConvertsADetachedRootVolumeToADataDisk() {
+        mockVolumeForConditionalDetach(Volume.Type.ROOT, 1);
+
+        volumeDao.detachVolume(DETACHED_VOLUME_ID, DETACHED_FROM_VM_ID);
+
+        ArgumentCaptor<VolumeVO> updated = ArgumentCaptor.forClass(VolumeVO.class);
+        Mockito.verify(volumeDao).update(updated.capture(), Mockito.<SearchCriteria<VolumeVO>>any());
+        Assert.assertEquals(Volume.Type.DATADISK, updated.getValue().getVolumeType());
+    }
 }
diff --git a/server/src/main/java/com/cloud/storage/VolumeApiServiceImpl.java b/server/src/main/java/com/cloud/storage/VolumeApiServiceImpl.java
index e19c11e18068..69ca4926e585 100644
--- a/server/src/main/java/com/cloud/storage/VolumeApiServiceImpl.java
+++ b/server/src/main/java/com/cloud/storage/VolumeApiServiceImpl.java
@@ -3626,8 +3626,15 @@ private Volume orchestrateDetachVolumeFromVM(long vmId, long volumeId) {
         }
 
         if (!sendCommand || (answer != null && answer.getResult())) {
-            // Mark the volume as detached
-            _volsDao.detachVolume(volume.getId());
+            // Mark the volume as detached, only if it is still attached to this VM. The host has already been told to
+            // unplug the disk by this point, so the remainder of this block is deliberately skipped when the volume
+            // moved: revoking access or publishing a detach event would assert something that is no longer true.
+            if (!_volsDao.detachVolume(volume.getId(), vmId)) {
+                Volume current = _volsDao.findById(volumeId);
+                logger.warn("Volume {} is no longer attached to VM {} (its attachment now points at VM id {}); leaving it unchanged",
+                        volume, vm, current == null ? null : current.getInstanceId());
+                return current;
+            }
 
             if (answer != null) {
                 String datastoreName = answer.getContextParam("datastoreName");
diff --git a/server/src/test/java/com/cloud/storage/VolumeApiServiceImplTest.java b/server/src/test/java/com/cloud/storage/VolumeApiServiceImplTest.java
index b40d8232c6a5..e32f6c2141c0 100644
--- a/server/src/test/java/com/cloud/storage/VolumeApiServiceImplTest.java
+++ b/server/src/test/java/com/cloud/storage/VolumeApiServiceImplTest.java
@@ -37,7 +37,9 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 
@@ -73,6 +75,7 @@
 import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreVO;
+import org.apache.cloudstack.storage.to.VolumeObjectTO;
 import org.apache.commons.collections.CollectionUtils;
 import org.junit.After;
 import org.junit.Assert;
@@ -89,8 +92,12 @@
 import org.mockito.Mockito;
 import org.mockito.Spy;
 import org.mockito.junit.MockitoJUnitRunner;
+import org.mockito.verification.VerificationMode;
 import org.springframework.test.util.ReflectionTestUtils;
 
+import com.cloud.agent.AgentManager;
+import com.cloud.agent.api.Answer;
+import com.cloud.agent.api.Command;
 import com.cloud.api.query.dao.ServiceOfferingJoinDao;
 import com.cloud.configuration.ConfigurationManager;
 import com.cloud.configuration.Resource.ResourceType;
@@ -141,6 +148,7 @@
 import com.cloud.utils.fsm.NoTransitionException;
 import com.cloud.vm.DiskProfile;
 import com.cloud.vm.UserVmManager;
+import com.cloud.vm.UserVmService;
 import com.cloud.vm.UserVmVO;
 import com.cloud.vm.VirtualMachine;
 import com.cloud.vm.VirtualMachine.State;
@@ -203,6 +211,10 @@ public class VolumeApiServiceImplTest {
     @Mock
     private UserVmManager userVmManager;
     @Mock
+    private UserVmService userVmServiceMock;
+    @Mock
+    private AgentManager agentManagerMock;
+    @Mock
     private DataCenterDao _dcDao;
     @Mock
     private ResourceLimitService resourceLimitServiceMock;
@@ -2941,4 +2953,125 @@ public void testResizeVolumeInternal_VMware_VMRunning_ShouldThrowStateGuardError
                     e.getMessage() != null && e.getMessage().contains("VM should be in"));
         }
     }
+
+    /**
+     * Sets up a detach of volume 100 from the stopped VM (id 2) that setUp() returns for any VM lookup. The stubs cover
+     * the whole detach path so that, without the stale-detach guard, the tests fail on their assertions rather than
+     * on a missing mock.
+     */
+    private VolumeVO mockVolumeForDetachFromStoppedVm(Long attachedVmId) {
+        VolumeVO volume = Mockito.mock(VolumeVO.class);
+        lenient().doReturn(100L).when(volume).getId();
+        Mockito.doReturn(attachedVmId).when(volume).getInstanceId();
+        lenient().doReturn(null).when(volume).getPoolId();
+        Mockito.doReturn(volume).when(volumeDaoMock).findById(100L);
+        lenient().when(virtualMachineManager.findClusterAndHostIdForVm(anyLong())).thenReturn(new Pair<>(null, null));
+        return volume;
+    }
+
+    private void verifyDetachUsageEvent(MockedStatic<UsageEventUtils> usageEventUtilsMocked, VerificationMode mode) {
+        usageEventUtilsMocked.verify(() -> UsageEventUtils.publishUsageEvent(eq(EventTypes.EVENT_VOLUME_DETACH), anyLong(), anyLong(), anyLong(),
+                nullable(String.class), nullable(Long.class), nullable(Long.class), nullable(Long.class), nullable(String.class), nullable(String.class),
+                nullable(Long.class), anyBoolean()), mode);
+    }
+
+    private void verifyStaleDetachSkipped(MockedStatic<UsageEventUtils> usageEventUtilsMocked) throws Exception {
+        Mockito.verify(volumeDaoMock, Mockito.never()).detachVolume(anyLong());
+        Mockito.verify(volumeDaoMock, Mockito.never()).detachVolume(anyLong(), anyLong());
+        Mockito.verify(virtualMachineManager, Mockito.never()).findClusterAndHostIdForVm(anyLong());
+        Mockito.verify(userVmManager, Mockito.never()).collectVmDiskStatistics(any());
+        Mockito.verify(agentManagerMock, Mockito.never()).send(nullable(Long.class), any(Command.class));
+        verifyDetachUsageEvent(usageEventUtilsMocked, Mockito.never());
+    }
+
+    @Test
+    public void detachVolumeViaDestroyVMTestSkipsDetachWhenVolumeAttachedToAnotherVm() throws Exception {
+        VolumeVO volume = mockVolumeForDetachFromStoppedVm(3L);
+
+        try (MockedStatic<UsageEventUtils> usageEventUtilsMocked = Mockito.mockStatic(UsageEventUtils.class)) {
+            Volume result = volumeApiServiceImpl.detachVolumeViaDestroyVM(2L, 100L);
+
+            Assert.assertSame(volume, result);
+            verifyStaleDetachSkipped(usageEventUtilsMocked);
+        }
+    }
+
+    @Test
+    public void detachVolumeViaDestroyVMTestSkipsDetachWhenVolumeAlreadyDetached() throws Exception {
+        VolumeVO volume = mockVolumeForDetachFromStoppedVm(null);
+
+        try (MockedStatic<UsageEventUtils> usageEventUtilsMocked = Mockito.mockStatic(UsageEventUtils.class)) {
+            Volume result = volumeApiServiceImpl.detachVolumeViaDestroyVM(2L, 100L);
+
+            Assert.assertSame(volume, result);
+            verifyStaleDetachSkipped(usageEventUtilsMocked);
+        }
+    }
+
+    @Test
+    public void detachVolumeViaDestroyVMTestClearsAttachmentWhenStillAttachedToVm() {
+        mockVolumeForDetachFromStoppedVm(2L);
+        Mockito.doReturn(true).when(volumeDaoMock).detachVolume(100L, 2L);
+
+        try (MockedStatic<UsageEventUtils> usageEventUtilsMocked = Mockito.mockStatic(UsageEventUtils.class)) {
+            volumeApiServiceImpl.detachVolumeViaDestroyVM(2L, 100L);
+
+            Mockito.verify(volumeDaoMock).detachVolume(100L, 2L);
+            Mockito.verify(volumeDaoMock, Mockito.never()).detachVolume(anyLong());
+            verifyDetachUsageEvent(usageEventUtilsMocked, Mockito.times(1));
+        }
+    }
+
+    @Test
+    public void detachVolumeViaDestroyVMTestKeepsAttachmentWhenReattachedDuringDetach() {
+        mockVolumeForDetachFromStoppedVm(2L);
+        Mockito.doReturn(false).when(volumeDaoMock).detachVolume(100L, 2L);
+
+        try (MockedStatic<UsageEventUtils> usageEventUtilsMocked = Mockito.mockStatic(UsageEventUtils.class)) {
+            volumeApiServiceImpl.detachVolumeViaDestroyVM(2L, 100L);
+
+            Mockito.verify(volumeDaoMock).detachVolume(100L, 2L);
+            Mockito.verify(volumeDaoMock, Mockito.never()).detachVolume(anyLong());
+            verifyDetachUsageEvent(usageEventUtilsMocked, Mockito.never());
+        }
+    }
+
+    /**
+     * The race this guard exists for only happens on the path that talks to the host: the attachment row is not
+     * cleared until after the agent round-trip returns, and a newer attach can land inside that window. The other
+     * tests here run against a stopped VM, so they never enter that branch. This one forces it.
+     */
+    @Test
+    public void detachVolumeViaDestroyVMTestKeepsAttachmentWhenReattachedDuringTheAgentRoundTrip() throws Exception {
+        VolumeVO volume = mockVolumeForDetachFromStoppedVm(2L);
+        Mockito.doReturn(1L).when(volume).getPoolId();
+
+        StoragePoolVO pool = Mockito.mock(StoragePoolVO.class);
+        Mockito.doReturn(pool).when(primaryDataStoreDaoMock).findByIdIncludingRemoved(1L);
+        Mockito.doReturn(pool).when(primaryDataStoreDaoMock).findById(1L);
+        Mockito.doReturn(true).when(volumeApiServiceImpl).isSendCommandForVmVolumeAttachDetach(nullable(HostVO.class), Mockito.eq(pool));
+
+        VolumeInfo volumeInfo = Mockito.mock(VolumeInfo.class);
+        Mockito.doReturn(Mockito.mock(VolumeObjectTO.class)).when(volumeInfo).getTO();
+        Mockito.doReturn(volumeInfo).when(volumeDataFactoryMock).getVolume(100L);
+        Mockito.doReturn(new Pair<List<String>, Set<String>>(new ArrayList<>(), new HashSet<>()))
+                .when(volumeOrchestrationService).getVolumeCheckpointPathsAndImageStoreUrls(Mockito.eq(100L), nullable(HypervisorType.class));
+
+        Answer answer = Mockito.mock(Answer.class);
+        Mockito.doReturn(true).when(answer).getResult();
+        Mockito.doReturn(answer).when(agentManagerMock).send(nullable(Long.class), any(Command.class));
+
+        // the volume moved to another VM while the host was being asked to unplug it
+        Mockito.doReturn(false).when(volumeDaoMock).detachVolume(100L, 2L);
+
+        try (MockedStatic<UsageEventUtils> usageEventUtilsMocked = Mockito.mockStatic(UsageEventUtils.class)) {
+            volumeApiServiceImpl.detachVolumeViaDestroyVM(2L, 100L);
+
+            // the round-trip really happened - otherwise this test proves nothing the others do not
+            Mockito.verify(agentManagerMock).send(nullable(Long.class), any(Command.class));
+            Mockito.verify(volumeDaoMock).detachVolume(100L, 2L);
+            Mockito.verify(volumeDaoMock, Mockito.never()).detachVolume(anyLong());
+            verifyDetachUsageEvent(usageEventUtilsMocked, Mockito.never());
+        }
+    }
 }