diff --git a/changelogs/unreleased/10419-chlins b/changelogs/unreleased/10419-chlins
new file mode 100644
index 000000000..62d3a1382
--- /dev/null
+++ b/changelogs/unreleased/10419-chlins
@@ -0,0 +1 @@
+Add in-place restore pre-flight check: target PVC must not be in use
diff --git a/design/volume-data-inplace-restore/volume-data-inplace-restore.md b/design/volume-data-inplace-restore/volume-data-inplace-restore.md
index b178b9314..4977c09d1 100644
--- a/design/volume-data-inplace-restore/volume-data-inplace-restore.md
+++ b/design/volume-data-inplace-restore/volume-data-inplace-restore.md
@@ -236,7 +236,13 @@ The key requirements for this approach are:
 Before initiating an in-place restore for a volume, Velero performs the following pre-flight checks to ensure the operation is safe and valid:
 
 #### 1. PVC is Not Actively Used by a Running Pod
-Velero verifies that the target PVC is not currently mounted or consumed by any running Pods in the cluster. If the PVC is in use, Velero will skip the in-place restore for that volume and log an error. This enforces the prerequisite that users must completely delete consuming workloads prior to the restore, which prevents data corruption and avoids deadlocks caused by the Kubernetes `pvc-protection` finalizer during PVC recreation.
+Velero verifies that the target PVC is not currently mounted or consumed by any active Pods in the cluster. If the PVC is in use, Velero will skip the in-place restore for that volume and log an error. This enforces the prerequisite that users must completely delete consuming workloads prior to the restore, which prevents data corruption and avoids deadlocks caused by the Kubernetes `pvc-protection` finalizer during PVC recreation.
+
+The "in use" semantics align with the Kubernetes `pvc-protection` controller: Pods in a terminal phase (`Succeeded`/`Failed`) do not block the restore, all other phases do, and terminating Pods are flagged in the error message so users know to simply wait and retry.
+
+The check runs on both restore paths before any side effect on the existing PVC/PV: in the PVC CSI RIA before deleting the existing PVC, and before creating the `PodVolumeRestore` on the file system path. On the file system path, Pods gated by this restore's `restore-wait` init container (identified by the restore UID in its args, and not yet terminated) are exempted: they must mount the PVC for the node-agent to restore the data, and they cannot write to the volume until this restore's `PodVolumeRestore`s complete. Leftover Pods, controller-recreated Pods, and Pods gated by a different restore still block.
+
+This check is a fail-fast validation, not an atomic guarantee; the `pvc-protection` finalizer remains the actual safety gate for PVC deletion. A residual `VolumeAttachment` check (e.g. a `Failed` Pod imposed by the control plane after a non-graceful node shutdown, where the node never unmounted the volume) may be added as a future enhancement.
 
 #### 2. PVC is Bound to the Original PV
 Velero checks whether the existing PVC in the cluster is still bound to the same PersistentVolume (PV) it was bound to at the time of the backup. If the PVC is bound to a different PV, performing an in-place restore (especially an incremental one that relies on Changed Block Tracking) may be unsafe or result in unpredictable behavior. If this check fails, Velero will log an error and skip the in-place restore for that volume.
diff --git a/pkg/podvolume/restorer.go b/pkg/podvolume/restorer.go
index 53d35215c..e135ba860 100644
--- a/pkg/podvolume/restorer.go
+++ b/pkg/podvolume/restorer.go
@@ -38,6 +38,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/label"
 	"github.com/vmware-tanzu/velero/pkg/nodeagent"
 	"github.com/vmware-tanzu/velero/pkg/repository"
+	"github.com/vmware-tanzu/velero/pkg/restore/inplace"
 	uploaderutil "github.com/vmware-tanzu/velero/pkg/uploader/util"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
@@ -179,6 +180,18 @@ func (r *restorer) RestorePodVolumes(data RestoreData, tracker *volume.RestoreVo
 			}
 		}
 
