diff --git a/api/src/main/java/com/cloud/event/EventTypes.java b/api/src/main/java/com/cloud/event/EventTypes.java
index 54ba385ec6c0..b87c231b6e96 100644
--- a/api/src/main/java/com/cloud/event/EventTypes.java
+++ b/api/src/main/java/com/cloud/event/EventTypes.java
@@ -882,6 +882,7 @@ public class EventTypes {
 
     // Backup
     public static final String EVENT_HOST_AGENT_INSTALL = "HOST.AGENT.INSTALL";
+    public static final String EVENT_BACKUP_AGENT_INSTALL = "BACKUP.AGENT.INSTALL";
 
     static {
 
@@ -1433,6 +1434,7 @@ public class EventTypes {
         entityEventDetails.put(EVENT_BACKUP_REPOSITORY_UPDATE, BackupRepositoryService.class);
         // Backup
         entityEventDetails.put(EVENT_HOST_AGENT_INSTALL, Backup.class);
+        entityEventDetails.put(EVENT_BACKUP_AGENT_INSTALL, Host.class);
     }
 
     public static boolean isNetworkEvent(String eventType) {
diff --git a/api/src/main/java/org/apache/cloudstack/api/response/UserVmResponse.java b/api/src/main/java/org/apache/cloudstack/api/response/UserVmResponse.java
index 9ed99447ca74..587ffda691b2 100644
--- a/api/src/main/java/org/apache/cloudstack/api/response/UserVmResponse.java
+++ b/api/src/main/java/org/apache/cloudstack/api/response/UserVmResponse.java
@@ -360,6 +360,10 @@ public class UserVmResponse extends BaseResponseWithTagInformation implements Co
     @Param(description = "device ID of the volume currently being flattened by SharedMountPoint fast clone.")
     private Long cloneFastFlattenDeviceId;
 
+    @SerializedName("activebackupstatus")
+    @Param(description = "Active backup status of the virtual machine.")
+    private String activeBackupStatus;
+
     @SerializedName("readonlydetails")
     @Param(description = "List of read-only Instance details as comma separated string.", since = "4.16.0")
     private String readOnlyDetails;
@@ -1175,6 +1179,10 @@ public void setCloneFastFlattenDeviceId(Long cloneFastFlattenDeviceId) {
         this.cloneFastFlattenDeviceId = cloneFastFlattenDeviceId;
     }
 
+    public void setActiveBackupStatus(String activeBackupStatus) {
+        this.activeBackupStatus = activeBackupStatus;
+    }
+
     public void setReadOnlyDetails(String readOnlyDetails) {
         this.readOnlyDetails = readOnlyDetails;
     }
@@ -1223,6 +1231,10 @@ public Long getCloneFastFlattenDeviceId() {
         return cloneFastFlattenDeviceId;
     }
 
+    public String getActiveBackupStatus() {
+        return activeBackupStatus;
+    }
+
     public String getReadOnlyDetails() {
         return readOnlyDetails;
     }
diff --git a/engine/schema/src/main/java/com/cloud/event/dao/EventDao.java b/engine/schema/src/main/java/com/cloud/event/dao/EventDao.java
index c50451b03e4a..36fff2b900b0 100644
--- a/engine/schema/src/main/java/com/cloud/event/dao/EventDao.java
+++ b/engine/schema/src/main/java/com/cloud/event/dao/EventDao.java
@@ -33,6 +33,8 @@ public interface EventDao extends GenericDao<EventVO, Long> {
 
     public List<EventVO> listToArchiveOrDeleteEvents(List<Long> ids, String type, Date startDate, Date endDate, List<Long> accountIds);
 
+    boolean existsByTypeAndResource(String type, long resourceId, String resourceType);
+
     public void archiveEvents(List<EventVO> events);
 
 }
diff --git a/engine/schema/src/main/java/com/cloud/event/dao/EventDaoImpl.java b/engine/schema/src/main/java/com/cloud/event/dao/EventDaoImpl.java
index e748e98900eb..c519309999cc 100644
--- a/engine/schema/src/main/java/com/cloud/event/dao/EventDaoImpl.java
+++ b/engine/schema/src/main/java/com/cloud/event/dao/EventDaoImpl.java
@@ -98,6 +98,16 @@ public List<EventVO> listToArchiveOrDeleteEvents(List<Long> ids, String type, Da
         return search(sc, null);
     }
 
+    @Override
+    public boolean existsByTypeAndResource(String type, long resourceId, String resourceType) {
+        SearchCriteria<EventVO> sc = createSearchCriteria();
+        sc.addAnd("type", Op.EQ, type);
+        sc.addAnd("resourceId", Op.EQ, resourceId);
+        sc.addAnd("resourceType", Op.EQ, resourceType);
+        sc.addAnd("archived", Op.EQ, false);
+        return findOneIncludingRemovedBy(sc) != null;
+    }
+
     @Override
     public void archiveEvents(List<EventVO> events) {
         if (events != null && !events.isEmpty()) {
diff --git a/plugins/backup/ablestack-commvault/src/main/java/org/apache/cloudstack/backup/AblestackCommvaultBackupProvider.java b/plugins/backup/ablestack-commvault/src/main/java/org/apache/cloudstack/backup/AblestackCommvaultBackupProvider.java
index 3e9abbb12698..9f433f96a1bf 100644
--- a/plugins/backup/ablestack-commvault/src/main/java/org/apache/cloudstack/backup/AblestackCommvaultBackupProvider.java
+++ b/plugins/backup/ablestack-commvault/src/main/java/org/apache/cloudstack/backup/AblestackCommvaultBackupProvider.java
@@ -51,6 +51,7 @@
 import com.cloud.utils.exception.CloudRuntimeException;
 import com.cloud.event.ActionEventUtils;
 import com.cloud.event.EventTypes;
+import com.cloud.event.dao.EventDao;
 import com.cloud.vm.VMInstanceVO;
 import com.cloud.vm.VirtualMachine;
 import com.cloud.vm.dao.VMInstanceDao;
@@ -90,7 +91,6 @@
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.Date;
-import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
@@ -105,6 +105,7 @@
 import java.util.Comparator;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import java.util.concurrent.TimeUnit;
 import javax.inject.Inject;
 
 import static org.apache.cloudstack.backup.BackupManager.BackupChainSize;
@@ -119,6 +120,8 @@ public class AblestackCommvaultBackupProvider extends AdapterBase implements Bac
     private static final String BACKUP_TYPE_INCREMENTAL = "INCREMENTAL";
     private static final String BACKUP_ENGINE_QCOW2 = "QCOW2";
     private static final String BACKUP_ENGINE_RBD_DIFF = "RBD_DIFF";
+    private static final long COMMVAULT_INSTALL_JOB_POLL_INTERVAL_MS = TimeUnit.SECONDS.toMillis(30);
+    private static final long COMMVAULT_INSTALL_JOB_WAIT_TIMEOUT_MS = TimeUnit.HOURS.toMillis(12);
     private static final String DETAIL_CHECKPOINT_NAME = "commvault.checkpoint.name";
     private static final String DETAIL_CHECKPOINT_PATH = "commvault.checkpoint.path";
     private static final String DETAIL_CHECKPOINT_XML = "commvault.checkpoint.xml";
@@ -138,6 +141,7 @@ public class AblestackCommvaultBackupProvider extends AdapterBase implements Bac
     private static final String DETAIL_FAILURE_PHASE = "commvault.failure.phase";
     private static final String DETAIL_FAILURE_REASON = "commvault.failure.reason";
     private static final String ERROR_REASON_METADATA_FINALIZE = "metadata-finalize";
+    private static final String COMMVAULT_PERMANENT_INSTALL_FAILURE_MESSAGE = "Commvault backup agent automatic installation cannot continue because required install media is missing in the Commvault Software Cache.";
     private static final int BASE_MAJOR = 11;
     private static final int BASE_FR = 32;
     private static final int BASE_MT = 89;
@@ -226,6 +230,9 @@ public class AblestackCommvaultBackupProvider extends AdapterBase implements Bac
     @Inject
     private DiskOfferingDao diskOfferingDao;
 
+    @Inject
+    private EventDao eventDao;
+
 
     private Long getClusterIdFromRootVolume(VirtualMachine vm) {
         VolumeVO rootVolume = volumeDao.getInstanceRootVolume(vm.getId());
@@ -2259,73 +2266,97 @@ private boolean reconcileErrorBackupWithCompletedJob(VirtualMachine vm, Backup b
 
     @Override
     public boolean checkBackupAgent(final Long zoneId) {
-        Map<String, String> checkResult = new HashMap<>();
         final AblestackCommvaultClient client = getClient(zoneId);
         String csVersionInfo = client.getCvtVersion();
         boolean version = versionCheck(csVersionInfo);
         if (version) {
             List<HostVO> Hosts = hostDao.findByDataCenterId(zoneId);
+            if (CollectionUtils.isEmpty(Hosts)) {
+                LOG.warn("No hosts found in zone [{}] for Commvault backup agent readiness check.", zoneId);
+                return false;
+            }
+            int targetHostCount = 0;
             for (final HostVO host : Hosts) {
                 if (host.getStatus() == Status.Up && host.getHypervisorType() == Hypervisor.HypervisorType.KVM) {
+                    targetHostCount++;
                     String checkHost = client.getClientId(host.getName());
                     if (checkHost == null) {
+                        LOG.info("Commvault client is not registered for host [{}] using host name [{}].", host.getPrivateIpAddress(), host.getName());
                         return false;
                     } else {
                         boolean installJob = client.getInstallActiveJob(host.getPrivateIpAddress());
                         boolean checkInstall = client.getClientProps(checkHost);
                         if (installJob || !checkInstall) {
                             if (!checkInstall) {
-                                LOG.error("The host is registered with the client, but the readiness status is not normal and you must manually check the client status.");
+                                LOG.error("The host is registered with the client, but the readiness status is not normal and you must manually check the client status. host=[{}], clientId=[{}]",
+                                        host.getPrivateIpAddress(), checkHost);
                             }
                             return false;
                         }
                     }
                 }
             }
+            if (targetHostCount == 0) {
+                LOG.warn("No Up KVM hosts found in zone [{}] for Commvault backup agent readiness check. The check will be retried.", zoneId);
+                return false;
+            }
+            LOG.info("Commvault backup agent readiness check passed for zone [{}].", zoneId);
             return true;
         }
+        LOG.warn("Commvault version check failed for zone [{}]. version=[{}]", zoneId, csVersionInfo);
         return false;
     }
 
     @Override
     public boolean installBackupAgent(final Long zoneId) {
-        Map<String, String> failResult = new HashMap<>();
         final AblestackCommvaultClient client = getClient(zoneId);
         List<HostVO> Hosts = hostDao.findByDataCenterId(zoneId);
+        if (CollectionUtils.isEmpty(Hosts)) {
+            LOG.warn("No hosts found in zone [{}] for Commvault backup agent automatic installation.", zoneId);
+            return false;
+        }
+        int targetHostCount = 0;
         for (final HostVO host : Hosts) {
             if (host.getStatus() == Status.Up && host.getHypervisorType() == Hypervisor.HypervisorType.KVM) {
+                targetHostCount++;
                 String commCell = client.getCommcell();
                 JSONObject jsonObject = new JSONObject(commCell);
                 String commCellId = String.valueOf(jsonObject.get("commCellId"));
                 String commServeHostName = String.valueOf(jsonObject.get("commCellName"));
                 Ternary<String, String, String> credentials = getKVMHyperisorCredentials(host);
-                boolean installJob = true;
-                LOG.info("checking for install agent on the Commvault Backup Provider in host " + host.getPrivateIpAddress());
                 // 설치가 진행중인 호스트가 있는지 확인
-                while (installJob) {
-                    installJob = client.getInstallActiveJob(host.getName());
-                    try {
-                        Thread.sleep(30000);
-                    } catch (InterruptedException e) {
-                        LOG.error("checkBackupAgent get install active job result sleep interrupted error");
-                    }
+                if (!waitForInstallActiveJobToFinish(client, host)) {
+                    publishBackupAgentInstallFailureEventIfNeeded(host);
+                    return false;
                 }
                 String checkHost = client.getClientId(host.getName());
                 // 호스트가 클라이언트에 등록되지 않은 경우
                 if (checkHost == null) {
+                    LOG.info("Commvault client is not registered for host [{}] using host name [{}]. Creating install task with client name [{}].",
+                            host.getPrivateIpAddress(), host.getName(), host.getPrivateIpAddress());
                     String jobId = client.installAgent(host.getPrivateIpAddress(), commCellId, commServeHostName, credentials.first(), credentials.second());
                     if (jobId != null) {
-                        String jobStatus = client.getJobStatus(jobId);
-                        if (!jobStatus.equalsIgnoreCase("Completed")) {
-                            LOG.error("installing agent on the Commvault Backup Provider failed jogId : " + jobId + " , jobStatus : " + jobStatus);
-                            ActionEventUtils.onActionEvent(User.UID_SYSTEM, Account.ACCOUNT_ID_SYSTEM, Domain.ROOT_DOMAIN, EventTypes.EVENT_HOST_AGENT_INSTALL,
-                                "Failed install the commvault client agent on the host : " + host.getPrivateIpAddress(), User.UID_SYSTEM, ApiCommandResourceType.Host.toString());
-                            failResult.put(host.getPrivateIpAddress(), jobId);
+                        LOG.info("Created Commvault backup agent install job [{}] for host [{}]. Waiting for completion.", jobId, host.getPrivateIpAddress());
+                        String jobStatus = client.getJobStatus(jobId, COMMVAULT_INSTALL_JOB_WAIT_TIMEOUT_MS);
+                        if (!"Completed".equalsIgnoreCase(jobStatus)) {
+                            String failureReason = client.getLastJobFailureReason();
+                            LOG.error("installing agent on the Commvault Backup Provider failed jogId : {} , jobStatus : {}, reason=[{}]",
+                                    jobId, jobStatus, failureReason);
+                            publishBackupAgentInstallFailureEventIfNeeded(host);
+                            if (isPermanentCommvaultInstallFailure(failureReason)) {
+                                throw new CloudRuntimeException(String.format("%s host=[%s], jobId=[%s], reason=[%s]",
+                                        COMMVAULT_PERMANENT_INSTALL_FAILURE_MESSAGE, host.getPrivateIpAddress(), jobId, failureReason));
+                            }
+                            return false;
                         }
+                        LOG.info("Completed Commvault backup agent install job [{}] for host [{}].", jobId, host.getPrivateIpAddress());
                     } else {
+                        LOG.error("installing agent on the Commvault Backup Provider failed to create install job on host [{}]", host.getPrivateIpAddress());
+                        publishBackupAgentInstallFailureEventIfNeeded(host);
                         return false;
                     }
                 } else {
+                    LOG.info("Commvault client [{}] already exists for host [{}]. Checking readiness.", checkHost, host.getPrivateIpAddress());
                     // 호스트가 클라이언트에는 등록되었지만 구성이 정상적으로 되지 않은 경우 준비 상태 체크
                     boolean checkInstall = client.getClientCheckReadiness(checkHost);
                     if (!checkInstall) {
@@ -2337,12 +2368,64 @@ public boolean installBackupAgent(final Long zoneId) {
                 }
             }
         }
-        if (!failResult.isEmpty()) {
+        if (targetHostCount == 0) {
+            LOG.warn("No Up KVM hosts found in zone [{}] for Commvault backup agent automatic installation. The installation will be retried.", zoneId);
             return false;
         }
         return true;
     }
 
+    private boolean waitForInstallActiveJobToFinish(AblestackCommvaultClient client, HostVO host) {
+        final long deadline = System.currentTimeMillis() + COMMVAULT_INSTALL_JOB_WAIT_TIMEOUT_MS;
+        boolean loggedWaiting = false;
+        while (hasInstallActiveJob(client, host)) {
+            if (!loggedWaiting) {
+                LOG.info("Waiting for existing Commvault backup agent install job to finish before creating a new install job. host=[{}], timeoutMillis=[{}]",
+                        host.getPrivateIpAddress(), COMMVAULT_INSTALL_JOB_WAIT_TIMEOUT_MS);
+                loggedWaiting = true;
+            }
+            if (System.currentTimeMillis() >= deadline) {
+                LOG.warn("Timed out waiting for existing Commvault client agent install job to finish. host=[{}], timeoutMillis=[{}]",
+                        host.getPrivateIpAddress(), COMMVAULT_INSTALL_JOB_WAIT_TIMEOUT_MS);
+                return false;
+            }
+            try {
+                Thread.sleep(COMMVAULT_INSTALL_JOB_POLL_INTERVAL_MS);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                LOG.error("Interrupted while waiting for Commvault client agent install job to finish. host=[{}]",
+                        host.getPrivateIpAddress(), e);
+                return false;
+            }
+        }
+        return true;
+    }
+
+    private boolean hasInstallActiveJob(AblestackCommvaultClient client, HostVO host) {
+        return client.getInstallActiveJob(host.getName()) || client.getInstallActiveJob(host.getPrivateIpAddress());
+    }
+
+    private boolean isPermanentCommvaultInstallFailure(String failureReason) {
+        if (StringUtils.isBlank(failureReason)) {
+            return false;
+        }
+        String normalizedReason = failureReason.toLowerCase(Locale.ROOT);
+        return normalizedReason.contains("software cache") &&
+                (normalizedReason.contains("required media version") || normalizedReason.contains("missing"));
+    }
+
+    private void publishBackupAgentInstallFailureEventIfNeeded(HostVO host) {
+        if (hasBackupAgentInstallFailureEvent(host.getId())) {
+            return;
+        }
+        ActionEventUtils.onActionEvent(User.UID_SYSTEM, Account.ACCOUNT_ID_SYSTEM, Domain.ROOT_DOMAIN, EventTypes.EVENT_BACKUP_AGENT_INSTALL,
+                "Failed to install the Commvault backup agent on host: " + host.getPrivateIpAddress(), host.getId(), ApiCommandResourceType.Host.toString());
+    }
+
+    private boolean hasBackupAgentInstallFailureEvent(long hostId) {
+        return eventDao.existsByTypeAndResource(EventTypes.EVENT_BACKUP_AGENT_INSTALL, hostId, ApiCommandResourceType.Host.toString());
+    }
+
     @Override
     public boolean importBackupPlan(final Long zoneId, final String retentionPeriod, final String externalId) {
         final AblestackCommvaultClient client = getClient(zoneId);
diff --git a/plugins/backup/ablestack-commvault/src/main/java/org/apache/cloudstack/backup/commvault/AblestackCommvaultClient.java b/plugins/backup/ablestack-commvault/src/main/java/org/apache/cloudstack/backup/commvault/AblestackCommvaultClient.java
index c6f4d9ee6675..c64542d264f6 100644
--- a/plugins/backup/ablestack-commvault/src/main/java/org/apache/cloudstack/backup/commvault/AblestackCommvaultClient.java
+++ b/plugins/backup/ablestack-commvault/src/main/java/org/apache/cloudstack/backup/commvault/AblestackCommvaultClient.java
@@ -78,6 +78,7 @@ public class AblestackCommvaultClient {
     private String cvtServerUsername;
     private String cvtServerPassword;
     private final int cvtServerPort = 22;
+    private String lastJobFailureReason;
 
     public AblestackCommvaultClient(final String url, final String username, final String password, final boolean validateCertificate, final int timeout) throws URISyntaxException, NoSuchAlgorithmException, KeyManagementException {
 
@@ -1151,11 +1152,22 @@ public String createBackup(String subclientId, String storagePolicyId, String di
     // POST https://<commserveIp>/commandcenter/api/jobDetails
     // 작업의 상세정보를 조회하는 API로 작업이 완료된 경우 최종 작업 상태를 반환
     public String getJobStatus(String jobId) {
+        return getJobStatus(jobId, 0);
+    }
+
+    public String getJobStatus(String jobId, long timeoutMillis) {
         String jobStatus = "Running";
         String errorStatus = "Failed";
         HttpURLConnection connection = null;
+        lastJobFailureReason = null;
         Set<String> runningStates = Set.of("Not Started", "Running", "Pending", "Waiting", "Queued", "Suspended", "Not started");
+        long deadline = timeoutMillis > 0 ? System.currentTimeMillis() + timeoutMillis : Long.MAX_VALUE;
         while (runningStates.contains(jobStatus)) {
+            if (System.currentTimeMillis() >= deadline) {
+                LOG.warn("Timed out waiting for Commvault job [{}] to complete. lastStatus=[{}], timeoutMillis=[{}]",
+                        jobId, jobStatus, timeoutMillis);
+                return "TimedOut";
+            }
             String postUrl = apiURI.toString() + "/jobDetails";
             try {
                 URL url = new URL(postUrl);
@@ -1191,11 +1203,13 @@ public String getJobStatus(String jobId) {
                     jobStatus = jsonObject.getJSONObject("job").getJSONObject("jobDetail").getJSONObject("progressInfo").getString("state");
                     if (jobStatus.equals(errorStatus)) {
                         String errorMessage = jsonObject.getJSONObject("job").getJSONObject("jobDetail").getJSONObject("progressInfo").getString("reasonForJobDelay");
+                        lastJobFailureReason = errorMessage;
                         LOG.error("commvault job failed reason : " + errorMessage);
                     }
                     try {
                         Thread.sleep(30000);
                     } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
                         LOG.error("getJobDetails result sleep interrupted error");
                         break;
                     }
@@ -1214,6 +1228,10 @@ public String getJobStatus(String jobId) {
         return jobStatus;
     }
 
+    public String getLastJobFailureReason() {
+        return lastJobFailureReason;
+    }
+
     // POST https://<commserveIp>/commandcenter/api/jobDetails
     // 작업의 상세 정보 조회하는 API
     public String getJobDetails(String jobId) {
@@ -1532,6 +1550,7 @@ public String getCommcell() {
             if (!commCell.isMissingNode()) {
                 return commCell.toString();
             }
+            LOG.warn("Commvault commcell response did not contain commCellInfo.commCellEntity.");
         } catch (final IOException e) {
             LOG.error("Failed to request getCommcell commvault api due to : ", e);
             checkResponseTimeOut(e);
@@ -1552,6 +1571,7 @@ public String getCvtVersion() {
             if (!csVersionInfo.isMissingNode()) {
                 return csVersionInfo.toString();
             }
+            LOG.warn("Commvault version response did not contain csVersionInfo.");
         } catch (final IOException e) {
             LOG.error("Failed to request getCvtVersion commvault api due to : ", e);
             checkResponseTimeOut(e);
@@ -1740,6 +1760,9 @@ public boolean getActiveJob(String vmName) {
     // GET https://<commserveIP>/commandcenter/api/Job?jobCategory=Active
     // 실행중인 Job 조회 API로, 호스트의 에이전트 설치 작업이 실행중인 경우 true 반환
     public boolean getInstallActiveJob(String hostName) {
+        if (StringUtils.isBlank(hostName)) {
+            return false;
+        }
         try {
             final HttpResponse response = get("/Job?jobCategory=Active");
             checkResponseOK(response);
@@ -1749,18 +1772,13 @@ public boolean getInstallActiveJob(String hostName) {
             JsonNode jobs = root.get("jobs");
             if (jobs != null && jobs.isArray()) {
                 for (JsonNode item : jobs) {
-                    JsonNode entity = item.get("jobSummary");
+                    JsonNode entity = item.path("jobSummary");
                     if (!entity.isMissingNode()) {
-                        JsonNode jobType = entity.path("jobType");
+                        JsonNode generalInfo = item.path("jobDetail").path("generalInfo");
                         JsonNode subclient = entity.path("subclient");
-                        String type = "Install Client";
-                        if (!jobType.isMissingNode() && type.equals(jobType.asText())) {
-                            if (!subclient.isMissingNode()) {
-                                JsonNode clientName = subclient.path("clientName");
-                                if (!clientName.isMissingNode() && hostName.equals(clientName.asText())) {
-                                    return true;
-                                }
-                            }
+                        if (isCommvaultBackupAgentInstallJobForHost(entity, generalInfo, subclient,
+                                hostName)) {
+                            return true;
                         }
                     }
                 }
@@ -1772,6 +1790,46 @@ public boolean getInstallActiveJob(String hostName) {
         return false;
     }
 
+    private boolean isInstallClientJobForHost(JsonNode entity, JsonNode subclient, String hostName) {
+        return StringUtils.equals("Install Client", entity.path("jobType").asText(null)) &&
+                StringUtils.equals(hostName, subclient.path("clientName").asText(null));
+    }
+
+    private boolean isCommvaultBackupAgentInstallJobForHost(JsonNode entity, JsonNode generalInfo, JsonNode subclient,
+            String hostName) {
+        return isInstallClientJobForHost(entity, subclient, hostName) ||
+                (isCommvaultSoftwareJob(entity, generalInfo) &&
+                        (isCommvaultDownloadSoftwareJob(entity, generalInfo) ||
+                                matchesInstallJobHost(entity, subclient, hostName)));
+    }
+
+    private boolean isCommvaultSoftwareJob(JsonNode entity, JsonNode generalInfo) {
+        return containsSoftwareJobText(entity.path("jobType").asText(null)) ||
+                containsSoftwareJobText(entity.path("localizedOperationName").asText(null)) ||
+                containsSoftwareJobText(generalInfo.path("operationType").asText(null));
+    }
+
+    private boolean isCommvaultDownloadSoftwareJob(JsonNode entity, JsonNode generalInfo) {
+        return containsDownloadSoftwareJobText(entity.path("jobType").asText(null)) ||
+                containsDownloadSoftwareJobText(entity.path("localizedOperationName").asText(null)) ||
+                containsDownloadSoftwareJobText(generalInfo.path("operationType").asText(null));
+    }
+
+    private boolean containsSoftwareJobText(String value) {
+        return StringUtils.containsIgnoreCase(value, "Install") || containsDownloadSoftwareJobText(value);
+    }
+
+    private boolean containsDownloadSoftwareJobText(String value) {
+        return StringUtils.containsIgnoreCase(value, "Download Software");
+    }
+
+    private boolean matchesInstallJobHost(JsonNode entity, JsonNode subclient, String hostName) {
+        return StringUtils.equalsIgnoreCase(hostName, subclient.path("clientName").asText(null)) ||
+                StringUtils.equalsIgnoreCase(hostName, entity.path("clientName").asText(null)) ||
+                StringUtils.equalsIgnoreCase(hostName, entity.path("server").asText(null)) ||
+                StringUtils.equalsIgnoreCase(hostName, entity.path("client").asText(null));
+    }
+
     public static String extractJobIdsFromJsonString(String jsonString) {
         Pattern pattern = Pattern.compile("\"jobIds\"\\s*:\\s*\\[(.*?)\\]");
         Matcher matcher = pattern.matcher(jsonString);
diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtAblestackCommvaultBackupHelper.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtAblestackCommvaultBackupHelper.java
index 38ff1ca79310..d994cb1e48e1 100644
--- a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtAblestackCommvaultBackupHelper.java
+++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtAblestackCommvaultBackupHelper.java
@@ -54,6 +54,7 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
+import java.util.concurrent.TimeUnit;
 import javax.xml.xpath.XPath;
 import javax.xml.xpath.XPathConstants;
 import javax.xml.xpath.XPathExpressionException;
@@ -64,6 +65,8 @@ class LibvirtAblestackCommvaultBackupHelper {
     static final Integer EXIT_CLEANUP_FAILED = 20;
     private static final int BACKUP_JOB_POLL_INTERVAL_MS = 10000;
     private static final DateTimeFormatter SCRIPT_LOG_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH-mm-ss>");
+    private static final String STAGING_IN_PROGRESS_MARKER = ".staging.inprogress";
+    private static final String STAGING_COMPLETE_MARKER = ".staging.complete";
 
     enum BackupExecutionMode {
         RUNNING("backup-running"),
@@ -101,8 +104,18 @@ Pair<Integer, String> executeBackup(AblestackCommvaultTakeBackupCommand command)
         String[] scriptCommand = buildBackupScriptCommand(command, diskPaths, executionMode);
         LOGGER.debug("Executing Commvault backup script command=[{}]", String.join(" ", scriptCommand));
         commands.add(scriptCommand);
-        final int timeout = command.getWait() > 0 ? command.getWait() * 1000 : resource.getCmdsTimeout();
-        return Script.executePipedCommands(commands, timeout);
+        final int commandWaitSeconds = command.getWait();
+        final long resourceTimeoutMillis = resource.getCmdsTimeout();
+        final long effectiveTimeoutMillis = commandWaitSeconds > 0 ? TimeUnit.SECONDS.toMillis(commandWaitSeconds) : resourceTimeoutMillis;
+        LOGGER.info(
+                "Executing running VM Commvault backup for vm=[{}], commandWaitSeconds=[{}], "
+                        + "resourceCmdsTimeoutMillis=[{}], effectiveTimeoutMillis=[{}]",
+                command.getVmName(),
+                commandWaitSeconds,
+                resourceTimeoutMillis,
+                effectiveTimeoutMillis
+        );
+        return Script.executePipedCommands(commands, effectiveTimeoutMillis);
     }
 
     List<String> resolveDiskPaths(List<PrimaryDataStoreTO> volumePools, List<String> volumePaths) {
@@ -172,6 +185,7 @@ private Pair<Integer, String> executeStoppedVmBackup(AblestackCommvaultTakeBacku
                 resource.validateLibvirtAndQemuVersionForIncrementalSnapshots();
             }
             Files.createDirectories(dest.resolve("checkpoints"));
+            markStagingInProgress(dest, command);
 
             conn = LibvirtConnection.getConnection();
             String dummyVmXml = buildDummyVmXml(dummyVmName, diskPaths);
@@ -199,7 +213,8 @@ private Pair<Integer, String> executeStoppedVmBackup(AblestackCommvaultTakeBacku
             }
 
             try {
-                waitForBackup(dummyVmName);
+                final long effectiveTimeoutMillis = command.getWait() > 0 ? TimeUnit.SECONDS.toMillis(command.getWait()) : resource.getCmdsTimeout();
+                waitForBackup(dummyVmName, effectiveTimeoutMillis);
             } catch (IOException e) {
                 cancelBackupJob(dummyVmName);
                 throw e;
@@ -209,6 +224,7 @@ private Pair<Integer, String> executeStoppedVmBackup(AblestackCommvaultTakeBacku
             Files.deleteIfExists(backupXml);
             Files.deleteIfExists(checkpointXml);
             Script.runSimpleBashScriptForExitValue("sync", resource.getCmdsTimeout(), false);
+            markStagingComplete(dest, command);
             LOGGER.info("Completed stopped VM Commvault backup for vm=[{}], dummyVm=[{}]", command.getVmName(), dummyVmName);
             return new Pair<>(0, "success");
         } catch (Exception e) {
@@ -223,6 +239,23 @@ private Pair<Integer, String> executeStoppedVmBackup(AblestackCommvaultTakeBacku
         }
     }
 
+    private void markStagingInProgress(Path dest, AblestackCommvaultTakeBackupCommand command) throws IOException {
+        Files.deleteIfExists(dest.resolve(STAGING_COMPLETE_MARKER));
+        Files.writeString(dest.resolve(STAGING_IN_PROGRESS_MARKER),
+                String.format("vm=%s%ncheckpoint=%s%n", command.getVmName(), command.getCheckpointName()),
+                StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
+    }
+
+    private void markStagingComplete(Path dest, AblestackCommvaultTakeBackupCommand command) throws IOException {
+        Path completeMarker = dest.resolve(STAGING_COMPLETE_MARKER);
+        Path tmpMarker = dest.resolve(STAGING_COMPLETE_MARKER + ".tmp");
+        Files.writeString(tmpMarker,
+                String.format("vm=%s%ncheckpoint=%s%n", command.getVmName(), command.getCheckpointName()),
+                StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
+        Files.move(tmpMarker, completeMarker, java.nio.file.StandardCopyOption.REPLACE_EXISTING);
+        Files.deleteIfExists(dest.resolve(STAGING_IN_PROGRESS_MARKER));
+    }
+
     private boolean cleanupStoppedBackupPath(Path dest) {
         if (dest == null || !Files.exists(dest)) {
             return true;
@@ -442,9 +475,9 @@ private Path writeCheckpointXml(Path dest, AblestackCommvaultTakeBackupCommand c
         return checkpointXml;
     }
 
-    private void waitForBackup(String vmName) throws IOException {
-        int timeout = resource.getCmdsTimeout();
-        while (timeout > 0) {
+    private void waitForBackup(String vmName, long timeoutMillis) throws IOException {
+        long remainingMillis = timeoutMillis;
+        while (remainingMillis > 0) {
             String result = checkBackupJob(vmName);
             if (result != null && result.contains("Completed") && result.contains("Backup")) {
                 return;
@@ -452,15 +485,17 @@ private void waitForBackup(String vmName) throws IOException {
             if (result != null && result.contains("Failed")) {
                 throw new IOException("Virsh backup job failed for dummy VM " + vmName);
             }
-            timeout -= BACKUP_JOB_POLL_INTERVAL_MS;
+            long sleepMillis = Math.min(BACKUP_JOB_POLL_INTERVAL_MS, remainingMillis);
             try {
-                Thread.sleep(BACKUP_JOB_POLL_INTERVAL_MS);
+                Thread.sleep(sleepMillis);
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
                 throw new IOException(e);
             }
+            remainingMillis -= sleepMillis;
         }
-        throw new IOException("Timed out waiting for backup job of dummy VM " + vmName);
+        throw new IOException("Timed out waiting for backup job of dummy VM " + vmName + " after "+ timeoutMillis + " milliseconds"
+        );
     }
 
     private void cancelBackupJob(String vmName) {
diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtAblestackCommvaultRestoreBackupCommandWrapper.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtAblestackCommvaultRestoreBackupCommandWrapper.java
index b8a0013a8709..f083e89b03b9 100644
--- a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtAblestackCommvaultRestoreBackupCommandWrapper.java
+++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtAblestackCommvaultRestoreBackupCommandWrapper.java
@@ -93,7 +93,7 @@ public Answer execute(AblestackCommvaultRestoreBackupCommand command, LibvirtCom
         try {
             validateChainStatePlan(volumeChainStates, restorePlan);
             if (AblestackBackupFrameworkUtils.hasRestoreStage(restorePlan, BackupRestoreStage.PREPARE_SOURCE) && hostName != null) {
-                fetchBackupFile(hostName, backupPath);
+                fetchBackupFile(hostName, backupPath, timeout);
             }
             if (AblestackBackupFrameworkUtils.hasRestoreStage(restorePlan, BackupRestoreStage.PREPARE_SOURCE) && backupSourceHosts != null && !backupSourceHosts.isEmpty()) {
                 LinkedHashSet<String> sourceHosts = new LinkedHashSet<>(backupSourceHosts);
@@ -101,7 +101,7 @@ public Answer execute(AblestackCommvaultRestoreBackupCommand command, LibvirtCom
                     if (StringUtils.isBlank(sourceHost) || Objects.equals(sourceHost, hostName)) {
                         continue;
                     }
-                    fetchBackupFile(sourceHost, backupPath);
+                    fetchBackupFile(sourceHost, backupPath, timeout);
                 }
             }
             if (Objects.isNull(vmExists)) {
@@ -900,7 +900,8 @@ private String getXmlForRbdDisk(KVMStoragePoolManager storagePoolMgr, PrimaryDat
         return diskBuilder.toString();
     }
 
-    private void fetchBackupFile(String hostName, String backupPath) {
+    private void fetchBackupFile(String hostName, String backupPath, int timeout) {
+        int timeoutMillis = timeout * 1000;
         int mkdirExit = Script.runSimpleBashScriptForExitValue(String.format(MKDIR_P, backupPath));
         if (mkdirExit != 0) {
             throw new CloudRuntimeException(String.format("Failed to create local backup directory: %s", backupPath));
@@ -909,7 +910,7 @@ private void fetchBackupFile(String hostName, String backupPath) {
         String cmd = String.format(RSYNC_DIR_FROM_REMOTE, hostName, backupPath, backupPath);
         logger.debug("Fetching commvault backup directory from remote host. cmd={}", cmd);
 
-        int exit = Script.runSimpleBashScriptForExitValue(cmd);
+        int exit = Script.runSimpleBashScriptForExitValue(cmd, timeoutMillis, false);
         if (exit != 0) {
             throw new CloudRuntimeException(String.format(
                     "Failed to fetch backup directory from remote host [%s]. remotePath=[%s], localPath=[%s]",
diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtAblestackNasBackupHelper.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtAblestackNasBackupHelper.java
index c04fb30f63fa..5bfa880feb05 100644
--- a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtAblestackNasBackupHelper.java
+++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtAblestackNasBackupHelper.java
@@ -54,6 +54,7 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
+import java.util.concurrent.TimeUnit;
 import javax.xml.xpath.XPath;
 import javax.xml.xpath.XPathConstants;
 import javax.xml.xpath.XPathExpressionException;
@@ -65,6 +66,8 @@ class LibvirtAblestackNasBackupHelper {
     private static final int BACKUP_JOB_POLL_INTERVAL_MS = 10000;
     private static final int UNMOUNT_TIMEOUT_SECONDS = 60;
     private static final DateTimeFormatter SCRIPT_LOG_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH-mm-ss>");
+    private static final String IN_PROGRESS_MARKER = ".backup.inprogress";
+    private static final String COMPLETE_MARKER = ".backup.complete";
 
     enum BackupExecutionMode {
         RUNNING("backup-running"),
@@ -102,7 +105,18 @@ Pair<Integer, String> executeBackup(AblestackNasTakeBackupCommand command) {
         String[] scriptCommand = buildBackupScriptCommand(command, diskPaths, executionMode);
         LOGGER.debug("Executing NAS backup script command=[{}]", String.join(" ", scriptCommand));
         commands.add(scriptCommand);
-        return Script.executePipedCommands(commands, resource.getCmdsTimeout());
+        final int commandWaitSeconds = command.getWait();
+        final long resourceTimeoutMillis = resource.getCmdsTimeout();
+        final long effectiveTimeoutMillis = commandWaitSeconds > 0 ? TimeUnit.SECONDS.toMillis(commandWaitSeconds) : resourceTimeoutMillis;
+        LOGGER.info(
+                "Executing running VM NAS backup for vm=[{}], commandWaitSeconds=[{}], "
+                        + "resourceCmdsTimeoutMillis=[{}], effectiveTimeoutMillis=[{}]",
+                command.getVmName(),
+                commandWaitSeconds,
+                resourceTimeoutMillis,
+                effectiveTimeoutMillis
+        );
+        return Script.executePipedCommands(commands, effectiveTimeoutMillis);
     }
 
     List<String> resolveDiskPaths(List<PrimaryDataStoreTO> volumePools, List<String> volumePaths) {
@@ -213,6 +227,7 @@ private Pair<Integer, String> executeStoppedVmBackup(AblestackNasTakeBackupComma
             mountPoint = mountRepository(command);
             dest = mountPoint.resolve(command.getBackupPath());
             Files.createDirectories(dest.resolve("checkpoints"));
+            markBackupInProgress(dest, command);
 
             conn = LibvirtConnection.getConnection();
             String dummyVmXml = buildDummyVmXml(dummyVmName, diskPaths, conn);
@@ -239,7 +254,8 @@ private Pair<Integer, String> executeStoppedVmBackup(AblestackNasTakeBackupComma
             }
 
             try {
-                waitForBackup(dummyVmName);
+                final long effectiveTimeoutMillis = command.getWait() > 0 ? TimeUnit.SECONDS.toMillis(command.getWait()) : resource.getCmdsTimeout();
+                waitForBackup(dummyVmName, effectiveTimeoutMillis);
             } catch (IOException e) {
                 cancelBackupJob(dummyVmName);
                 throw e;
@@ -254,6 +270,7 @@ private Pair<Integer, String> executeStoppedVmBackup(AblestackNasTakeBackupComma
             Files.deleteIfExists(backupXml);
             Files.deleteIfExists(checkpointXml);
             runCommand(String.format("sync"));
+            markBackupComplete(dest, command);
             String output = listTopLevelFileSizes(dest);
             LOGGER.info("Completed stopped VM NAS backup for vm=[{}], dummyVm=[{}]", command.getVmName(), dummyVmName);
             return new Pair<>(0, output);
@@ -303,6 +320,23 @@ private boolean cleanupStoppedBackup(AblestackNasTakeBackupCommand command, Path
         return unmountRepository(command, mountPoint) && success;
     }
 
+    private void markBackupInProgress(Path dest, AblestackNasTakeBackupCommand command) throws IOException {
+        Files.deleteIfExists(dest.resolve(COMPLETE_MARKER));
+        Files.writeString(dest.resolve(IN_PROGRESS_MARKER),
+                String.format("vm=%s%ncheckpoint=%s%n", command.getVmName(), command.getCheckpointName()),
+                StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
+    }
+
+    private void markBackupComplete(Path dest, AblestackNasTakeBackupCommand command) throws IOException {
+        Path completeMarker = dest.resolve(COMPLETE_MARKER);
+        Path tmpMarker = dest.resolve(COMPLETE_MARKER + ".tmp");
+        Files.writeString(tmpMarker,
+                String.format("vm=%s%ncheckpoint=%s%n", command.getVmName(), command.getCheckpointName()),
+                StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
+        Files.move(tmpMarker, completeMarker, java.nio.file.StandardCopyOption.REPLACE_EXISTING);
+        Files.deleteIfExists(dest.resolve(IN_PROGRESS_MARKER));
+    }
+
     private boolean unmountRepository(AblestackNasTakeBackupCommand command, Path mountPoint) {
         if (mountPoint == null) {
             return true;
@@ -499,9 +533,9 @@ private Path writeCheckpointXml(Path dest, AblestackNasTakeBackupCommand command
         return checkpointXml;
     }
 
-    private void waitForBackup(String vmName) throws IOException {
-        int timeout = resource.getCmdsTimeout();
-        while (timeout > 0) {
+    private void waitForBackup(String vmName, long timeoutMillis) throws IOException {
+        long remainingMillis = timeoutMillis;
+        while (remainingMillis > 0) {
             String result = checkBackupJob(vmName);
             if (result != null && result.contains("Completed") && result.contains("Backup")) {
                 return;
@@ -509,15 +543,17 @@ private void waitForBackup(String vmName) throws IOException {
             if (result != null && result.contains("Failed")) {
                 throw new IOException("Virsh backup job failed for dummy VM " + vmName);
             }
-            timeout -= BACKUP_JOB_POLL_INTERVAL_MS;
+            long sleepMillis = Math.min(BACKUP_JOB_POLL_INTERVAL_MS, remainingMillis);
             try {
-                Thread.sleep(BACKUP_JOB_POLL_INTERVAL_MS);
+                Thread.sleep(sleepMillis);
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
                 throw new IOException(e);
             }
+            remainingMillis -= sleepMillis;
         }
-        throw new IOException("Timed out waiting for backup job of dummy VM " + vmName);
+        throw new IOException("Timed out waiting for backup job of dummy VM " + vmName + " after " + timeoutMillis + " milliseconds"
+        );
     }
 
     private void cancelBackupJob(String vmName) {
diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtAblestackNetBackupHelper.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtAblestackNetBackupHelper.java
index fea3792f818d..65e6803d0b95 100644
--- a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtAblestackNetBackupHelper.java
+++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtAblestackNetBackupHelper.java
@@ -54,6 +54,7 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
+import java.util.concurrent.TimeUnit;
 import javax.xml.xpath.XPath;
 import javax.xml.xpath.XPathConstants;
 import javax.xml.xpath.XPathExpressionException;
@@ -64,6 +65,8 @@ class LibvirtAblestackNetBackupHelper {
     static final Integer EXIT_CLEANUP_FAILED = 20;
     private static final int BACKUP_JOB_POLL_INTERVAL_MS = 10000;
     private static final DateTimeFormatter SCRIPT_LOG_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH-mm-ss>");
+    private static final String STAGING_IN_PROGRESS_MARKER = ".staging.inprogress";
+    private static final String STAGING_COMPLETE_MARKER = ".staging.complete";
 
     enum BackupExecutionMode {
         RUNNING("backup-running"),
@@ -103,8 +106,18 @@ Pair<Integer, String> executeBackup(AblestackNetBackupTakeBackupCommand command)
             String[] scriptCommand = buildBackupScriptCommand(command, diskPaths, executionMode);
             LOGGER.debug("Executing NetBackup script command=[{}]", String.join(" ", scriptCommand));
             commands.add(scriptCommand);
-            final int timeout = command.getWait() > 0 ? command.getWait() * 1000 : resource.getCmdsTimeout();
-            return Script.executePipedCommands(commands, timeout);
+            final int commandWaitSeconds = command.getWait();
+            final long resourceTimeoutMillis = resource.getCmdsTimeout();
+            final long effectiveTimeoutMillis = commandWaitSeconds > 0 ? TimeUnit.SECONDS.toMillis(commandWaitSeconds) : resourceTimeoutMillis;
+            LOGGER.info(
+                    "Executing running VM NetBackup backup for vm=[{}], commandWaitSeconds=[{}], "
+                            + "resourceCmdsTimeoutMillis=[{}], effectiveTimeoutMillis=[{}]",
+                    command.getVmName(),
+                    commandWaitSeconds,
+                    resourceTimeoutMillis,
+                    effectiveTimeoutMillis
+            );
+            return Script.executePipedCommands(commands, effectiveTimeoutMillis);
         } finally {
             cleanupParentCheckpointWorkspace(parentCheckpointWorkspace);
         }
@@ -215,6 +228,7 @@ private Pair<Integer, String> executeStoppedVmBackup(AblestackNetBackupTakeBacku
                 resource.validateLibvirtAndQemuVersionForIncrementalSnapshots();
             }
             Files.createDirectories(dest.resolve("checkpoints"));
+            markStagingInProgress(dest, command);
 
             conn = LibvirtConnection.getConnection();
             String dummyVmXml = buildDummyVmXml(dummyVmName, diskPaths);
@@ -242,7 +256,8 @@ private Pair<Integer, String> executeStoppedVmBackup(AblestackNetBackupTakeBacku
             }
 
             try {
-                waitForBackup(dummyVmName);
+                final long effectiveTimeoutMillis = command.getWait() > 0 ? TimeUnit.SECONDS.toMillis(command.getWait()) : resource.getCmdsTimeout();
+                waitForBackup(dummyVmName, effectiveTimeoutMillis);
             } catch (IOException e) {
                 cancelBackupJob(dummyVmName);
                 throw e;
@@ -252,6 +267,7 @@ private Pair<Integer, String> executeStoppedVmBackup(AblestackNetBackupTakeBacku
             Files.deleteIfExists(backupXml);
             Files.deleteIfExists(checkpointXml);
             Script.runSimpleBashScriptForExitValue("sync", resource.getCmdsTimeout(), false);
+            markStagingComplete(dest, command);
             LOGGER.info("Completed stopped VM NetBackup backup for vm=[{}], dummyVm=[{}]", command.getVmName(), dummyVmName);
             return new Pair<>(0, "success");
         } catch (Exception e) {
@@ -264,6 +280,23 @@ private Pair<Integer, String> executeStoppedVmBackup(AblestackNetBackupTakeBacku
         }
     }
 
+    private void markStagingInProgress(Path dest, AblestackNetBackupTakeBackupCommand command) throws IOException {
+        Files.deleteIfExists(dest.resolve(STAGING_COMPLETE_MARKER));
+        Files.writeString(dest.resolve(STAGING_IN_PROGRESS_MARKER),
+                String.format("vm=%s%ncheckpoint=%s%n", command.getVmName(), command.getCheckpointName()),
+                StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
+    }
+
+    private void markStagingComplete(Path dest, AblestackNetBackupTakeBackupCommand command) throws IOException {
+        Path completeMarker = dest.resolve(STAGING_COMPLETE_MARKER);
+        Path tmpMarker = dest.resolve(STAGING_COMPLETE_MARKER + ".tmp");
+        Files.writeString(tmpMarker,
+                String.format("vm=%s%ncheckpoint=%s%n", command.getVmName(), command.getCheckpointName()),
+                StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
+        Files.move(tmpMarker, completeMarker, java.nio.file.StandardCopyOption.REPLACE_EXISTING);
+        Files.deleteIfExists(dest.resolve(STAGING_IN_PROGRESS_MARKER));
+    }
+
     private Path ensureParentCheckpointMaterialized(AblestackNetBackupTakeBackupCommand command) {
         if (!isIncremental(command)) {
             return null;
@@ -488,9 +521,9 @@ private Path writeCheckpointXml(Path dest, AblestackNetBackupTakeBackupCommand c
         return checkpointXml;
     }
 
-    private void waitForBackup(String vmName) throws IOException {
-        int timeout = resource.getCmdsTimeout();
-        while (timeout > 0) {
+    private void waitForBackup(String vmName, long timeoutMillis) throws IOException {
+        long remainingMillis = timeoutMillis;
+        while (remainingMillis > 0) {
             String result = checkBackupJob(vmName);
             if (result != null && result.contains("Completed") && result.contains("Backup")) {
                 return;
@@ -498,15 +531,19 @@ private void waitForBackup(String vmName) throws IOException {
             if (result != null && result.contains("Failed")) {
                 throw new IOException("Virsh backup job failed for dummy VM " + vmName);
             }
-            timeout -= BACKUP_JOB_POLL_INTERVAL_MS;
+            long sleepMillis = Math.min(
+                    BACKUP_JOB_POLL_INTERVAL_MS,
+                    remainingMillis
+            );
             try {
-                Thread.sleep(BACKUP_JOB_POLL_INTERVAL_MS);
+                Thread.sleep(sleepMillis);
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
                 throw new IOException(e);
             }
+            remainingMillis -= sleepMillis;
         }
-        throw new IOException("Timed out waiting for backup job of dummy VM " + vmName);
+        throw new IOException("Timed out waiting for backup job of dummy VM " + vmName + " after " + timeoutMillis + " milliseconds");
     }
 
     private void cancelBackupJob(String vmName) {
diff --git a/scripts/vm/hypervisor/kvm/ablestack_cvtbackup.sh b/scripts/vm/hypervisor/kvm/ablestack_cvtbackup.sh
index 50255cf5fb91..dd47e6db3af1 100644
--- a/scripts/vm/hypervisor/kvm/ablestack_cvtbackup.sh
+++ b/scripts/vm/hypervisor/kvm/ablestack_cvtbackup.sh
@@ -42,6 +42,8 @@ logFile="/var/log/cloudstack/agent/agent.log"
 CREATED_RBD_SNAPSHOTS=()
 
 EXIT_CLEANUP_FAILED=20
+STAGING_IN_PROGRESS_MARKER=".staging.inprogress"
+STAGING_COMPLETE_MARKER=".staging.complete"
 
 log() {
   [[ "$verb" -eq 1 ]] && builtin echo "$@"
@@ -391,6 +393,7 @@ EOF
 
 backup_running_vm() {
   mkdir -p "$dest/checkpoints" || { echo "Failed to create backup directory $dest"; exit 1; }
+  mark_staging_in_progress
   local parent_checkpoint_file=""
   if [[ "$BACKUP_TYPE" == "INCREMENTAL" && -n "$PARENT_CHECKPOINT_PATH" ]]; then
     parent_checkpoint_file="$PARENT_CHECKPOINT_PATH"
@@ -459,7 +462,7 @@ backup_running_vm() {
       Completed) break ;;
       Failed)
         log -ne "FAILED libvirt backup job vm=[$VM] checkpoint=[$CHECKPOINT_NAME]"
-        echo "Virsh backup job failed"; cleanup ;;
+        echo "Virsh backup job failed"; cleanup; exit 1 ;;
     esac
     wait_count=$((wait_count + 1))
     if (( wait_count % 12 == 0 )); then
@@ -472,10 +475,12 @@ backup_running_vm() {
   dump_checkpoint_xml "$VM"
   rm -f "$dest/backup.xml" "$dest/checkpoint.xml"
   sync
+  mark_staging_complete
 }
 
 backup_rbd_volumes() {
   mkdir -p "$dest/checkpoints" || { echo "Failed to create backup directory $dest"; exit 1; }
+  mark_staging_in_progress
   backup_domain_information "$VM"
   trap 'log -ne "FAILED RBD backup unexpected error line=[$LINENO] op=[$OP] vm=[$VM] checkpoint=[$CHECKPOINT_NAME]"; cleanup_created_rbd_snapshots' ERR
   trap 'log -ne "FAILED RBD backup interrupted op=[$OP] vm=[$VM] checkpoint=[$CHECKPOINT_NAME]"; cleanup_created_rbd_snapshots; exit 1' INT TERM
@@ -544,6 +549,8 @@ backup_rbd_volumes() {
   trap - ERR
   trap - INT TERM
   CREATED_RBD_SNAPSHOTS=()
+  sync
+  mark_staging_complete
 }
 
 has_child_backup() {
@@ -670,6 +677,20 @@ cleanup_unreferenced_qcow2_bitmaps() {
   done < <(split_csv "$CLEANUP_CHECKPOINT_NAMES")
 }
 
+mark_staging_in_progress() {
+  rm -f "$dest/$STAGING_COMPLETE_MARKER"
+  printf 'started_at=%s\nvm=%s\ncheckpoint=%s\n' "$(date '+%Y-%m-%d %H:%M:%S')" "$VM" "$CHECKPOINT_NAME" > "$dest/$STAGING_IN_PROGRESS_MARKER"
+  sync "$dest/$STAGING_IN_PROGRESS_MARKER" 2>/dev/null || true
+}
+
+mark_staging_complete() {
+  local tmp_marker="$dest/$STAGING_COMPLETE_MARKER.tmp"
+  printf 'completed_at=%s\nvm=%s\ncheckpoint=%s\n' "$(date '+%Y-%m-%d %H:%M:%S')" "$VM" "$CHECKPOINT_NAME" > "$tmp_marker"
+  mv -f "$tmp_marker" "$dest/$STAGING_COMPLETE_MARKER"
+  rm -f "$dest/$STAGING_IN_PROGRESS_MARKER"
+  sync "$dest/$STAGING_COMPLETE_MARKER" 2>/dev/null || true
+}
+
 delete_backup() {
   if [[ -f "$dest/rbd-backup.meta" ]]; then
     source "$dest/rbd-backup.meta"
diff --git a/scripts/vm/hypervisor/kvm/ablestack_nasbackup.sh b/scripts/vm/hypervisor/kvm/ablestack_nasbackup.sh
index 0e923a3058c6..447ed74c6f2b 100755
--- a/scripts/vm/hypervisor/kvm/ablestack_nasbackup.sh
+++ b/scripts/vm/hypervisor/kvm/ablestack_nasbackup.sh
@@ -46,6 +46,8 @@ UNMOUNT_TIMEOUT=60
 CREATED_RBD_SNAPSHOTS=()
 
 EXIT_CLEANUP_FAILED=20
+IN_PROGRESS_MARKER=".backup.inprogress"
+COMPLETE_MARKER=".backup.complete"
 
 log() {
   [[ "$verb" -eq 1 ]] && builtin echo "$@"
@@ -107,6 +109,7 @@ backup_running_vm() {
   mount_operation
   mkdir -p "$dest" || { echo "Failed to create backup directory $dest"; exit 1; }
   mkdir -p "$dest/checkpoints" || { echo "Failed to create checkpoint directory $dest/checkpoints"; exit 1; }
+  mark_backup_in_progress
 
   local parent_checkpoint_file=""
   if [[ "$BACKUP_TYPE" == "INCREMENTAL" && -n "$PARENT_CHECKPOINT_PATH" ]]; then
@@ -176,7 +179,8 @@ backup_running_vm() {
       Failed)
         log -ne "FAILED libvirt backup job vm=[$VM] checkpoint=[$CHECKPOINT_NAME]"
         echo "Virsh backup job failed"
-        cleanup ;;
+        cleanup
+        exit 1 ;;
     esac
     wait_count=$((wait_count + 1))
     if (( wait_count % 12 == 0 )); then
@@ -207,6 +211,7 @@ backup_running_vm() {
   rm -f "$dest/backup.xml"
   rm -f "$dest/checkpoint.xml"
   sync
+  mark_backup_complete
 
   # Print statistics
   virsh -c qemu:///system domjobinfo "$VM" --completed
@@ -220,6 +225,7 @@ backup_rbd_volumes() {
   log -ne "Entered backup_rbd_volumes with DISK_PATHS=[$DISK_PATHS], BACKUP_FILES=[$BACKUP_FILES], BACKUP_DIR=[$BACKUP_DIR]"
   mount_operation
   mkdir -p "$dest" || { echo "Failed to create backup directory $dest"; exit 1; }
+  mark_backup_in_progress
 
   backup_domain_information "$VM"
   trap 'log -ne "FAILED RBD backup unexpected error line=[$LINENO] op=[$OP] vm=[$VM] checkpoint=[$CHECKPOINT_NAME]"; cleanup_created_rbd_snapshots' ERR
@@ -305,6 +311,7 @@ backup_rbd_volumes() {
   CREATED_RBD_SNAPSHOTS=()
 
   sync
+  mark_backup_complete
   log -ne "RBD backup completed checkpoint=[$CHECKPOINT_NAME] parent=[$PARENT_CHECKPOINT_NAME]"
   timeout "$UNMOUNT_TIMEOUT" umount "$mount_point" 2>>"$logFile" || { log "WARNING: umount of $mount_point failed or timed out"; true; }
   rmdir "$mount_point" 2>>"$logFile" || { log "WARNING: rmdir of $mount_point failed"; true; }
@@ -514,6 +521,49 @@ get_backup_stats() {
   rmdir $mount_point
 }
 
+inspect_backup() {
+  mount_operation
+
+  local required_files_present=true
+  local backup_path_exists=false
+  local complete=false
+  local in_progress=false
+  local missing_files=""
+
+  if [[ -d "$dest" ]]; then
+    backup_path_exists=true
+  fi
+  if [[ -f "$dest/$COMPLETE_MARKER" ]]; then
+    complete=true
+  fi
+  if [[ -f "$dest/$IN_PROGRESS_MARKER" ]]; then
+    in_progress=true
+  fi
+
+  while IFS= read -r backup_file; do
+    [[ -z "$backup_file" ]] && continue
+    if [[ ! -s "$dest/$backup_file" ]]; then
+      required_files_present=false
+      missing_files="${missing_files}${missing_files:+,}${backup_file}"
+    fi
+  done < <(split_csv "$BACKUP_FILES")
+
+  if [[ -n "$CHECKPOINT_NAME" && ! -f "$dest/checkpoints/$CHECKPOINT_NAME.xml" && ! -f "$dest/checkpoints/$CHECKPOINT_NAME.meta" ]]; then
+    required_files_present=false
+    missing_files="${missing_files}${missing_files:+,}checkpoints/$CHECKPOINT_NAME"
+  fi
+
+  echo "backupPathExists=$backup_path_exists"
+  echo "complete=$complete"
+  echo "inProgress=$in_progress"
+  echo "requiredFilesPresent=$required_files_present"
+  echo "size=$(du -sb "$dest" 2>/dev/null | cut -f1 || echo 0)"
+  echo "missingFiles=$missing_files"
+
+  umount "$mount_point"
+  rmdir "$mount_point"
+}
+
 mount_operation() {
   mount_point=$(mktemp -d -t csbackup.XXXXX)
   dest="$mount_point/${BACKUP_DIR}"
@@ -530,6 +580,20 @@ mount_operation() {
   fi
 }
 
+mark_backup_in_progress() {
+  rm -f "$dest/$COMPLETE_MARKER"
+  printf 'started_at=%s\nvm=%s\ncheckpoint=%s\n' "$(date '+%Y-%m-%d %H:%M:%S')" "$VM" "$CHECKPOINT_NAME" > "$dest/$IN_PROGRESS_MARKER"
+  sync "$dest/$IN_PROGRESS_MARKER" 2>/dev/null || true
+}
+
+mark_backup_complete() {
+  local tmp_marker="$dest/$COMPLETE_MARKER.tmp"
+  printf 'completed_at=%s\nvm=%s\ncheckpoint=%s\n' "$(date '+%Y-%m-%d %H:%M:%S')" "$VM" "$CHECKPOINT_NAME" > "$tmp_marker"
+  mv -f "$tmp_marker" "$dest/$COMPLETE_MARKER"
+  rm -f "$dest/$IN_PROGRESS_MARKER"
+  sync "$dest/$COMPLETE_MARKER" 2>/dev/null || true
+}
+
 cleanup() {
   local status=0
 
@@ -899,4 +963,6 @@ elif [ "$OP" = "delete" ]; then
   delete_backup
 elif [ "$OP" = "stats" ]; then
   get_backup_stats
+elif [ "$OP" = "inspect" ]; then
+  inspect_backup
 fi
diff --git a/scripts/vm/hypervisor/kvm/ablestack_netbackup.sh b/scripts/vm/hypervisor/kvm/ablestack_netbackup.sh
index cc2a0622be02..8e165a41c1b6 100644
--- a/scripts/vm/hypervisor/kvm/ablestack_netbackup.sh
+++ b/scripts/vm/hypervisor/kvm/ablestack_netbackup.sh
@@ -42,6 +42,8 @@ logFile="/var/log/cloudstack/agent/agent.log"
 CREATED_RBD_SNAPSHOTS=()
 
 EXIT_CLEANUP_FAILED=20
+STAGING_IN_PROGRESS_MARKER=".staging.inprogress"
+STAGING_COMPLETE_MARKER=".staging.complete"
 
 log() {
   [[ "$verb" -eq 1 ]] && builtin echo "$@"
@@ -450,6 +452,7 @@ EOF
 
 backup_running_vm() {
   mkdir -p "$dest/checkpoints" || { echo "Failed to create backup directory $dest"; exit 1; }
+  mark_staging_in_progress
   local parent_checkpoint_file=""
   if [[ "$BACKUP_TYPE" == "INCREMENTAL" && -n "$PARENT_CHECKPOINT_PATH" ]]; then
     parent_checkpoint_file="$PARENT_CHECKPOINT_PATH"
@@ -518,7 +521,7 @@ backup_running_vm() {
       Completed) break ;;
       Failed)
         log -ne "FAILED libvirt backup job vm=[$VM] checkpoint=[$CHECKPOINT_NAME]"
-        echo "Virsh backup job failed"; cleanup ;;
+        echo "Virsh backup job failed"; cleanup; exit 1 ;;
     esac
     wait_count=$((wait_count + 1))
     if (( wait_count % 12 == 0 )); then
@@ -531,10 +534,12 @@ backup_running_vm() {
   dump_checkpoint_xml "$VM"
   rm -f "$dest/backup.xml" "$dest/checkpoint.xml"
   sync
+  mark_staging_complete
 }
 
 backup_rbd_volumes() {
   mkdir -p "$dest/checkpoints" || { echo "Failed to create backup directory $dest"; exit 1; }
+  mark_staging_in_progress
   backup_domain_information "$VM"
   trap 'log -ne "FAILED RBD backup unexpected error line=[$LINENO] op=[$OP] vm=[$VM] checkpoint=[$CHECKPOINT_NAME]"; cleanup_created_rbd_snapshots' ERR
   trap 'log -ne "FAILED RBD backup interrupted op=[$OP] vm=[$VM] checkpoint=[$CHECKPOINT_NAME]"; cleanup_created_rbd_snapshots; exit 1' INT TERM
@@ -603,6 +608,8 @@ backup_rbd_volumes() {
   trap - ERR
   trap - INT TERM
   CREATED_RBD_SNAPSHOTS=()
+  sync
+  mark_staging_complete
 }
 
 has_child_backup() {
@@ -729,6 +736,20 @@ cleanup_unreferenced_qcow2_bitmaps() {
   done < <(split_csv "$CLEANUP_CHECKPOINT_NAMES")
 }
 
+mark_staging_in_progress() {
+  rm -f "$dest/$STAGING_COMPLETE_MARKER"
+  printf 'started_at=%s\nvm=%s\ncheckpoint=%s\n' "$(date '+%Y-%m-%d %H:%M:%S')" "$VM" "$CHECKPOINT_NAME" > "$dest/$STAGING_IN_PROGRESS_MARKER"
+  sync "$dest/$STAGING_IN_PROGRESS_MARKER" 2>/dev/null || true
+}
+
+mark_staging_complete() {
+  local tmp_marker="$dest/$STAGING_COMPLETE_MARKER.tmp"
+  printf 'completed_at=%s\nvm=%s\ncheckpoint=%s\n' "$(date '+%Y-%m-%d %H:%M:%S')" "$VM" "$CHECKPOINT_NAME" > "$tmp_marker"
+  mv -f "$tmp_marker" "$dest/$STAGING_COMPLETE_MARKER"
+  rm -f "$dest/$STAGING_IN_PROGRESS_MARKER"
+  sync "$dest/$STAGING_COMPLETE_MARKER" 2>/dev/null || true
+}
+
 delete_backup() {
   if [[ -f "$dest/rbd-backup.meta" ]]; then
     source "$dest/rbd-backup.meta"
diff --git a/scripts/vm/hypervisor/kvm/netbackup-host-bpstart-notify.sh b/scripts/vm/hypervisor/kvm/netbackup-host-bpstart-notify.sh
index e86bfb01e476..a0203064e9e9 100755
--- a/scripts/vm/hypervisor/kvm/netbackup-host-bpstart-notify.sh
+++ b/scripts/vm/hypervisor/kvm/netbackup-host-bpstart-notify.sh
@@ -99,9 +99,9 @@ while IFS= read -r vm_name; do
 done < <(list_target_vms)
 
 if [[ ${vm_count} -eq 0 ]]; then
-  update_runtime_status "MOLD_BACKUP_FAILED_ALL"
-  log -ne "No running VMs found on host for NetBackup staging"
-  exit 1
+  update_runtime_status "MOLD_BACKUP_SKIPPED_NO_TARGET"
+  log -ne "No target running VMs found on host for NetBackup staging; skipping successfully"
+  exit 0
 fi
 
 if [[ ${success_count} -gt 0 && ${failed_count} -eq 0 ]]; then
diff --git a/server/src/main/java/com/cloud/api/query/dao/UserVmJoinDaoImpl.java b/server/src/main/java/com/cloud/api/query/dao/UserVmJoinDaoImpl.java
index 1c3baef2ef05..e8fa86574f03 100644
--- a/server/src/main/java/com/cloud/api/query/dao/UserVmJoinDaoImpl.java
+++ b/server/src/main/java/com/cloud/api/query/dao/UserVmJoinDaoImpl.java
@@ -43,6 +43,8 @@
 import org.apache.cloudstack.api.response.SecurityGroupResponse;
 import org.apache.cloudstack.api.response.UserVmResponse;
 import org.apache.cloudstack.api.response.VnfNicResponse;
+import org.apache.cloudstack.backup.Backup;
+import org.apache.cloudstack.backup.dao.BackupDao;
 import org.apache.cloudstack.context.CallContext;
 import org.apache.cloudstack.extension.ExtensionHelper;
 import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
@@ -148,6 +150,8 @@ public class UserVmJoinDaoImpl extends GenericDaoBaseWithTagInformation<UserVmJo
     VMTemplateDao vmTemplateDao;
     @Inject
     ExtensionHelper extensionHelper;
+    @Inject
+    BackupDao backupDao;
 
     private final SearchBuilder<UserVmJoinVO> VmDetailSearch;
     private final SearchBuilder<UserVmJoinVO> activeVmByIsoSearch;
@@ -231,6 +235,7 @@ public UserVmResponse newUserVmResponse(ResponseView view, String objectName, Us
         if (isFastCloneFlattenActive(fastCloneStatus)) {
             setFastCloneFlattenVolume(userVmResponse, userVm.getId());
         }
+        setActiveBackupStatus(userVmResponse, userVm.getId());
 
         User user = _userDao.getUser(userVm.getUserId());
         if (user != null) {
@@ -589,6 +594,20 @@ protected boolean isFastCloneFlattenActive(VMInstanceDetailVO fastCloneStatus) {
                 FAST_CLONE_FLATTEN_PENDING.equalsIgnoreCase(fastCloneStatus.getValue());
     }
 
+    protected void setActiveBackupStatus(UserVmResponse userVmResponse, long vmId) {
+        List<Backup> backups = backupDao.listByVmId(null, vmId);
+        if (CollectionUtils.isEmpty(backups)) {
+            return;
+        }
+
+        for (Backup backup : backups) {
+            if (Backup.Status.BackingUp.equals(backup.getStatus())) {
+                userVmResponse.setActiveBackupStatus(backup.getStatus().name());
+                return;
+            }
+        }
+    }
+
     protected void setFastCloneFlattenVolume(UserVmResponse userVmResponse, long vmId) {
         VolumeVO volume = findFastCloneFlattenVolume(vmId, FAST_CLONE_FLATTEN_RUNNING);
         if (volume == null) {
diff --git a/server/src/main/java/com/cloud/vm/UserVmManagerImpl.java b/server/src/main/java/com/cloud/vm/UserVmManagerImpl.java
index c44acb6f807a..39c7da935063 100644
--- a/server/src/main/java/com/cloud/vm/UserVmManagerImpl.java
+++ b/server/src/main/java/com/cloud/vm/UserVmManagerImpl.java
@@ -115,6 +115,7 @@
 import org.apache.cloudstack.api.command.user.vmgroup.DeleteVMGroupCmd;
 import org.apache.cloudstack.api.command.user.volume.ChangeOfferingForVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.ResizeVolumeCmd;
+import org.apache.cloudstack.backup.Backup;
 import org.apache.cloudstack.backup.BackupManager;
 import org.apache.cloudstack.backup.BackupProvider;
 import org.apache.cloudstack.backup.BackupScheduleVO;
@@ -10881,6 +10882,7 @@ public void validateCloneCondition(CloneVMCmd cmd) throws InvalidParameterValueE
         if (curVm == null) {
             throw new CloudRuntimeException("the VM doesn't exist or not registered in management server!");
         }
+        checkNoActiveBackupForClone(curVm.getId());
         UserVmVO vmStatus = _vmDao.findById(cmd.getId());
         if (vmStatus.getHypervisorType() != HypervisorType.KVM && vmStatus.getHypervisorType() != HypervisorType.Simulator) {
             throw new CloudRuntimeException("The clone operation is only supported on KVM and Simulator!");
@@ -10954,10 +10956,23 @@ public void validateCloneCondition(CloneVMCmd cmd) throws InvalidParameterValueE
         _resourceLimitMgr.checkResourceLimit(activeOwner, ResourceType.primary_storage, totalSize);
     }
 
+    protected void checkNoActiveBackupForClone(long vmId) {
+        List<Backup> backups = backupDao.listByVmId(null, vmId);
+        if (CollectionUtils.isEmpty(backups)) {
+            return;
+        }
+        for (Backup backup : backups) {
+            if (Backup.Status.BackingUp.equals(backup.getStatus())) {
+                throw new CloudRuntimeException(String.format("Unable to clone VM while backup is currently in progress for VM [%s]. Please retry after backup completes.", vmId));
+            }
+        }
+    }
+
     @Override
     @ActionEvent(eventType = EventTypes.EVENT_VM_CLONE, eventDescription = "VM CLONE", async = true)
     public Optional<UserVm> cloneVirtualMachine(CloneVMCmd cmd) throws ResourceAllocationException, ResourceUnavailableException, InsufficientCapacityException {
         UserVmVO curVm = _vmDao.findById(cmd.getId());
+        checkNoActiveBackupForClone(curVm.getId());
         Account curVmAccount = _accountDao.findById(curVm.getAccountId());
         long zoneId = cmd.getTargetVM().getDataCenterId();
         String clone_type = cmd.getType();
diff --git a/server/src/main/java/org/apache/cloudstack/backup/BackupManagerImpl.java b/server/src/main/java/org/apache/cloudstack/backup/BackupManagerImpl.java
index 7ad959ec7027..c156d50e9c92 100644
--- a/server/src/main/java/org/apache/cloudstack/backup/BackupManagerImpl.java
+++ b/server/src/main/java/org/apache/cloudstack/backup/BackupManagerImpl.java
@@ -26,6 +26,7 @@
 import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
@@ -33,6 +34,7 @@
 import java.util.Timer;
 import java.util.TimerTask;
 import java.util.Iterator;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -144,11 +146,13 @@
 import com.cloud.storage.Storage;
 import com.cloud.storage.Volume;
 import com.cloud.storage.VolumeApiService;
+import com.cloud.storage.VolumeDetailVO;
 import com.cloud.storage.VolumeVO;
 import com.cloud.storage.dao.DiskOfferingDao;
 import com.cloud.storage.dao.GuestOSDao;
 import com.cloud.storage.dao.VMTemplateDao;
 import com.cloud.storage.dao.VolumeDao;
+import com.cloud.storage.dao.VolumeDetailsDao;
 import com.cloud.template.VirtualMachineTemplate;
 import com.cloud.user.Account;
 import com.cloud.user.AccountManager;
@@ -218,6 +222,8 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
     @Inject
     private VolumeDao volumeDao;
     @Inject
+    private VolumeDetailsDao volumeDetailsDao;
+    @Inject
     private DataCenterDao dataCenterDao;
     @Inject
     private BackgroundPollManager backgroundPollManager;
@@ -269,6 +275,11 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
     private Date currentTimestamp;
     private static final int POST_RESTORE_MAINTENANCE_MAX_RETRIES = 5;
     private static final long POST_RESTORE_MAINTENANCE_RETRY_INTERVAL_MS = 60_000L;
+    private static final int COMMVAULT_BACKUP_AGENT_INSTALL_RETRY_ATTEMPTS = 10;
+    private static final long COMMVAULT_BACKUP_AGENT_INSTALL_RETRY_INTERVAL_MS = TimeUnit.SECONDS.toMillis(30);
+    private static final String FAST_CLONE_FLATTEN_STATUS = "clone.fast.flatten.status";
+    private static final String FAST_CLONE_FLATTEN_PENDING = "pending";
+    private static final String FAST_CLONE_FLATTEN_RUNNING = "running";
 
     private static Map<String, BackupProvider> backupProvidersMap = new HashMap<>();
     private static final String ABLESTACK_NETBACKUP_PROVIDER_NAME = "ablestack-netbackup";
@@ -1008,6 +1019,7 @@ public boolean createBackup(CreateBackupCmd cmd, Object job) throws ResourceAllo
         boolean isScheduledBackup = backupScheduleId != null;
         logger.info("Starting VM backup request [vmId: {}, vmUuid: {}, vmName: {}, provider: {}, offeringId: {}, scheduleId: {}, scheduled: {}]",
                 vm.getId(), vm.getUuid(), vm.getInstanceName(), offering.getProvider(), offering.getId(), backupScheduleId, isScheduledBackup);
+        checkNoActiveFastCloneFlattenForBackup(vmId);
         Account owner = accountManager.getAccount(vm.getAccountId());
 
         Long backupSize = 0L;
@@ -1081,7 +1093,27 @@ protected Long calculateBackupSize(final Long vmId) {
         return backupSize;
     }
 
-        @Override
+    protected void checkNoActiveFastCloneFlattenForBackup(final Long vmId) {
+        final List<VolumeVO> volumes = volumeDao.findByInstance(vmId);
+        if (CollectionUtils.isEmpty(volumes)) {
+            return;
+        }
+
+        for (VolumeVO volume : volumes) {
+            final VolumeDetailVO fastCloneFlattenStatus = volumeDetailsDao.findDetail(volume.getId(), FAST_CLONE_FLATTEN_STATUS);
+            if (fastCloneFlattenStatus == null || StringUtils.isBlank(fastCloneFlattenStatus.getValue())) {
+                continue;
+            }
+            if (FAST_CLONE_FLATTEN_PENDING.equalsIgnoreCase(fastCloneFlattenStatus.getValue()) ||
+                    FAST_CLONE_FLATTEN_RUNNING.equalsIgnoreCase(fastCloneFlattenStatus.getValue())) {
+                throw new CloudRuntimeException(String.format(
+                        "Unable to create VM backup while SharedMountPoint clone flatten is %s for volume [%s]. Please retry after flatten completes.",
+                        fastCloneFlattenStatus.getValue(), volume.getUuid()));
+            }
+        }
+    }
+
+    @Override
     @ActionEvent(eventType = EventTypes.EVENT_VM_BACKUP_CREATE, eventDescription = "creating VM backup for NetBackup", async = true)
     public boolean createNetBackup(final CreateNetBackupCmd cmd) throws ResourceAllocationException {
         final Long vmId = cmd.getVmId();
@@ -1117,6 +1149,7 @@ public boolean createNetBackup(final CreateNetBackupCmd cmd) throws ResourceAllo
             throw new CloudRuntimeException("Failed to get NetBackup provider for existing offering assignment");
         }
 
+        checkNoActiveFastCloneFlattenForBackup(vm.getId());
         final VMInstanceVO assignedVm = transactionAssignVMToBackupOffering(vm, netBackupOffering, backupProvider);
         if (assignedVm == null) {
             throw new CloudRuntimeException(String.format("Failed to assign existing NetBackup offering [%s] to VM [%s].",
@@ -2961,6 +2994,7 @@ public void scheduleBackups() {
     @Override
     public boolean start() {
         initializeBackupProviderMap();
+        startConfiguredCommvaultBackupAgentInstallTask();
 
         currentTimestamp = new Date();
         for (final BackupScheduleVO backupSchedule : backupScheduleDao.listAll()) {
@@ -2982,6 +3016,110 @@ protected void runInContext() {
         return true;
     }
 
+    private void startConfiguredCommvaultBackupAgentInstallTask() {
+        Thread installTask = new Thread(new ManagedContextRunnable() {
+            @Override
+            protected void runInContext() {
+                for (int attempt = 1; attempt <= COMMVAULT_BACKUP_AGENT_INSTALL_RETRY_ATTEMPTS; attempt++) {
+                    if (attempt > 1 && !waitBeforeNextCommvaultBackupAgentInstallAttempt()) {
+                        return;
+                    }
+                    logger.info("Running Commvault backup agent auto-install attempt [{}/{}].",
+                            attempt, COMMVAULT_BACKUP_AGENT_INSTALL_RETRY_ATTEMPTS);
+                    if (installConfiguredCommvaultBackupAgents()) {
+                        logger.info("Commvault backup agent auto-install finished on attempt [{}/{}].",
+                                attempt, COMMVAULT_BACKUP_AGENT_INSTALL_RETRY_ATTEMPTS);
+                        return;
+                    }
+                }
+                logger.warn("Commvault backup agent auto-install did not complete after [{}] attempts.",
+                        COMMVAULT_BACKUP_AGENT_INSTALL_RETRY_ATTEMPTS);
+            }
+        }, "CommvaultBackupAgentInstallTask");
+        installTask.setDaemon(true);
+        installTask.start();
+    }
+
+    private boolean waitBeforeNextCommvaultBackupAgentInstallAttempt() {
+        try {
+            Thread.sleep(COMMVAULT_BACKUP_AGENT_INSTALL_RETRY_INTERVAL_MS);
+            return true;
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            logger.warn("Interrupted while waiting to retry Commvault backup agent auto-install.", e);
+            return false;
+        }
+    }
+
+    private boolean installConfiguredCommvaultBackupAgents() {
+        if (!BackupFrameworkEnabled.value()) {
+            logger.debug("Skipping Commvault backup agent auto-install because backup framework is disabled globally.");
+            return true;
+        }
+        boolean completed = true;
+        for (final DataCenter dataCenter : dataCenterDao.listAllZones()) {
+            if (dataCenter == null || isDisabled(dataCenter.getId())) {
+                logger.debug("Skipping Commvault backup agent auto-install because backup framework is disabled in zone [{}].",
+                        dataCenter == null ? "NULL Zone!" : dataCenter);
+                continue;
+            }
+            completed &= installConfiguredCommvaultBackupAgent(dataCenter.getId());
+        }
+        return completed;
+    }
+
+    private boolean installConfiguredCommvaultBackupAgent(final Long zoneId) {
+        final String providersConfig = BackupProviderPlugin.valueIn(zoneId);
+        if (StringUtils.isBlank(providersConfig) || Arrays.stream(providersConfig.split(","))
+                .map(String::trim)
+                .noneMatch(BackupProviderNameUtils::isCommvaultFamily)) {
+            return true;
+        }
+
+        final GlobalLock installLock = GlobalLock.getInternLock("commvault.backup.agent.install." + zoneId);
+        try {
+            if (!installLock.lock(5)) {
+                logger.debug("Skipping Commvault backup agent auto-install in zone [{}] because another management server is running it.", zoneId);
+                return true;
+            }
+            try {
+                BackupProvider provider = getBackupProvider(BackupProviderNameUtils.ABLESTACK_COMMVAULT);
+                logger.info("Checking Commvault backup agent installation in zone [{}] during management server startup.", zoneId);
+                if (!provider.checkBackupAgent(zoneId)) {
+                    logger.info("Commvault backup agent is not ready in zone [{}]. Starting automatic installation.", zoneId);
+                    if (!provider.installBackupAgent(zoneId)) {
+                        logger.warn("Commvault backup agent automatic installation did not complete successfully in zone [{}].", zoneId);
+                        return false;
+                    }
+                }
+                return true;
+            } finally {
+                installLock.unlock();
+            }
+        } catch (Exception e) {
+            if (isPermanentCommvaultBackupAgentInstallFailure(e)) {
+                logger.error("Stopping Commvault backup agent auto-install retries in zone [{}] due to a permanent configuration failure: {}",
+                        zoneId, e.getMessage());
+                return true;
+            }
+            logger.warn("Failed to run Commvault backup agent automatic installation in zone [{}]: {}", zoneId, e.getMessage(), e);
+            return false;
+        } finally {
+            installLock.releaseRef();
+        }
+    }
+
+    private boolean isPermanentCommvaultBackupAgentInstallFailure(final Exception e) {
+        String message = e == null ? null : e.getMessage();
+        if (StringUtils.isBlank(message)) {
+            return false;
+        }
+        String normalizedMessage = message.toLowerCase(Locale.ROOT);
+        return normalizedMessage.contains("commvault backup agent automatic installation cannot continue") ||
+                (normalizedMessage.contains("software cache") &&
+                        (normalizedMessage.contains("required install media") || normalizedMessage.contains("required media version") || normalizedMessage.contains("missing")));
+    }
+
     private VMInstanceVO findVmById(final Long vmId) {
         final VMInstanceVO vm = vmInstanceDao.findById(vmId);
         if (vm == null) {
diff --git a/server/src/test/java/com/cloud/api/query/dao/UserVmJoinDaoImplTest.java b/server/src/test/java/com/cloud/api/query/dao/UserVmJoinDaoImplTest.java
index c2a60f6d65b6..10f2f090a18e 100755
--- a/server/src/test/java/com/cloud/api/query/dao/UserVmJoinDaoImplTest.java
+++ b/server/src/test/java/com/cloud/api/query/dao/UserVmJoinDaoImplTest.java
@@ -29,6 +29,7 @@
 import org.apache.cloudstack.api.ApiConstants;
 import org.apache.cloudstack.api.ResponseObject;
 import org.apache.cloudstack.api.response.UserVmResponse;
+import org.apache.cloudstack.backup.dao.BackupDao;
 import org.apache.cloudstack.extension.ExtensionHelper;
 import org.junit.After;
 import org.junit.Assert;
@@ -91,6 +92,8 @@ public class UserVmJoinDaoImplTest extends GenericDaoBaseWithTagInformationBaseT
     private ExtensionHelper extensionHelper;
     @Mock
     private VbmcDao vbmcDao;
+    @Mock
+    private BackupDao backupDao;
 
     private UserVmJoinVO userVm = new UserVmJoinVO();
     private UserVmResponse userVmResponse = new UserVmResponse();
@@ -143,6 +146,7 @@ private void prepareNewUserVmResponseForVnfAppliance() {
         Mockito.doReturn(Arrays.asList()).when(_volsDao).findUsableVolumesForInstance(vmId);
         Mockito.doReturn(Arrays.asList()).when(extensionHelper).getExtensionReservedResourceDetails(Mockito.anyLong());
         Mockito.doReturn(Collections.emptyList()).when(vbmcDao).listByVmId(vmId);
+        Mockito.doReturn(Collections.emptyList()).when(backupDao).listByVmId(null, vmId);
 
         VnfTemplateNicVO vnfNic1 = new VnfTemplateNicVO(templateId, 0L, "eth0", true, true, "first");
         VnfTemplateNicVO vnfNic2 = new VnfTemplateNicVO(templateId, 1L, "eth1", true, true, "second");
diff --git a/server/src/test/java/org/apache/cloudstack/backup/BackupManagerTest.java b/server/src/test/java/org/apache/cloudstack/backup/BackupManagerTest.java
index b16f99de8f38..9d5c6aafd057 100644
--- a/server/src/test/java/org/apache/cloudstack/backup/BackupManagerTest.java
+++ b/server/src/test/java/org/apache/cloudstack/backup/BackupManagerTest.java
@@ -115,11 +115,13 @@
 import com.cloud.storage.VMTemplateVO;
 import com.cloud.storage.Volume;
 import com.cloud.storage.VolumeApiService;
+import com.cloud.storage.VolumeDetailVO;
 import com.cloud.storage.VolumeVO;
 import com.cloud.storage.dao.DiskOfferingDao;
 import com.cloud.storage.dao.GuestOSDao;
 import com.cloud.storage.dao.VMTemplateDao;
 import com.cloud.storage.dao.VolumeDao;
+import com.cloud.storage.dao.VolumeDetailsDao;
 import com.cloud.user.Account;
 import com.cloud.user.AccountManager;
 import com.cloud.user.AccountVO;
@@ -170,6 +172,9 @@ public class BackupManagerTest {
     @Mock
     VolumeDao volumeDao;
 
+    @Mock
+    VolumeDetailsDao volumeDetailsDao;
+
     @Mock
     VMInstanceDao vmInstanceDao;
 
@@ -746,6 +751,44 @@ public void createBackupTestCreateScheduledBackup() throws ResourceAllocationExc
         }
     }
 
+    @Test
+    public void createBackupFailsWhenFastCloneFlattenIsActive() {
+        Long vmId = 1L;
+        Long zoneId = 2L;
+        Long backupOfferingId = 4L;
+
+        when(vmInstanceDao.findById(vmId)).thenReturn(vmInstanceVOMock);
+        when(vmInstanceVOMock.getDataCenterId()).thenReturn(zoneId);
+        when(vmInstanceVOMock.getBackupOfferingId()).thenReturn(backupOfferingId);
+
+        overrideBackupFrameworkConfigValue();
+        when(backupOfferingDao.findById(backupOfferingId)).thenReturn(backupOfferingVOMock);
+        when(backupOfferingVOMock.isUserDrivenBackupAllowed()).thenReturn(true);
+        when(backupOfferingVOMock.getProvider()).thenReturn("testbackupprovider");
+
+        BackupProvider backupProvider = mock(BackupProvider.class);
+        when(backupProvider.getName()).thenReturn("testbackupprovider");
+        Map<String, BackupProvider> backupProvidersMap = new HashMap<>();
+        backupProvidersMap.put(backupProvider.getName().toLowerCase(), backupProvider);
+        ReflectionTestUtils.setField(backupManager, "backupProvidersMap", backupProvidersMap);
+
+        VolumeVO volume = mock(VolumeVO.class);
+        when(volume.getId()).thenReturn(11L);
+        when(volume.getUuid()).thenReturn("volume-uuid");
+        when(volumeDao.findByInstance(vmId)).thenReturn(List.of(volume));
+        when(volumeDetailsDao.findDetail(11L, "clone.fast.flatten.status"))
+                .thenReturn(new VolumeDetailVO(11L, "clone.fast.flatten.status", "pending", false));
+
+        CreateBackupCmd cmd = Mockito.mock(CreateBackupCmd.class);
+        when(cmd.getVmId()).thenReturn(vmId);
+        when(cmd.getQuiesceVM()).thenReturn(null);
+
+        CloudRuntimeException exception = Assert.assertThrows(CloudRuntimeException.class, () -> backupManager.createBackup(cmd, asyncJobVOMock));
+
+        assertTrue(exception.getMessage().contains("SharedMountPoint clone flatten is pending"));
+        verify(backupProvider, never()).takeBackup(any(), any(), any());
+    }
+
     @Test(expected = ResourceAllocationException.class)
     public void createBackupTestResourceLimitReached() throws ResourceAllocationException {
         Long vmId = 1L;
diff --git a/ui/public/locales/en.json b/ui/public/locales/en.json
index 810bdd81360d..03cb2b8732ce 100644
--- a/ui/public/locales/en.json
+++ b/ui/public/locales/en.json
@@ -2952,6 +2952,7 @@
   "message.autoscale.vmprofile.update": "The autoscale Instance profile can be updated only when autoscaling group is DISABLED.",
   "message.backup.attach.restore": "Please confirm that you want to restore and attach the volume from the backup?",
   "message.backup.create": "Are you sure you want to create an Instance backup?",
+  "message.backup.in.progress": "Backup is currently in progress. Please retry after backup completes.",
   "message.backup.offering.remove": "Are you sure you want to remove Instance from backup offering and delete the backup chain?",
   "message.backup.restore": "Please confirm that you want to restore the Instance backup?",
   "message.cancel.shutdown": "Please confirm that you would like to cancel the  shutdown on this Management server. It will resume accepting any new Async Jobs.",
@@ -4413,6 +4414,7 @@
   "sg.remove": "SG.REMOVE",
   "sg.update": "SG.UPDATE",
   "host.agent.install": "HOST.AGENT.INSTALL",
+  "backup.agent.install": "BACKUP.AGENT.INSTALL",
   "host.reconnect": "HOST.RECONNECT",
   "host.declare.degraded": "HOST.DECLARE.DEGRADED",
   "host.cancel.degraded": "HOST.CANCEL.DEGRADED",
diff --git a/ui/public/locales/ko_KR.json b/ui/public/locales/ko_KR.json
index d8d955301c46..7d76350435df 100644
--- a/ui/public/locales/ko_KR.json
+++ b/ui/public/locales/ko_KR.json
@@ -2952,6 +2952,7 @@
   "message.autoscale.vmprofile.update": "오토스케일 VM 그룹이 비활성화된 경우에만 오토스케일 VM 프로필을 업데이트할 수 있습니다.",
   "message.backup.attach.restore": "백업에서 볼륨을 복원하고 연결할 것인지 확인하십시오.",
   "message.backup.create": "VM 백업을 생성 하시겠습니까?",
+  "message.backup.in.progress": "백업이 진행 중입니다. 백업 완료 후 다시 시도해 주세요.",
   "message.backup.offering.remove": "백업 오퍼링에서 VM을 제거하고 백업 체인을 삭제하시겠습니까?",
   "message.backup.restore": "VM 백업을 복원 할 것인지 확인하십시오.",
   "message.cancel.shutdown": "이 관리 서버에서 종료를 취소할 것인지 확인하십시오. 새 비동기 Jobs 수락을 재개합니다.",
@@ -4413,6 +4414,7 @@
   "sg.remove": "SG 제거",
   "sg.update": "SG 편집",
   "host.agent.install": "호스트 에이전트 설치",
+  "backup.agent.install": "백업 에이전트 설치",
   "host.reconnect": "호스트 재연결",
   "host.declare.degraded": "호스트 DEGRADED 선언",
   "host.cancel.degraded": "호스트 DEGRADED 취소",
diff --git a/ui/src/config/section/compute.js b/ui/src/config/section/compute.js
index 970c6b9ab3dc..bfc0e9cfb755 100644
--- a/ui/src/config/section/compute.js
+++ b/ui/src/config/section/compute.js
@@ -24,16 +24,26 @@ import kubernetesIcon from '@/assets/icons/kubernetes.svg?inline'
 
 const activeFastCloneStatuses = ['pending', 'running']
 const runningFastCloneStatuses = ['running']
+const activeBackupStatuses = ['backingup']
 const fastCloneOperationBlockedLabel = 'message.sharedmountpoint.clone.flatten.in.progress'
+const backupOperationBlockedLabel = 'message.backup.in.progress'
 
 const getFastCloneStatus = (record) => {
   return String(record?.clonefaststatus || record?.details?.['clone.fast.status'] || '').toLowerCase()
 }
 
+const getActiveBackupStatus = (record) => {
+  return String(record?.activebackupstatus || '').toLowerCase()
+}
+
 const isFastCloneFlattenActive = (record) => {
   return activeFastCloneStatuses.includes(getFastCloneStatus(record))
 }
 
+const isBackupActive = (record) => {
+  return activeBackupStatuses.includes(getActiveBackupStatus(record))
+}
+
 const isFastCloneFlattenRunning = (record) => {
   return runningFastCloneStatuses.includes(getFastCloneStatus(record))
 }
@@ -42,6 +52,10 @@ const hasFastCloneFlattenSelection = (selectedItems) => {
   return Array.isArray(selectedItems) && selectedItems.some(item => isFastCloneFlattenActive(item))
 }
 
+const hasActiveBackupSelection = (selectedItems) => {
+  return Array.isArray(selectedItems) && selectedItems.some(item => isBackupActive(item))
+}
+
 const hasFastCloneFlattenRunningSelection = (selectedItems) => {
   return Array.isArray(selectedItems) && selectedItems.some(item => isFastCloneFlattenRunning(item))
 }
@@ -50,6 +64,10 @@ const disableDuringFastCloneFlatten = (record, store, selectedItems) => {
   return isFastCloneFlattenActive(record) || hasFastCloneFlattenSelection(selectedItems)
 }
 
+const disableDuringBackup = (record, store, selectedItems) => {
+  return isBackupActive(record) || hasActiveBackupSelection(selectedItems)
+}
+
 const disableDuringFastCloneFlattenRunning = (record, store, selectedItems) => {
   return isFastCloneFlattenRunning(record) || hasFastCloneFlattenRunningSelection(selectedItems)
 }
@@ -58,6 +76,19 @@ const getFastCloneOperationTooltip = (record, store, selectedItems, fallbackLabe
   return disableDuringFastCloneFlatten(record, store, selectedItems) ? fastCloneOperationBlockedLabel : fallbackLabel
 }
 
+const getBackupOperationTooltip = (record, store, selectedItems, fallbackLabel) => {
+  return disableDuringBackup(record, store, selectedItems) ? backupOperationBlockedLabel : fallbackLabel
+}
+
+const getCloneOperationTooltip = (record, store, selectedItems) => {
+  return getBackupOperationTooltip(
+    record,
+    store,
+    selectedItems,
+    getFastCloneOperationTooltip(record, store, selectedItems, 'label.action.clone.vm')
+  )
+}
+
 const getFastCloneRunningOperationTooltip = (record, store, selectedItems, fallbackLabel) => {
   return disableDuringFastCloneFlattenRunning(record, store, selectedItems) ? fastCloneOperationBlockedLabel : fallbackLabel
 }
@@ -241,8 +272,12 @@ export default {
           dataView: true,
           popup: true,
           show: (record) => { return ['Running', 'Stopped'].includes(record.state) && record.vmtype !== 'sharedfsvm' },
-          disabled: (record, store, selectedItems) => { return (record.hostcontrolstate === 'Offline' && record.hypervisor === 'KVM') || disableDuringFastCloneFlatten(record, store, selectedItems) },
-          tooltip: (record, store, selectedItems) => getFastCloneOperationTooltip(record, store, selectedItems, 'label.action.clone.vm'),
+          disabled: (record, store, selectedItems) => {
+            return (record.hostcontrolstate === 'Offline' && record.hypervisor === 'KVM') ||
+              disableDuringFastCloneFlatten(record, store, selectedItems) ||
+              disableDuringBackup(record, store, selectedItems)
+          },
+          tooltip: (record, store, selectedItems) => getCloneOperationTooltip(record, store, selectedItems),
           component: shallowRef(defineAsyncComponent(() => import('@/views/compute/CloneVM.vue')))
         },
         {
@@ -323,7 +358,8 @@ export default {
           docHelp: 'adminguide/virtual_machines.html#creating-vm-backups',
           dataView: true,
           show: (record) => { return record.backupofferingid },
-          disabled: (record) => { return record.hostcontrolstate === 'Offline' },
+          disabled: (record, store, selectedItems) => { return record.hostcontrolstate === 'Offline' || disableDuringFastCloneFlatten(record, store, selectedItems) },
+          tooltip: (record, store, selectedItems) => getFastCloneOperationTooltip(record, store, selectedItems, 'label.create.backup'),
           popup: true,
           component: shallowRef(defineAsyncComponent(() => import('@/views/compute/StartBackup.vue')))
         },
diff --git a/ui/src/config/section/network.js b/ui/src/config/section/network.js
index 70ae3a5109a8..26cc19c3892b 100644
--- a/ui/src/config/section/network.js
+++ b/ui/src/config/section/network.js
@@ -22,6 +22,17 @@ import { isAdmin } from '@/role'
 import { isZoneCreated } from '@/utils/zone'
 import { vueProps } from '@/vue-app'
 
+const activeFastCloneStatuses = ['pending', 'running']
+const fastCloneOperationBlockedLabel = 'message.sharedmountpoint.clone.flatten.in.progress'
+
+const getFastCloneStatus = (record) => {
+  return String(record?.clonefaststatus || record?.details?.['clone.fast.status'] || '').toLowerCase()
+}
+
+const isFastCloneFlattenActive = (record) => {
+  return activeFastCloneStatuses.includes(getFastCloneStatus(record))
+}
+
 export default {
   name: 'network',
   title: 'label.network',
@@ -564,6 +575,8 @@ export default {
           dataView: true,
           args: ['virtualmachineid'],
           show: (record) => { return record.backupofferingid },
+          disabled: (record) => { return isFastCloneFlattenActive(record) },
+          tooltip: (record) => { return isFastCloneFlattenActive(record) ? fastCloneOperationBlockedLabel : 'label.create.backup' },
           mapping: {
             virtualmachineid: {
               value: (record, params) => { return record.id }
diff --git a/ui/src/views/compute/StartBackup.vue b/ui/src/views/compute/StartBackup.vue
index 496a10598719..1acf85005fba 100644
--- a/ui/src/views/compute/StartBackup.vue
+++ b/ui/src/views/compute/StartBackup.vue
@@ -122,6 +122,7 @@ export default {
         }
         this.loading = true
         postAPI('createBackup', data).then(response => {
+          this.$emit('refresh-data')
           this.$pollJob({
             jobId: response.createbackupresponse.jobid,
             title: this.$t('label.create.backup'),