@@ -146,11 +146,15 @@ type LifecycleCapabilities interface {
 // WalCapabilities describes a set of behavior needed to archive and recover WALs
 type WalCapabilities interface {
 	// ArchiveWAL calls the loaded plugins to archive a WAL file.
-	// This call is a no-op if there's no plugin implementing WAL archiving
+	// This call is a no-op if there's no plugin implementing WAL archiving.
+	// checkEmptyWalArchive tells the plugin whether it should verify the
+	// archive destination is empty before this upload; the caller already
+	// combines the Cluster's annotation with the first-archive marker file.
 	ArchiveWAL(
 		ctx context.Context,
 		cluster client.Object,
 		sourceFileName string,
+		checkEmptyWalArchive bool,
 	) error
 
 	// RestoreWAL calls the loaded plugins to archive a WAL file.
 
@@ -26,7 +26,10 @@ import (
 	"slices"
 
 	restore "github.com/cloudnative-pg/cnpg-i/pkg/restore/job"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
 )
 
 // ErrNoPluginSupportsRestoreJobHooksCapability is raised when no plugin supports the restore job hooks capability
@@ -62,8 +65,15 @@ func (data *data) innerRestore(
 		if err != nil {
 			return nil, err
 		}
+
+		// Computed here so the plugin gets the operator's decision instead
+		// of re-deriving it independently (e.g. from a Cluster annotation).
+		checkEmptyWalArchive := utils.IsEmptyWalArchiveCheckEnabled(
+			&metav1.ObjectMeta{Annotations: cluster.GetAnnotations()})
+
 		request := restore.RestoreRequest{
-			ClusterDefinition: clusterDefinition,
+			ClusterDefinition:    clusterDefinition,
+			CheckEmptyWalArchive: &checkEmptyWalArchive,
 		}
 		res, err := plugin.RestoreJobHooksClient().Restore(ctx, &request)
 		if err != nil {
 
@@ -35,14 +35,16 @@ func (data *data) ArchiveWAL(
 	ctx context.Context,
 	cluster client.Object,
 	sourceFileName string,
+	checkEmptyWalArchive bool,
 ) error {
-	return wrapAsPluginErrorIfNeeded(data.innerArchiveWAL(ctx, cluster, sourceFileName))
+	return wrapAsPluginErrorIfNeeded(data.innerArchiveWAL(ctx, cluster, sourceFileName, checkEmptyWalArchive))
 }
 
 func (data *data) innerArchiveWAL(
 	ctx context.Context,
 	cluster client.Object,
 	sourceFileName string,
+	checkEmptyWalArchive bool,
 ) error {
 	contextLogger := log.FromContext(ctx)
 
@@ -64,8 +66,9 @@ func (data *data) innerArchiveWAL(
 
 		pluginLogger := contextLogger.WithValues("pluginName", plugin.Name())
 		request := wal.WALArchiveRequest{
-			ClusterDefinition: serializedCluster,
-			SourceFileName:    sourceFileName,
+			ClusterDefinition:    serializedCluster,
+			SourceFileName:       sourceFileName,
+			CheckEmptyWalArchive: &checkEmptyWalArchive,
 		}
 
 		pluginLogger.Trace(
 
@@ -204,8 +204,8 @@ func internalRun(
 	}
 
 	// Step 1: Check if the archive location is safe to perform archiving
-	if utils.IsEmptyWalArchiveCheckEnabled(&cluster.ObjectMeta) {
-		if err := checkWalArchive(ctx, cluster, walArchiver, pgData); err != nil {
+	if shouldCheckEmptyWalArchive(ctx, cluster, pgData) {
+		if err := checkWalArchive(ctx, cluster, walArchiver); err != nil {
 			return err
 		}
 	}
@@ -301,7 +301,21 @@ func archiveWALViaPlugins(
 		}
 	}
 
-	return client.ArchiveWAL(ctx, cluster, postgres.BuildWALPath(pgData, walName))
+	// The marker file is ours to manage (created after bootstrap, removed
+	// once the first WAL archives), so we resolve the full decision here
+	// rather than asking the plugin to inspect it too.
+	checkEmptyWalArchive := shouldCheckEmptyWalArchive(ctx, cluster, pgData)
+
+	return client.ArchiveWAL(ctx, cluster, postgres.BuildWALPath(pgData, walName), checkEmptyWalArchive)
+}
+
+// shouldCheckEmptyWalArchive combines the Cluster's annotation with the
+// first-archive marker file to decide whether the WAL archive destination
+// should be verified before this upload. Used identically by the in-tree
+// archiving path and by archiveWALViaPlugins, so the composition rule lives
+// in exactly one place.
+func shouldCheckEmptyWalArchive(ctx context.Context, cluster *apiv1.Cluster, pgData string) bool {
+	return utils.IsEmptyWalArchiveCheckEnabled(&cluster.ObjectMeta) && isCheckWalArchiveFlagFilePresent(ctx, pgData)
 }
 
 // isCheckWalArchiveFlagFilePresent returns true if the file CheckEmptyWalArchiveFile is present in the PGDATA directory
@@ -326,7 +340,6 @@ func checkWalArchive(
 	ctx context.Context,
 	cluster *apiv1.Cluster,
 	walArchiver *barmanArchiver.WALArchiver,
-	pgData string,
 ) error {
 	contextLogger := log.FromContext(ctx)
 	checkWalOptions, err := walArchiver.BarmanCloudCheckWalArchiveOptions(
@@ -336,10 +349,6 @@ func checkWalArchive(
 		return err
 	}
 
-	if !isCheckWalArchiveFlagFilePresent(ctx, pgData) {
-		return nil
-	}
-
 	if err := walArchiver.CheckWalArchiveDestination(ctx, checkWalOptions); err != nil {
 		contextLogger.Error(err, "while barman-cloud-check-wal-archive")
 		return err