+		// Pre-flight checks for in-place restore. Pods gated by this
+		// restore's restore-wait init container are excluded: they must mount
+		// the PVC so the volume gets mounted on the node for the node-agent
+		// to write into, and they cannot write to it themselves until this
+		// restore's PodVolumeRestores complete.
+		if data.Restore.IsVolumeDataInplaceRestore() && pvc != nil {
+			if err := inplace.CheckPVCNotInUse(r.ctx, r.crClient, pvc, data.Restore.UID); err != nil {
+				errs = append(errs, err)
+				continue
+			}
+		}
+
 		volumeRestore := newPodVolumeRestore(data.Restore, data.Pod, data.BackupLocation, volume, backupInfo.snapshotID, backupInfo.snapshotSize, "", backupInfo.uploaderType, data.SourceNamespace, pvc)
 		if err := veleroclient.CreateRetryGenerateName(r.crClient, r.ctx, volumeRestore); err != nil {
 			errs = append(errs, errors.WithStack(err))
diff --git a/pkg/podvolume/restorer_test.go b/pkg/podvolume/restorer_test.go
index bd67e8480..e75f2f42b 100644
--- a/pkg/podvolume/restorer_test.go
+++ b/pkg/podvolume/restorer_test.go
@@ -37,6 +37,7 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/repository"
+	"github.com/vmware-tanzu/velero/pkg/restorehelper"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 )
 
@@ -119,6 +120,21 @@ func TestGetVolumesRepositoryType(t *testing.T) {
 	}
 }
 
