diff --git a/server/src/main/java/com/cloud/storage/StorageManagerImpl.java b/server/src/main/java/com/cloud/storage/StorageManagerImpl.java
index 9f9928bfb663..d43940800e03 100644
--- a/server/src/main/java/com/cloud/storage/StorageManagerImpl.java
+++ b/server/src/main/java/com/cloud/storage/StorageManagerImpl.java
@@ -2145,6 +2145,7 @@ public void cleanupStorage(boolean recurring) {
                         }
 
                         try {
+                            cleanupSnapshotRecordsInPrimaryStorageOnly(vol);
                             VolumeInfo volumeInfo = volFactory.getVolume(vol.getId());
                             if (volumeInfo != null) {
                                 volService.ensureVolumeIsExpungeReady(vol.getId());
@@ -2292,6 +2293,47 @@ private void deleteSnapshot(SnapshotVO snapshot, SnapshotDataStoreVO snapshotDat
         }
     }
 
+    /**
+     * Cleans up snapshot DB records for snapshots that exist only on primary storage (no secondary copy).
+     * This handles the case where snapshot.backup.to.secondary=false and the volume is being deleted
+     * during VM expunge. Each primary-only snapshot is routed through the storage driver via
+     * {@link #deleteSnapshot(SnapshotVO, SnapshotDataStoreVO)} so the underlying snapshot data is
+     * actually removed before any DB state is advanced. The parent snapshot row is only marked
+     * Destroyed once no live store refs remain.
+     */
+    protected void cleanupSnapshotRecordsInPrimaryStorageOnly(VolumeVO volume) {
+        List<SnapshotVO> snapshots = _snapshotDao.listByVolumeId(volume.getId());
+        if (CollectionUtils.isEmpty(snapshots)) {
+            return;
+        }
+        for (SnapshotVO snapshot : snapshots) {
+            if (Snapshot.State.Destroyed.equals(snapshot.getState())) {
+                continue;
+            }
+            List<SnapshotDataStoreVO> snapshotsOnPrimaryStorage = _snapshotStoreDao.listBySnapshotAndDataStoreRole(snapshot.getId(), DataStoreRole.Primary);
+            List<SnapshotDataStoreVO> snapshotsOnSecondaryStorage = _snapshotStoreDao.listBySnapshotAndDataStoreRole(snapshot.getId(), DataStoreRole.Image);
+            if (CollectionUtils.isEmpty(snapshotsOnPrimaryStorage) || CollectionUtils.isNotEmpty(snapshotsOnSecondaryStorage)) {
+                continue;
+            }
+            logger.info("Cleaning up snapshot {} (primary-only, no secondary copy) as volume {} is being deleted", snapshot, volume);
+            for (SnapshotDataStoreVO snapshotOnPrimaryStorage : snapshotsOnPrimaryStorage) {
+                deleteSnapshot(snapshot, snapshotOnPrimaryStorage);
+            }
+            List<SnapshotDataStoreVO> remaining = _snapshotStoreDao.findBySnapshotIdAndNotInDestroyedHiddenState(snapshot.getId());
+            if (CollectionUtils.isEmpty(remaining)) {
+                Snapshot.State previousState = snapshot.getState();
+                snapshot.setState(Snapshot.State.Destroyed);
+                _snapshotDao.update(snapshot.getId(), snapshot);
+                annotationDao.removeByEntityType(AnnotationService.EntityType.SNAPSHOT.name(), snapshot.getUuid());
+                if (previousState != Snapshot.State.Error && previousState != Snapshot.State.Destroyed) {
+                    _resourceLimitMgr.decrementResourceCount(snapshot.getAccountId(), ResourceType.snapshot);
+                }
+            } else {
+                logger.warn("Storage driver did not remove all primary store refs for snapshot {}; leaving parent record for the next scavenger pass to retry", snapshot);
+            }
+        }
+    }
+
     private void removeSnapshotsInErrorStatus() {
         List<SnapshotVO> snapshotsInErrorStatus = _snapshotDao.listAllByStatusIncludingRemoved(Snapshot.State.Error);
         for (SnapshotVO snapshotVO : snapshotsInErrorStatus) {
diff --git a/server/src/test/java/com/cloud/storage/StorageManagerImplTest.java b/server/src/test/java/com/cloud/storage/StorageManagerImplTest.java
index 8f88800d549f..4bdc2ff067f7 100644
--- a/server/src/test/java/com/cloud/storage/StorageManagerImplTest.java
+++ b/server/src/test/java/com/cloud/storage/StorageManagerImplTest.java
@@ -18,6 +18,7 @@
 
 import java.lang.reflect.Field;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
@@ -47,6 +48,8 @@
 import org.apache.cloudstack.storage.datastore.db.ObjectStoreDao;
 import org.apache.cloudstack.storage.datastore.db.ObjectStoreVO;
 import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
+import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreDao;
+import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreVO;
 import org.apache.cloudstack.storage.datastore.db.StoragePoolDetailVO;
 import org.apache.cloudstack.storage.datastore.db.StoragePoolDetailsDao;
 import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
@@ -85,6 +88,7 @@
 import com.cloud.host.Host;
 import com.cloud.hypervisor.Hypervisor.HypervisorType;
 import com.cloud.hypervisor.HypervisorGuruManager;
+import com.cloud.storage.dao.SnapshotDao;
 import com.cloud.storage.dao.VolumeDao;
 import com.cloud.user.AccountManagerImpl;
 import com.cloud.utils.Pair;
@@ -129,6 +133,18 @@ public class StorageManagerImplTest {
     AccountManagerImpl accountMgr;
     @Mock
     StoragePoolDetailsDao storagePoolDetailsDao;
+    @Mock
+    SnapshotDao snapshotDao;
+    @Mock
+    SnapshotDataStoreDao snapshotStoreDao;
+    @Mock
+    org.apache.cloudstack.engine.subsystem.api.storage.SnapshotDataFactory snapshotFactory;
+    @Mock
+    org.apache.cloudstack.engine.subsystem.api.storage.SnapshotService _snapshotService;
+    @Mock
+    com.cloud.user.ResourceLimitService _resourceLimitMgr;
+    @Mock
+    org.apache.cloudstack.annotation.dao.AnnotationDao annotationDao;
 
     @Mock
     ClusterDao clusterDao;
@@ -1716,4 +1732,113 @@ public void testDiscoverObjectStoreInitializationFailure() {
 
         storageManagerImpl.discoverObjectStore(name, url, size, providerName, details);
     }
+
+    @Test
+    public void testCleanupSnapshotRecordsInPrimaryStorageOnly() {
+        VolumeVO volume = Mockito.mock(VolumeVO.class);
+        Mockito.when(volume.getId()).thenReturn(1L);
+
+        SnapshotVO snapshot = Mockito.mock(SnapshotVO.class);
+        Mockito.when(snapshot.getId()).thenReturn(10L);
+        Mockito.when(snapshot.getAccountId()).thenReturn(7L);
+        Mockito.when(snapshot.getUuid()).thenReturn("snap-uuid");
+        Mockito.when(snapshot.getState()).thenReturn(Snapshot.State.BackedUp);
+        Mockito.when(snapshotDao.listByVolumeId(1L)).thenReturn(List.of(snapshot));
+
+        SnapshotDataStoreVO snapshotOnPrimaryStorage = Mockito.mock(SnapshotDataStoreVO.class);
+        Mockito.when(snapshotOnPrimaryStorage.getSnapshotId()).thenReturn(10L);
+        Mockito.when(snapshotOnPrimaryStorage.getDataStoreId()).thenReturn(50L);
+        Mockito.when(snapshotOnPrimaryStorage.getRole()).thenReturn(DataStoreRole.Primary);
+        Mockito.when(snapshotStoreDao.listBySnapshotAndDataStoreRole(10L, DataStoreRole.Primary)).thenReturn(List.of(snapshotOnPrimaryStorage));
+        Mockito.when(snapshotStoreDao.listBySnapshotAndDataStoreRole(10L, DataStoreRole.Image)).thenReturn(Collections.emptyList());
+
+        org.apache.cloudstack.engine.subsystem.api.storage.SnapshotInfo snapshotInfo =
+                Mockito.mock(org.apache.cloudstack.engine.subsystem.api.storage.SnapshotInfo.class);
+        DataStore dataStore = Mockito.mock(DataStore.class);
+        Mockito.when(dataStore.getUuid()).thenReturn("ds-uuid");
+        Mockito.when(snapshotInfo.getDataStore()).thenReturn(dataStore);
+        Mockito.when(snapshotFactory.getSnapshotIncludingRemoved(10L, 50L, DataStoreRole.Primary)).thenReturn(snapshotInfo);
+        Mockito.when(_snapshotService.deleteSnapshot(snapshotInfo)).thenReturn(true);
+        Mockito.when(snapshotStoreDao.findBySnapshotIdAndNotInDestroyedHiddenState(10L)).thenReturn(Collections.emptyList());
+
+        storageManagerImpl.cleanupSnapshotRecordsInPrimaryStorageOnly(volume);
+
+        Mockito.verify(_snapshotService).deleteSnapshot(snapshotInfo);
+        Mockito.verify(snapshot).setState(Snapshot.State.Destroyed);
+        Mockito.verify(snapshotDao).update(10L, snapshot);
+        Mockito.verify(_resourceLimitMgr).decrementResourceCount(7L, com.cloud.configuration.Resource.ResourceType.snapshot);
+        Mockito.verify(annotationDao).removeByEntityType(
+                org.apache.cloudstack.annotation.AnnotationService.EntityType.SNAPSHOT.name(), "snap-uuid");
+    }
+
+    @Test
+    public void testCleanupSnapshotRecordsInPrimaryStorageOnlyLeavesParentWhenStorageDeleteFails() {
+        VolumeVO volume = Mockito.mock(VolumeVO.class);
+        Mockito.when(volume.getId()).thenReturn(1L);
+
+        SnapshotVO snapshot = Mockito.mock(SnapshotVO.class);
+        Mockito.when(snapshot.getId()).thenReturn(10L);
+        Mockito.when(snapshot.getState()).thenReturn(Snapshot.State.BackedUp);
+        Mockito.when(snapshotDao.listByVolumeId(1L)).thenReturn(List.of(snapshot));
+
+        SnapshotDataStoreVO snapshotOnPrimaryStorage = Mockito.mock(SnapshotDataStoreVO.class);
+        Mockito.when(snapshotOnPrimaryStorage.getSnapshotId()).thenReturn(10L);
+        Mockito.when(snapshotOnPrimaryStorage.getDataStoreId()).thenReturn(50L);
+        Mockito.when(snapshotOnPrimaryStorage.getRole()).thenReturn(DataStoreRole.Primary);
+        Mockito.when(snapshotStoreDao.listBySnapshotAndDataStoreRole(10L, DataStoreRole.Primary)).thenReturn(List.of(snapshotOnPrimaryStorage));
+        Mockito.when(snapshotStoreDao.listBySnapshotAndDataStoreRole(10L, DataStoreRole.Image)).thenReturn(Collections.emptyList());
+
+        org.apache.cloudstack.engine.subsystem.api.storage.SnapshotInfo snapshotInfo =
+                Mockito.mock(org.apache.cloudstack.engine.subsystem.api.storage.SnapshotInfo.class);
+        DataStore dataStore = Mockito.mock(DataStore.class);
+        Mockito.when(dataStore.getUuid()).thenReturn("ds-uuid");
+        Mockito.when(snapshotInfo.getDataStore()).thenReturn(dataStore);
+        Mockito.when(snapshotFactory.getSnapshotIncludingRemoved(10L, 50L, DataStoreRole.Primary)).thenReturn(snapshotInfo);
+        Mockito.when(_snapshotService.deleteSnapshot(snapshotInfo)).thenReturn(false);
+        Mockito.when(snapshotStoreDao.findBySnapshotIdAndNotInDestroyedHiddenState(10L)).thenReturn(List.of(snapshotOnPrimaryStorage));
+
+        storageManagerImpl.cleanupSnapshotRecordsInPrimaryStorageOnly(volume);
+
+        Mockito.verify(_snapshotService).deleteSnapshot(snapshotInfo);
+        Mockito.verify(snapshot, Mockito.never()).setState(Snapshot.State.Destroyed);
+        Mockito.verify(snapshotDao, Mockito.never()).update(Mockito.anyLong(), Mockito.any(SnapshotVO.class));
+        Mockito.verifyNoInteractions(_resourceLimitMgr);
+        Mockito.verifyNoInteractions(annotationDao);
+    }
+
+    @Test
+    public void testCleanupSnapshotRecordsInPrimaryStorageOnlySkipsWhenSecondaryExists() {
+        VolumeVO volume = Mockito.mock(VolumeVO.class);
+        Mockito.when(volume.getId()).thenReturn(1L);
+
+        SnapshotVO snapshot = Mockito.mock(SnapshotVO.class);
+        Mockito.when(snapshot.getId()).thenReturn(10L);
+        Mockito.when(snapshot.getState()).thenReturn(Snapshot.State.BackedUp);
+        Mockito.when(snapshotDao.listByVolumeId(1L)).thenReturn(List.of(snapshot));
+
+        SnapshotDataStoreVO snapshotOnPrimaryStorage = Mockito.mock(SnapshotDataStoreVO.class);
+        SnapshotDataStoreVO snapshotOnSecondaryStorage = Mockito.mock(SnapshotDataStoreVO.class);
+        Mockito.when(snapshotStoreDao.listBySnapshotAndDataStoreRole(10L, DataStoreRole.Primary)).thenReturn(List.of(snapshotOnPrimaryStorage));
+        Mockito.when(snapshotStoreDao.listBySnapshotAndDataStoreRole(10L, DataStoreRole.Image)).thenReturn(List.of(snapshotOnSecondaryStorage));
+
+        storageManagerImpl.cleanupSnapshotRecordsInPrimaryStorageOnly(volume);
+
+        Mockito.verifyNoInteractions(_snapshotService);
+        Mockito.verify(snapshotDao, Mockito.never()).update(Mockito.anyLong(), Mockito.any(SnapshotVO.class));
+    }
+
+    @Test
+    public void testCleanupSnapshotRecordsInPrimaryStorageOnlySkipsDestroyedSnapshots() {
+        VolumeVO volume = Mockito.mock(VolumeVO.class);
+        Mockito.when(volume.getId()).thenReturn(1L);
+
+        SnapshotVO snapshot = Mockito.mock(SnapshotVO.class);
+        Mockito.when(snapshot.getState()).thenReturn(Snapshot.State.Destroyed);
+        Mockito.when(snapshotDao.listByVolumeId(1L)).thenReturn(List.of(snapshot));
+
+        storageManagerImpl.cleanupSnapshotRecordsInPrimaryStorageOnly(volume);
+
+        Mockito.verify(snapshotStoreDao, Mockito.never()).listBySnapshotAndDataStoreRole(Mockito.anyLong(), Mockito.any());
+        Mockito.verifyNoInteractions(_snapshotService);
+    }
 }