+// createGatedPodObj returns the restored pod as it exists in the cluster:
+// running with the restore-wait init container injected by the given restore.
+func createGatedPodObj(restoreUID string, volumeNum int) *corev1api.Pod {
+	pod := createPodObj(true, true, true, volumeNum)
+	pod.Spec.InitContainers = append([]corev1api.Container{{
+		Name: restorehelper.WaitInitContainer,
+		Args: []string{restoreUID},
+	}}, pod.Spec.InitContainers...)
+	pod.Status.InitContainerStatuses = []corev1api.ContainerStatus{{
+		Name:  restorehelper.WaitInitContainer,
+		State: corev1api.ContainerState{Running: &corev1api.ContainerStateRunning{}},
+	}}
+	return pod
+}
+
 func createNodeAgentDaemonset() *appsv1api.DaemonSet {
 	ds := &appsv1api.DaemonSet{
 		ObjectMeta: metav1.ObjectMeta{
@@ -181,6 +197,7 @@ func TestRestorePodVolumes(t *testing.T) {
 		pvbs            []*velerov1api.PodVolumeBackup
 		restoredPod     *corev1api.Pod
 		sourceNamespace string
+		inplace         bool
 		errs            []expectError
 	}{
 		{
@@ -340,6 +357,86 @@ func TestRestorePodVolumes(t *testing.T) {
 				completedPVR,
 			},
 		},
+		{
+			name: "in-place restore blocked when the PVC is used by another running pod",
+			pvbs: []*velerov1api.PodVolumeBackup{
+				createPVBObj(true, true, 1, "kopia"),
+			},
+			inplace: true,
+			kubeClientObj: []runtime.Object{
+				createNodeAgentDaemonset(),
+				createPVCObj(1),
+				func() *corev1api.Pod {
+					pod := builder.ForPod("fake-ns", "other-pod").
+						Volumes(builder.ForVolume("fake-volume-1").PersistentVolumeClaimSource("fake-pvc-1").Result()).
+						Result()
+					pod.Status.Phase = corev1api.PodRunning
+					return pod
+				}(),
+			},
+			ctlClientObj: []runtime.Object{
+				createBackupRepoObj(),
+			},
+			restoredPod:     createPodObj(true, true, true, 1),
+			sourceNamespace: "fake-ns",
+			bsl:             "fake-bsl",
+			runtimeScheme:   scheme,
+			errs: []expectError{
+				{
+					err:        "in-place restore pre-flight check failed",
+					prefixOnly: true,
+				},
+			},
+		},
+		{
+			name: "in-place restore blocked when the pod is gated by a different restore",
+			pvbs: []*velerov1api.PodVolumeBackup{
+				createPVBObj(true, true, 1, "kopia"),
+			},
+			inplace: true,
+			kubeClientObj: []runtime.Object{
+				createNodeAgentDaemonset(),
+				createPVCObj(1),
+				createGatedPodObj("old-restore-uid", 1),
+			},
+			ctlClientObj: []runtime.Object{
+				createBackupRepoObj(),
+			},
+			restoredPod:     createPodObj(true, true, true, 1),
+			sourceNamespace: "fake-ns",
+			bsl:             "fake-bsl",
+			runtimeScheme:   scheme,
+			errs: []expectError{
+				{
+					err:        "in-place restore pre-flight check failed",
+					prefixOnly: true,
+				},
+			},
+		},
+		{
+			name: "in-place restore proceeds when the PVC is only used by the gated restored pod",
+			pvbs: []*velerov1api.PodVolumeBackup{
+				createPVBObj(true, true, 1, "kopia"),
+			},
+			inplace: true,
+			kubeClientObj: []runtime.Object{
+				createNodeAgentDaemonset(),
+				createNodeObj(),
+				createPVCObj(1),
+				createGatedPodObj("fake-restore-uid", 1),
+				createNodeAgentPodObj(true),
+			},
+			ctlClientObj: []runtime.Object{
+				createBackupRepoObj(),
+			},
+			restoredPod:     createPodObj(true, true, true, 1),
+			sourceNamespace: "fake-ns",
+			bsl:             "fake-bsl",
+			runtimeScheme:   scheme,
+			retPVRs: []*velerov1api.PodVolumeRestore{
+				completedPVR,
+			},
+		},
 	}
 
 	for _, test := range tests {
@@ -362,7 +459,12 @@ func TestRestorePodVolumes(t *testing.T) {
 
 			ensurer := repository.NewEnsurer(fakeCRClient, velerotest.NewLogger(), time.Millisecond)
 
-			restoreObj := builder.ForRestore(velerov1api.DefaultNamespace, "fake-restore").Result()
+			restoreBuilder := builder.ForRestore(velerov1api.DefaultNamespace, "fake-restore").
+				ObjectMeta(builder.WithUID("fake-restore-uid"))
+			if test.inplace {
+				restoreBuilder = restoreBuilder.ExistingVolumeDataPolicy(string(velerov1api.VolumeDataPolicyTypeFull))
+			}
+			restoreObj := restoreBuilder.Result()
 
 			rs := newRestorer(ctx, repository.NewRepoLocker(), ensurer, pvrInformer, kubeClient, fakeCRClient, restoreObj, velerotest.NewLogger())
 
diff --git a/pkg/restore/actions/csi/pvc_action.go b/pkg/restore/actions/csi/pvc_action.go
index a14b985a7..5bf8f45d1 100644
--- a/pkg/restore/actions/csi/pvc_action.go
+++ b/pkg/restore/actions/csi/pvc_action.go
@@ -44,6 +44,7 @@ import (
 	plugincommon "github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
 	riav2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/restoreitemaction/v2"
+	"github.com/vmware-tanzu/velero/pkg/restore/inplace"
 	uploaderUtil "github.com/vmware-tanzu/velero/pkg/uploader/util"
 	"github.com/vmware-tanzu/velero/pkg/util"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
@@ -236,6 +237,12 @@ func (p *pvcRestoreItemAction) executeWithDataMove(logger *logrus.Entry, input *
 		if existingPVC.Status.Phase != corev1api.ClaimBound {
 			return nil, errors.New("ExistingVolumeDataPolicy is in-place restore, but the existing PVC is not bound.")
 		}
+
+		// Pre-flight checks must pass before any side effect on the existing PVC/PV.
+		if err := inplace.CheckPVCNotInUse(ctx, p.crClient, existingPVC, input.Restore.UID); err != nil {
+			return nil, errors.WithStack(err)
+		}
+
 		// take a CSI snapshot of the existing PVC as the baseline of CBT
 		if input.Restore.IsVolumeDataInplaceIncrementalRestore() && datamover.IsVeleroBlockDataMover(dataUploadResult.DataMover) {
 			logger.Info("ExistingVolumeDataPolicy is in-place incremental restore and data mover is velero-block. Taking a CSI snapshot of the existing PVC as the baseline of CBT...")
diff --git a/pkg/restore/actions/csi/pvc_action_test.go b/pkg/restore/actions/csi/pvc_action_test.go
index 47e8937a1..15cc4ca4d 100644
--- a/pkg/restore/actions/csi/pvc_action_test.go
+++ b/pkg/restore/actions/csi/pvc_action_test.go
@@ -741,6 +741,105 @@ func TestExecuteInplaceRestore(t *testing.T) {
 	require.Equal(t, "testPV", dataDownloadList.Items[0].Spec.TargetVolume.PV)
 }
 
+// TestExecuteInplaceRestorePreflight verifies the RIA fails the item without
+// side effects when the pre-flight check fails. The in-use semantics are
+// covered by the pkg/restore/inplace unit tests.
+func TestExecuteInplaceRestorePreflight(t *testing.T) {
+	newPodUsingPVC := func(phase corev1api.PodPhase) *corev1api.Pod {
+		pod := builder.ForPod("velero", "consumer-pod").
+			Volumes(builder.ForVolume("data").PersistentVolumeClaimSource("testPVC").Result()).
+			Result()
+		pod.Status.Phase = phase
+		return pod
+	}
+
+	tests := []struct {
+		name        string
+		pod         *corev1api.Pod
+		expectBlock bool
+	}{
+		{
+			name: "no pod, restore proceeds",
+		},
+		{
+			name:        "active pod blocks the restore",
+			pod:         newPodUsingPVC(corev1api.PodRunning),
+			expectBlock: true,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			existingPVC := builder.ForPersistentVolumeClaim("velero", "testPVC").
+				VolumeName("testPV").
+				Phase(corev1api.ClaimBound).Result()
+			existingPV := builder.ForPersistentVolume("testPV").Result()
+			backup := builder.ForBackup("velero", "testBackup").SnapshotMoveData(true).Result()
+			restore := builder.ForRestore("velero", "testRestore").Backup("testBackup").
+				ObjectMeta(builder.WithUID("uid")).ExistingVolumeDataPolicy("full").Result()
+			pvcFromBackup := builder.ForPersistentVolumeClaim("velero", "testPVC").
+				ObjectMeta(builder.WithAnnotations(
+					velerov1api.VolumeSnapshotLabel, "vsName",
+					velerov1api.DataUploadNameAnnotation, "velero/testDU",
+				)).Result()
+			dataUploadResult := builder.ForConfigMap("velero", "testCM").Data("uid", "{}").
+				ObjectMeta(builder.WithLabels(
+					velerov1api.RestoreUIDLabel, "uid",
+					velerov1api.PVCNamespaceNameLabel, "velero.testPVC",
+					velerov1api.ResourceUsageLabel, label.GetValidName(string(velerov1api.VeleroResourceUsageDataUploadResult)),
+				)).Result()
+
+			crObjects := []runtime.Object{existingPVC, existingPV, backup, dataUploadResult}
+			kubeObjects := []runtime.Object{existingPVC, existingPV}
+			if tc.pod != nil {
+				crObjects = append(crObjects, tc.pod)
+				kubeObjects = append(kubeObjects, tc.pod)
+			}
+
+			pvcRIA := pvcRestoreItemAction{
+				log:        logrus.New(),
+				crClient:   velerotest.NewFakeControllerRuntimeClient(t, crObjects...),
+				kubeClient: fake.NewSimpleClientset(kubeObjects...),
+			}
+
+			pvcMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pvcFromBackup.DeepCopy())
+			require.NoError(t, err)
+			pvcFromBackupMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pvcFromBackup)
+			require.NoError(t, err)
+
+			_, err = pvcRIA.Execute(&velero.RestoreItemActionExecuteInput{
+				Item:           &unstructured.Unstructured{Object: pvcMap},
+				ItemFromBackup: &unstructured.Unstructured{Object: pvcFromBackupMap},
+				Restore:        restore,
+			})
+
+			gotPVC, getErr := pvcRIA.kubeClient.CoreV1().PersistentVolumeClaims("velero").Get(t.Context(), "testPVC", metav1.GetOptions{})
+			dataDownloadList := new(velerov2alpha1.DataDownloadList)
+			require.NoError(t, pvcRIA.crClient.List(t.Context(), dataDownloadList, &crclient.ListOptions{}))
+
+			if tc.expectBlock {
+				require.Error(t, err)
+				require.Contains(t, err.Error(), "pre-flight check failed")
+				require.Contains(t, err.Error(), "consumer-pod")
+				// No side effects: PVC untouched with the original volumeName,
+				// PV reclaim policy not patched, no DataDownload created.
+				require.NoError(t, getErr)
+				require.Equal(t, "testPV", gotPVC.Spec.VolumeName)
+				gotPV, pvErr := pvcRIA.kubeClient.CoreV1().PersistentVolumes().Get(t.Context(), "testPV", metav1.GetOptions{})
+				require.NoError(t, pvErr)
+				require.Equal(t, existingPV.Spec.PersistentVolumeReclaimPolicy, gotPV.Spec.PersistentVolumeReclaimPolicy)
+				require.Empty(t, dataDownloadList.Items)
+			} else {
+				require.NoError(t, err)
+				// The in-place restore proceeded: the existing PVC is deleted
+				// and a DataDownload is created.
+				require.True(t, apierrors.IsNotFound(getErr))
+				require.Len(t, dataDownloadList.Items, 1)
+			}
+		})
+	}
+}
+
 func TestPVCAppliesTo(t *testing.T) {
 	p := pvcRestoreItemAction{
 		log: logrus.StandardLogger(),
diff --git a/pkg/restore/inplace/preflight.go b/pkg/restore/inplace/preflight.go
new file mode 100644
index 000000000..c1a4fa3a4
--- /dev/null
+++ b/pkg/restore/inplace/preflight.go
@@ -0,0 +1,131 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package inplace holds the pre-flight checks for in-place volume data
+// restores. The checks must pass before Velero performs any side effect on
+// the existing PVC/PV.
+package inplace
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/cockroachdb/errors"
+	corev1api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	crclient "sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/vmware-tanzu/velero/pkg/restorehelper"
+)
+
+// CheckPVCNotInUse verifies the target PVC is not used by any active pod,
+// aligned with the pvc-protection controller semantics: terminal-phase pods
+// (Succeeded/Failed) don't block; all other phases do, and terminating pods
+// are flagged so the message can hint the user to wait.
+//
+// Pods gated by this restore's restore-wait init container are exempted: on
+// the file system restore path the restored pods must mount the PVC for the
+// node-agent to restore the data, and an RWX PVC may be mounted by several of
+// them. Such a pod cannot write to the volume since its workload containers
+// are blocked until this restore's PodVolumeRestores complete (see
+// gatedByThisRestore). Any other pod, including one gated by a different
+// restore whose release timing is out of our control, still blocks.
+func CheckPVCNotInUse(
+	ctx context.Context,
+	cli crclient.Client,
+	pvc *corev1api.PersistentVolumeClaim,
+	restoreUID types.UID,
+) error {
+	podList := new(corev1api.PodList)
+	if err := cli.List(ctx, podList, &crclient.ListOptions{Namespace: pvc.Namespace}); err != nil {
+		return errors.Wrapf(err, "failed to check whether PVC %s/%s is in use: failed to list pods in namespace %s", pvc.Namespace, pvc.Name, pvc.Namespace)
+	}
+
+	podsInUse := []string{}
+	terminatingOnly := true
+	for i := range podList.Items {
+		pod := &podList.Items[i]
+		if !podUsesPVC(pod, pvc.Name) ||
+			pod.Status.Phase == corev1api.PodSucceeded || pod.Status.Phase == corev1api.PodFailed ||
+			gatedByThisRestore(pod, restoreUID) {
+			continue
+		}
+		state := string(pod.Status.Phase)
+		if pod.DeletionTimestamp != nil {
+			state += ", terminating"
+		} else {
+			terminatingOnly = false
+		}
+		podsInUse = append(podsInUse, fmt.Sprintf("%s (%s)", pod.Name, state))
+	}
+	if len(podsInUse) == 0 {
+		return nil
+	}
+
+	hint := "delete the workloads consuming the PVC and retry"
+	if terminatingOnly {
+		hint = "the pod(s) are terminating; retry after they are fully removed"
+	}
+	return errors.Errorf("in-place restore pre-flight check failed, skipping volume data restore: PVC %s/%s is still in use by pod(s) [%s]: %s",
+		pvc.Namespace, pvc.Name, strings.Join(podsInUse, ", "), hint)
+}
+
+func podUsesPVC(pod *corev1api.Pod, pvcName string) bool {
+	for _, vol := range pod.Spec.Volumes {
+		if vol.PersistentVolumeClaim != nil && vol.PersistentVolumeClaim.ClaimName == pvcName {
+			return true
+		}
+	}
+	return false
+}
+
+// gatedByThisRestore reports whether the pod is blocked by the restore-wait
+// init container injected by this restore, identified by the restore UID in
+// the init container's args. Such a pod cannot write to the volume: its
+// workload containers won't start until this restore's PodVolumeRestores
+// complete and write the done signal. A terminated init container means the
+// gate is already open, so the pod is no longer exempted. Pods gated by a
+// different restore are not exempted either, since their release timing is
+// unrelated to this restore.
+func gatedByThisRestore(pod *corev1api.Pod, restoreUID types.UID) bool {
+	if restoreUID == "" {
+		return false
+	}
+
+	idx := -1
+	for i, c := range pod.Spec.InitContainers {
+		if c.Name == restorehelper.WaitInitContainer {
+			if len(c.Args) == 0 || c.Args[0] != string(restoreUID) {
+				return false
+			}
+			idx = i
+			break
+		}
+	}
+	if idx < 0 {
+		return false
+	}
+
+	for _, cs := range pod.Status.InitContainerStatuses {
+		if cs.Name == restorehelper.WaitInitContainer {
+			return cs.State.Terminated == nil
+		}
+	}
+	// Statuses not populated yet: the init container hasn't run, so the gate
+	// is still closed.
+	return true
+}
diff --git a/pkg/restore/inplace/preflight_test.go b/pkg/restore/inplace/preflight_test.go
new file mode 100644
index 000000000..916b786f0
--- /dev/null
+++ b/pkg/restore/inplace/preflight_test.go
@@ -0,0 +1,202 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package inplace
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1api "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+
+	"github.com/vmware-tanzu/velero/pkg/restorehelper"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+func podUsingPVC(name, pvcName string, phase corev1api.PodPhase, terminating bool) *corev1api.Pod {
+	pod := &corev1api.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: "default",
+			UID:       types.UID(name + "-uid"),
+		},
+		Spec: corev1api.PodSpec{
+			Volumes: []corev1api.Volume{
+				{
+					Name: "data",
+					VolumeSource: corev1api.VolumeSource{
+						PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+							ClaimName: pvcName,
+						},
+					},
+				},
+			},
+		},
+		Status: corev1api.PodStatus{Phase: phase},
+	}
+	if terminating {
+		now := metav1.Now()
+		pod.DeletionTimestamp = &now
+		pod.Finalizers = []string{"fake-finalizer"}
+	}
+	return pod
+}
+
+// gatedPod adds the restore-wait init container (as injected by the
+// PodVolumeRestoreAction RIA, carrying the restore UID in args) to the pod.
+// terminated simulates the gate having already been released.
+func gatedPod(pod *corev1api.Pod, restoreUID string, terminated bool) *corev1api.Pod {
+	pod.Spec.InitContainers = append([]corev1api.Container{{
+		Name: restorehelper.WaitInitContainer,
+		Args: []string{restoreUID},
+	}}, pod.Spec.InitContainers...)
+	status := corev1api.ContainerStatus{
+		Name:  restorehelper.WaitInitContainer,
+		State: corev1api.ContainerState{Running: &corev1api.ContainerStateRunning{}},
+	}
+	if terminated {
+		status.State = corev1api.ContainerState{Terminated: &corev1api.ContainerStateTerminated{}}
+	}
+	pod.Status.InitContainerStatuses = []corev1api.ContainerStatus{status}
+	return pod
+}
+
+func TestCheckPVCNotInUse(t *testing.T) {
+	tests := []struct {
+		name          string
+		pods          []*corev1api.Pod
+		restoreUID    types.UID
+		expectPass    bool
+		expectMessage []string
+	}{
+		{
+			name:       "no pods, check passes",
+			expectPass: true,
+		},
+		{
+			name:          "active pod blocks with the delete hint",
+			pods:          []*corev1api.Pod{podUsingPVC("pod-1", "pvc-1", corev1api.PodRunning, false)},
+			expectMessage: []string{"pod-1 (Running)", "delete the workloads"},
+		},
+		{
+			name:          "unknown-phase pod blocks (node may be unreachable)",
+			pods:          []*corev1api.Pod{podUsingPVC("pod-1", "pvc-1", corev1api.PodUnknown, false)},
+			expectMessage: []string{"pod-1 (Unknown)"},
+		},
+		{
+			name:          "terminating pod blocks with the wait hint",
+			pods:          []*corev1api.Pod{podUsingPVC("pod-1", "pvc-1", corev1api.PodRunning, true)},
+			expectMessage: []string{"pod-1 (Running, terminating)", "retry after they are fully removed"},
+		},
+		{
+			name: "terminal-phase pods do not block",
+			pods: []*corev1api.Pod{
+				podUsingPVC("pod-1", "pvc-1", corev1api.PodSucceeded, false),
+				podUsingPVC("pod-2", "pvc-1", corev1api.PodFailed, false),
+			},
+			expectPass: true,
+		},
+		{
+			name:       "pod using another PVC does not block",
+			pods:       []*corev1api.Pod{podUsingPVC("pod-1", "other-pvc", corev1api.PodRunning, false)},
+			expectPass: true,
+		},
+		{
+			name: "pods gated by this restore do not block, other pods still do",
+			pods: []*corev1api.Pod{
+				gatedPod(podUsingPVC("restored-pod-1", "pvc-1", corev1api.PodPending, false), "restore-uid", false),
+				gatedPod(podUsingPVC("restored-pod-2", "pvc-1", corev1api.PodPending, false), "restore-uid", false),
+				podUsingPVC("other-pod", "pvc-1", corev1api.PodRunning, false),
+			},
+			restoreUID:    "restore-uid",
+			expectMessage: []string{"[other-pod (Running)]"},
+		},
+		{
+			name: "multiple pods gated by this restore pass the check",
+			pods: []*corev1api.Pod{
+				gatedPod(podUsingPVC("restored-pod-1", "pvc-1", corev1api.PodPending, false), "restore-uid", false),
+				gatedPod(podUsingPVC("restored-pod-2", "pvc-1", corev1api.PodPending, false), "restore-uid", false),
+			},
+			restoreUID: "restore-uid",
+			expectPass: true,
+		},
+		{
+			name: "pod gated by a different restore still blocks",
+			pods: []*corev1api.Pod{
+				gatedPod(podUsingPVC("old-restored-pod", "pvc-1", corev1api.PodPending, false), "old-restore-uid", false),
+			},
+			restoreUID:    "restore-uid",
+			expectMessage: []string{"[old-restored-pod (Pending)]"},
+		},
+		{
+			name: "pod whose restore-wait already terminated still blocks",
+			pods: []*corev1api.Pod{
+				gatedPod(podUsingPVC("released-pod", "pvc-1", corev1api.PodRunning, false), "restore-uid", true),
+			},
+			restoreUID:    "restore-uid",
+			expectMessage: []string{"[released-pod (Running)]"},
+		},
+		{
+			name: "gated pod without init container status yet passes the check",
+			pods: []*corev1api.Pod{
+				func() *corev1api.Pod {
+					pod := gatedPod(podUsingPVC("new-pod", "pvc-1", corev1api.PodPending, false), "restore-uid", false)
+					pod.Status.InitContainerStatuses = nil
+					return pod
+				}(),
+			},
+			restoreUID: "restore-uid",
+			expectPass: true,
+		},
+		{
+			name: "empty restore UID exempts nothing",
+			pods: []*corev1api.Pod{
+				gatedPod(podUsingPVC("restored-pod", "pvc-1", corev1api.PodPending, false), "", false),
+			},
+			restoreUID:    "",
+			expectMessage: []string{"[restored-pod (Pending)]"},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			objs := []runtime.Object{}
+			for _, pod := range tc.pods {
+				objs = append(objs, pod)
+			}
+			cli := velerotest.NewFakeControllerRuntimeClient(t, objs...)
+
+			pvc := &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{Name: "pvc-1", Namespace: "default"},
+			}
+
+			err := CheckPVCNotInUse(t.Context(), cli, pvc, tc.restoreUID)
+			if tc.expectPass {
+				require.NoError(t, err)
+				return
+			}
+			require.Error(t, err)
+			assert.Contains(t, err.Error(), "pre-flight check failed")
+			for _, fragment := range tc.expectMessage {
+				assert.Contains(t, err.Error(), fragment)
+			}
+		})
+	}
+}