diff --git a/.github/workflows/longhaul-smoke.yml b/.github/workflows/longhaul-smoke.yml
index 04a6be612..564abe44a 100644
--- a/.github/workflows/longhaul-smoke.yml
+++ b/.github/workflows/longhaul-smoke.yml
@@ -46,6 +46,19 @@
 #   (LONGHAUL_MIN_INSTANCES == LONGHAUL_MAX_INSTANCES) so the gate is a fast,
 #   deterministic data-durability check (writers + verifier) that fits a
 #   GitHub-hosted runner and finishes in a few minutes.
+#
+# Data-protection gate
+#   The backup verifier is exercised for real, not just compiled in. The kind
+#   cluster already has CSI VolumeSnapshot support (setup-test-environment runs
+#   deploy-csi-driver.sh: external-snapshotter + a default csi-hostpath
+#   VolumeSnapshotClass), so a single-instance cluster can complete snapshot
+#   backups — exactly as the e2e scheduled-backup test proves. The smoke run
+#   therefore sets a fast backup schedule (every minute) and asserts the driver
+#   actually scheduled AND completed at least one backup (with no retention leak
+#   or completion stall). Because the backup verify loop ticks every 5m, the
+#   bounded window is >= 6m so that tick fires once against several
+#   already-completed backups; a broken backup path (no completion) fails the
+#   PR rather than passing silently as a no-op.
 
 name: Long-Haul Smoke Gate
 
@@ -61,9 +74,9 @@ on:
   workflow_dispatch:
     inputs:
       max_duration:
-        description: "Bounded driver run length (Go duration, e.g. 3m)"
+        description: "Bounded driver run length (Go duration). Keep >= 6m so the 5m backup verify tick fires."
         required: false
-        default: "3m"
+        default: "6m"
 
 permissions:
   contents: read
@@ -115,7 +128,7 @@ jobs:
       # Must match the cluster name the composite action derives:
       #   documentdb-<test-type>-<architecture>-<test-scenario-name>
       KIND_CLUSTER: documentdb-longhaul-amd64-smoke
-      MAX_DURATION: ${{ github.event.inputs.max_duration || '3m' }}
+      MAX_DURATION: ${{ github.event.inputs.max_duration || '6m' }}
     steps:
       - name: Checkout
         uses: actions/checkout@v4
@@ -205,6 +218,10 @@ jobs:
           #  - RESET_DATA: fresh collection each CI run
           #  - MIN==MAX instances: disable disruptive scale ops (fast + stable)
           #  - short cadences so the verifier gets several cycles in the window
+          #  - BACKUP_*: exercise the data-protection verifier for real — a
+          #    per-minute schedule so backups are scheduled + completed well
+          #    before the backup verifier's 5m tick observes them. Retention is
+          #    1 day (>> the run) so no leak fires; we assert scheduled+completed.
           PATCH=$(jq -nc \
             --arg dur "${MAX_DURATION}" \
             '{data: {
@@ -216,7 +233,10 @@ jobs:
               LONGHAUL_RECOVERY_TIMEOUT: "2m",
               LONGHAUL_REPORT_INTERVAL: "30s",
               LONGHAUL_MIN_INSTANCES: "1",
-              LONGHAUL_MAX_INSTANCES: "1"
+              LONGHAUL_MAX_INSTANCES: "1",
+              LONGHAUL_BACKUP_ENABLED: "true",
+              LONGHAUL_BACKUP_SCHEDULE: "*/1 * * * *",
+              LONGHAUL_BACKUP_RETENTION_DAYS: "1"
             }}')
           kubectl patch configmap longhaul-test-config -n "${DB_NS}" --type merge -p "${PATCH}"
 
@@ -286,6 +306,44 @@ jobs:
           fi
           echo "✅ Long-haul smoke gate passed (exit 0, report PASS)."
 
+      - name: Assert data-protection verifier ran
+        run: |
+          set -euo pipefail
+          # The overall PASS only means "no leak/stall/data-loss". With zero
+          # backups that is trivially true, so a broken/no-op backup path would
+          # slip through. Assert the verifier actually scheduled AND completed a
+          # backup — proof the data-protection path functioned end-to-end.
+          report=$(kubectl get configmap longhaul-report -n "${DB_NS}" \
+            -o jsonpath='{.data.latest-report}' 2>/dev/null || echo "")
+          if [[ -z "${report}" ]]; then
+            echo "::error::longhaul-report has no latest-report body to inspect."
+            exit 1
+          fi
+
+          extract() { echo "${report}" | sed -n "s/^| $1 | \([0-9][0-9]*\) |.*/\1/p"; }
+          scheduled=$(extract "Backups Scheduled" || true)
+          completed=$(extract "Backups Completed" || true)
+          leaks=$(extract "Retention Leaks" || true)
+          stall=$(extract "Max Scheduled Without Completion" || true)
+
+          echo "===== Data Protection ====="
+          echo "${report}" | grep -E '^\| (Backups|Live Backup|Retention Leaks|Max Scheduled) ' || true
+          echo "Scheduled=${scheduled:-?} Completed=${completed:-?} Leaks=${leaks:-?} MaxNoCompletion=${stall:-?}"
+
+          if [[ -z "${scheduled}" || -z "${completed}" ]]; then
+            echo "::error::Could not parse backup metrics from the report."
+            exit 1
+          fi
+          if (( scheduled < 1 )); then
+            echo "::error::Backup verifier observed no scheduled backups (scheduled=${scheduled}); the data-protection path did not run."
+            exit 1
+          fi
+          if (( completed < 1 )); then
+            echo "::error::Backup verifier observed no completed backups (completed=${completed}); backups scheduled but never completed."
+            exit 1
+          fi
+          echo "✅ Data-protection verifier ran (scheduled=${scheduled}, completed=${completed})."
+
       - name: Diagnostics on failure
         if: failure()
         run: |
@@ -297,6 +355,11 @@ jobs:
           kubectl logs -n "${DB_NS}" -l app.kubernetes.io/name=longhaul-test --previous --tail=200 || true
           echo "===== longhaul-report ConfigMap ====="
           kubectl get configmap longhaul-report -n "${DB_NS}" -o yaml || true
+          echo "===== ScheduledBackup + child Backups ====="
+          kubectl get scheduledbackups.documentdb.io -n "${DB_NS}" -o wide || true
+          kubectl get backups.documentdb.io -n "${DB_NS}" -o wide || true
+          echo "===== VolumeSnapshots ====="
+          kubectl get volumesnapshots -n "${DB_NS}" -o wide || true
           echo "===== DocumentDB describe ====="
           kubectl describe documentdb "${DB_NAME}" -n "${DB_NS}" || true
           echo "===== Operator logs ====="
diff --git a/.gitignore b/.gitignore
index 3db05b243..8e4de1d5f 100644
--- a/.gitignore
+++ b/.gitignore
@@ -278,6 +278,9 @@ Backup*/
 !test/e2e/manifests/backup/
 !test/e2e/tests/backup/
 !test/e2e/pkg/e2eutils/backup/
+# The long-haul driver's data-protection component lives here and is
+# likewise swallowed by the generic Backup*/ rule above.
+!test/longhaul/backup/
 UpgradeLog*.XML
 UpgradeLog*.htm
 ServiceFabricBackup/
diff --git a/test/longhaul/README.md b/test/longhaul/README.md
index d87f153f1..c2087847b 100644
--- a/test/longhaul/README.md
+++ b/test/longhaul/README.md
@@ -132,8 +132,51 @@ All configuration is via environment variables.
 | `LONGHAUL_MIN_INSTANCES` | No | `1` | Minimum `spec.instancesPerNode` for scale-down operations (CRD lower bound: 1). |
 | `LONGHAUL_MAX_INSTANCES` | No | `3` | Maximum `spec.instancesPerNode` for scale-up operations (CRD upper bound: 3). |
 | `LONGHAUL_REPORT_INTERVAL` | No | `1h` | How often to write checkpoint reports to ConfigMap. |
+| `LONGHAUL_BACKUP_ENABLED` | No | `true` | Enable the ScheduledBackup + retention verifier. |
+| `LONGHAUL_BACKUP_SCHEDULE` | No | `0 */6 * * *` | Cron schedule for the canary `ScheduledBackup`. |
+| `LONGHAUL_BACKUP_RETENTION_DAYS` | No | `1` | Retention window applied to child backups; also used to derive the retention-leak deadline. |
 | `LONGHAUL_RESET_DATA` | No | `false` | If `true`, drop the workload collection on startup. Off by default so a Deployment pod restart preserves durability history. |
 
+### Data Protection (ScheduledBackup + retention)
+
+When `LONGHAUL_BACKUP_ENABLED` is true, the driver ensures a `ScheduledBackup`
+named `<cluster>-longhaul` exists and matches the run's schedule/retention
+(an existing CR is reconciled in place, never recreated, so backup history is
+preserved across restarts and parameter changes) and runs a verifier
+concurrently with the operation scheduler (backup is deliberately **not**
+isolated from topology/chaos, per the design).
+
+The verifier only checks the properties a **multi-day** run can establish —
+things unit and e2e tests cannot:
+
+- **Scheduling liveness** — `status.lastScheduledTime` keeps advancing; a stalled
+  scheduler (past `status.nextScheduledTime` + grace) raises a warning.
+- **Completion** — child `Backup` CRs keep reaching `completed`; only terminal
+  `failed` backups are counted as failures. A `skipped` backup is an intentional
+  no-op (e.g. the operator declines to back up a non-primary/standby) and is
+  **not** counted as a failure. If backups keep being scheduled but stop
+  completing for 3 consecutive schedules (a dead completion path — every backup
+  failing or hanging), the run is a **FAIL**. A single completed backup resets
+  this gap, so transient chaos-induced failures are tolerated.
+- **Retention leak** — no completed backup outlives its retention window
+  (`stoppedAt + spec.retentionDays*24h` + grace). The window is taken from each
+  backup's **own** `spec.retentionDays` (stamped at creation), so the check
+  stays correct even if a later run uses a different retention. A lingering
+  backup is a **FAIL**: expired backups aren't garbage-collected and the
+  population (and its PVCs / VolumeSnapshots) grows unbounded.
+
+It deliberately does **not** re-verify the operator's retention *arithmetic*
+(`expiredAt == stoppedAt + retentionDays*24h`) — that is a pure function already
+covered by the operator's unit tests and needs no accumulation. The oracle here
+is black-box: expired backups disappear. Because the minimum meaningful
+retention is 1 day, the leak check only fires on multi-day runs — exactly the
+accumulation window long-haul exists to cover.
+
+> **RBAC.** The driver ServiceAccount needs `create`/`get`/`list`/`update` on
+> `scheduledbackups.documentdb.io` and `list` on `backups.documentdb.io`. These
+> verbs are granted by the `longhaul-test` Role in `deploy/rbac.yaml`; without
+> them the backup verifier logs an error and the rest of the run continues.
+
 ## CI Safety
 
 The long haul test binary is deployed as a Kubernetes Deployment on a dedicated AKS
diff --git a/test/longhaul/backup/metrics.go b/test/longhaul/backup/metrics.go
new file mode 100644
index 000000000..6ecc08d20
--- /dev/null
+++ b/test/longhaul/backup/metrics.go
@@ -0,0 +1,139 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+// Package backup implements the data-protection component of the long haul
+// driver. It provisions a ScheduledBackup against the canary cluster and
+// continuously verifies the properties that only a multi-day run can
+// establish: that backups keep being produced on schedule and completing,
+// and that expired backups are actually garbage-collected so the backup
+// population stays bounded over time (no PVC / VolumeSnapshot accumulation).
+//
+// It deliberately does NOT re-verify the operator's retention *arithmetic*
+// (expiredAt == stoppedAt + retentionDays*24h) — that is a pure function
+// already covered by the operator's unit tests and needs no accumulation.
+// The oracle here is black-box: expired backups disappear.
+//
+// The component runs concurrently with the operation scheduler — per the
+// long-haul design, backup is deliberately NOT isolated so that
+// backup-vs-topology serialization bugs surface here rather than in
+// production.
+package backup
+
+import (
+	"sync/atomic"
+	"time"
+)
+
+// completionStallThreshold is the number of consecutive backups that may be
+// scheduled with no intervening completion before the run is failed. A single
+// recovered (completed) backup resets the running gap, so transient chaos-
+// induced failures stay well under this ceiling; only a wholly-broken
+// completion path (every backup failing or hanging) drives the gap this high.
+// At the default 6h schedule this is ~18h of zero successful backups.
+const completionStallThreshold = 3
+
+// Metrics tracks aggregate backup-verification counters using atomic
+// operations so the reporter goroutine can snapshot them without locking.
+//
+// Two independent oracles flip the run verdict to FAIL:
+//   - RetentionLeaks > 0: a completed backup outlived its retention window
+//     (the operator failed to garbage-collect it).
+//   - MaxScheduledWithoutCompletion >= completionStallThreshold: backups keep
+//     being scheduled but stop completing (a dead completion path the leak
+//     oracle alone would miss).
+//
+// The remaining counters are observational and feed the report.
+type Metrics struct {
+	// Scheduled counts backups observed to have been scheduled by the
+	// ScheduledBackup (advances of status.lastScheduledTime).
+	Scheduled atomic.Int64
+
+	// Completed is the number of child backups observed in the "completed"
+	// phase (deduplicated by name across verification cycles).
+	Completed atomic.Int64
+
+	// Failed is the number of child backups observed in a terminal failure
+	// phase (deduplicated by name).
+	Failed atomic.Int64
+
+	// RetentionLeaks counts completed backups still present past their
+	// retention window (stoppedAt + retentionDays*24h). Non-zero => FAIL.
+	RetentionLeaks atomic.Int64
+
+	// MaxScheduledWithoutCompletion is the high-water mark of consecutive
+	// backups scheduled with no intervening completion. It is the completion-
+	// liveness oracle: a wholly-broken backup path drives it monotonically
+	// upward while Completed stays flat. Reaching completionStallThreshold
+	// flips the verdict to FAIL. See observeCompletionGap for how it advances.
+	MaxScheduledWithoutCompletion atomic.Int64
+
+	// LastChildCount is the number of child backups observed on the most
+	// recent verification cycle (the live backup population — expected to
+	// stabilize near retentionWindow/scheduleInterval at steady state).
+	LastChildCount atomic.Int64
+
+	// LastScheduledUnix is the Unix timestamp of the most recently observed
+	// status.lastScheduledTime; 0 until the first backup is scheduled.
+	LastScheduledUnix atomic.Int64
+}
+
+// NewMetrics creates an empty Metrics.
+func NewMetrics() *Metrics {
+	return &Metrics{}
+}
+
+// observeCompletionGap records gap as a new high-water mark for consecutive
+// scheduled-without-completion backups if it exceeds the current maximum.
+func (m *Metrics) observeCompletionGap(gap int64) {
+	for {
+		cur := m.MaxScheduledWithoutCompletion.Load()
+		if gap <= cur {
+			return
+		}
+		if m.MaxScheduledWithoutCompletion.CompareAndSwap(cur, gap) {
+			return
+		}
+	}
+}
+
+// MetricsSnapshot is a point-in-time copy of Metrics.
+type MetricsSnapshot struct {
+	Scheduled                     int64
+	Completed                     int64
+	Failed                        int64
+	RetentionLeaks                int64
+	MaxScheduledWithoutCompletion int64
+	LastChildCount                int64
+	LastScheduled                 time.Time
+}
+
+// Snapshot captures the current metric values atomically.
+func (m *Metrics) Snapshot() MetricsSnapshot {
+	var lastScheduled time.Time
+	if unix := m.LastScheduledUnix.Load(); unix > 0 {
+		lastScheduled = time.Unix(unix, 0)
+	}
+	return MetricsSnapshot{
+		Scheduled:                     m.Scheduled.Load(),
+		Completed:                     m.Completed.Load(),
+		Failed:                        m.Failed.Load(),
+		RetentionLeaks:                m.RetentionLeaks.Load(),
+		MaxScheduledWithoutCompletion: m.MaxScheduledWithoutCompletion.Load(),
+		LastChildCount:                m.LastChildCount.Load(),
+		LastScheduled:                 lastScheduled,
+	}
+}
+
+// HasRetentionLeak returns true if any completed backup has outlived its
+// retention window. A true result flips the overall run verdict to FAIL.
+func (s MetricsSnapshot) HasRetentionLeak() bool {
+	return s.RetentionLeaks > 0
+}
+
+// HasCompletionStall returns true if backups kept being scheduled but stopped
+// completing for completionStallThreshold consecutive schedules. A true result
+// flips the overall run verdict to FAIL, catching a wholly-broken backup path
+// that the retention-leak oracle alone would miss.
+func (s MetricsSnapshot) HasCompletionStall() bool {
+	return s.MaxScheduledWithoutCompletion >= completionStallThreshold
+}
diff --git a/test/longhaul/backup/metrics_test.go b/test/longhaul/backup/metrics_test.go
new file mode 100644
index 000000000..4f7359617
--- /dev/null
+++ b/test/longhaul/backup/metrics_test.go
@@ -0,0 +1,69 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package backup
+
+import (
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Metrics", func() {
+	It("snapshots counters atomically", func() {
+		m := NewMetrics()
+		m.Scheduled.Add(3)
+		m.Completed.Add(2)
+		m.Failed.Add(1)
+		m.RetentionLeaks.Add(1)
+		m.LastChildCount.Store(7)
+		now := time.Now()
+		m.LastScheduledUnix.Store(now.Unix())
+
+		snap := m.Snapshot()
+		Expect(snap.Scheduled).To(Equal(int64(3)))
+		Expect(snap.Completed).To(Equal(int64(2)))
+		Expect(snap.Failed).To(Equal(int64(1)))
+		Expect(snap.RetentionLeaks).To(Equal(int64(1)))
+		Expect(snap.LastChildCount).To(Equal(int64(7)))
+		Expect(snap.LastScheduled.Unix()).To(Equal(now.Unix()))
+	})
+
+	It("reports zero LastScheduled when never scheduled", func() {
+		snap := NewMetrics().Snapshot()
+		Expect(snap.LastScheduled.IsZero()).To(BeTrue())
+	})
+
+	DescribeTable("HasRetentionLeak",
+		func(leaks int64, want bool) {
+			m := NewMetrics()
+			m.RetentionLeaks.Add(leaks)
+			Expect(m.Snapshot().HasRetentionLeak()).To(Equal(want))
+		},
+		Entry("clean", int64(0), false),
+		Entry("one leak", int64(1), true),
+		Entry("several leaks", int64(3), true),
+	)
+
+	DescribeTable("HasCompletionStall",
+		func(gap int64, want bool) {
+			m := NewMetrics()
+			m.observeCompletionGap(gap)
+			Expect(m.Snapshot().HasCompletionStall()).To(Equal(want))
+		},
+		Entry("no schedules", int64(0), false),
+		Entry("one in flight", int64(1), false),
+		Entry("two in flight", int64(2), false),
+		Entry("at threshold", int64(completionStallThreshold), true),
+		Entry("past threshold", int64(completionStallThreshold+2), true),
+	)
+
+	It("observeCompletionGap keeps only the high-water mark", func() {
+		m := NewMetrics()
+		m.observeCompletionGap(2)
+		m.observeCompletionGap(5)
+		m.observeCompletionGap(3) // lower value must not lower the mark
+		Expect(m.Snapshot().MaxScheduledWithoutCompletion).To(Equal(int64(5)))
+	})
+})
diff --git a/test/longhaul/backup/suite_test.go b/test/longhaul/backup/suite_test.go
new file mode 100644
index 000000000..752094654
--- /dev/null
+++ b/test/longhaul/backup/suite_test.go
@@ -0,0 +1,16 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package backup
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestBackup(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Long Haul Backup Suite")
+}
diff --git a/test/longhaul/backup/verifier.go b/test/longhaul/backup/verifier.go
new file mode 100644
index 000000000..068a0684e
--- /dev/null
+++ b/test/longhaul/backup/verifier.go
@@ -0,0 +1,274 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package backup
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+
+	previewv1 "github.com/documentdb/documentdb-operator/api/preview"
+	"github.com/documentdb/documentdb-operator/test/longhaul/journal"
+)
+
+const (
+	// gcGrace is how long past a backup's retention deadline
+	// (stoppedAt + retentionDays*24h) it may still exist before we treat it
+	// as a garbage-collection leak. It absorbs the backup controller's
+	// requeue latency and apiserver propagation.
+	gcGrace = 10 * time.Minute
+
+	// livenessGrace is how far past status.nextScheduledTime the scheduler
+	// may run before we warn that scheduling appears stalled.
+	livenessGrace = 5 * time.Minute
+
+	// verifyInterval is how often the verification loop runs.
+	verifyInterval = 5 * time.Minute
+)
+
+// Client is the subset of the cluster API the backup verifier needs.
+// monitor.K8sClusterClient satisfies it structurally.
+type Client interface {
+	EnsureScheduledBackup(ctx context.Context, name, schedule string, retentionDays int) error
+	GetScheduledBackup(ctx context.Context, name string) (*previewv1.ScheduledBackup, error)
+	ListScheduledChildBackups(ctx context.Context, scheduledBackupName string) ([]previewv1.Backup, error)
+}
+
+// Config parameterizes the backup verifier.
+type Config struct {
+	ScheduledBackupName string
+	Schedule            string
+	RetentionDays       int
+}
+
+// Verifier bootstraps a ScheduledBackup and continuously checks that the
+// operator schedules, completes, and retires backups per policy.
+type Verifier struct {
+	client  Client
+	journal *journal.Journal
+	metrics *Metrics
+	cfg     Config
+
+	// dedup state — the verification loop re-observes the same backups every
+	// cycle, so we count each transition/violation exactly once by name.
+	// Pruned to the live backup set each cycle (see checkChildBackups) so
+	// these maps stay bounded over a multi-day run.
+	seenCompleted  map[string]struct{}
+	seenFailed     map[string]struct{}
+	leakFlagged    map[string]struct{}
+	lastScheduled  time.Time
+	stallWarnedFor time.Time
+
+	// scheduledSinceCompleted counts backups scheduled since the last observed
+	// completion; it feeds the completion-stall oracle. Verifier-goroutine
+	// private (checkScheduling and recordPhase both run via checkOnce), so it
+	// needs no synchronization; only its high-water mark is published to
+	// Metrics for the reporter goroutine.
+	scheduledSinceCompleted int64
+}
+
+// NewVerifier creates a backup verifier. metrics must be non-nil.
+func NewVerifier(client Client, j *journal.Journal, metrics *Metrics, cfg Config) *Verifier {
+	return &Verifier{
+		client:        client,
+		journal:       j,
+		metrics:       metrics,
+		cfg:           cfg,
+		seenCompleted: make(map[string]struct{}),
+		seenFailed:    make(map[string]struct{}),
+		leakFlagged:   make(map[string]struct{}),
+	}
+}
+
+// Bootstrap ensures the ScheduledBackup CR exists and matches this run's
+// schedule/retention. It is safe to call on every startup: an existing CR is
+// reconciled in place (never recreated), so the accumulated backup history
+// survives driver restarts and parameter changes.
+func (v *Verifier) Bootstrap(ctx context.Context) error {
+	if err := v.client.EnsureScheduledBackup(ctx, v.cfg.ScheduledBackupName, v.cfg.Schedule, v.cfg.RetentionDays); err != nil {
+		return fmt.Errorf("ensure ScheduledBackup: %w", err)
+	}
+	v.journal.Info("backup", fmt.Sprintf("ScheduledBackup %q ensured (schedule=%q retentionDays=%d)",
+		v.cfg.ScheduledBackupName, v.cfg.Schedule, v.cfg.RetentionDays))
+	return nil
+}
+
+// Run starts the verification loop. It blocks until ctx is cancelled.
+func (v *Verifier) Run(ctx context.Context) {
+	v.journal.Info("backup", "backup verifier started")
+	defer v.journal.Info("backup", "backup verifier stopped")
+
+	ticker := time.NewTicker(verifyInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			v.checkOnce(ctx, time.Now())
+		}
+	}
+}
+
+// checkOnce runs a single verification pass. now is injected so unit tests
+// can drive time deterministically.
+func (v *Verifier) checkOnce(ctx context.Context, now time.Time) {
+	v.checkScheduling(ctx, now)
+	v.checkChildBackups(ctx, now)
+	// Publish the completion-stall high-water mark once per cycle, after both
+	// new schedules (which widen the gap) and completions (which reset it) have
+	// been applied — so a backup that completes in the same cycle it is
+	// observed properly cancels the schedule and the gap only grows when
+	// completions genuinely stop.
+	v.metrics.observeCompletionGap(v.scheduledSinceCompleted)
+}
+
+// checkScheduling inspects the ScheduledBackup status: it counts new
+// scheduling events and warns if the scheduler appears stalled.
+func (v *Verifier) checkScheduling(ctx context.Context, now time.Time) {
+	sb, err := v.client.GetScheduledBackup(ctx, v.cfg.ScheduledBackupName)
+	if err != nil {
+		v.journal.Warn("backup", fmt.Sprintf("cannot read ScheduledBackup (will retry): %v", err))
+		return
+	}
+
+	if sb.Status.LastScheduledTime != nil {
+		last := sb.Status.LastScheduledTime.Time
+		if last.After(v.lastScheduled) {
+			v.lastScheduled = last
+			v.metrics.Scheduled.Add(1)
+			v.metrics.LastScheduledUnix.Store(last.Unix())
+			// A new backup was scheduled; widen the completion-stall gap. A
+			// completion in checkChildBackups resets it, and checkOnce publishes
+			// the high-water mark after both run.
+			v.scheduledSinceCompleted++
+			v.journal.Info("backup", fmt.Sprintf("backup scheduled at %s", last.Format(time.RFC3339)))
+		}
+	}
+
+	// Liveness: if the next scheduled time is well in the past and no new
+	// backup has been scheduled since, surface a warning (once per overdue
+	// deadline). This is a Degraded signal, not a Fatal one — scheduling may
+	// be transiently delayed by apiserver load.
+	if next := sb.Status.NextScheduledTime; next != nil {
+		deadline := next.Time.Add(livenessGrace)
+		if now.After(deadline) && !v.lastScheduled.After(next.Time) && !v.stallWarnedFor.Equal(next.Time) {
+			v.stallWarnedFor = next.Time
+			v.journal.Warn("backup", fmt.Sprintf(
+				"scheduling appears stalled: nextScheduledTime %s passed %s ago with no new backup",
+				next.Time.Format(time.RFC3339), now.Sub(next.Time).Round(time.Second)))
+		}
+	}
+}
+
+// checkChildBackups inspects every child Backup for completion, terminal
+// failure, and retention leaks.
+func (v *Verifier) checkChildBackups(ctx context.Context, now time.Time) {
+	children, err := v.client.ListScheduledChildBackups(ctx, v.cfg.ScheduledBackupName)
+	if err != nil {
+		v.journal.Warn("backup", fmt.Sprintf("cannot list child backups (will retry): %v", err))
+		return
+	}
+
+	v.metrics.LastChildCount.Store(int64(len(children)))
+
+	present := make(map[string]struct{}, len(children))
+	for i := range children {
+		b := &children[i]
+		present[b.Name] = struct{}{}
+		v.recordPhase(b)
+		v.checkRetentionLeak(b, now)
+	}
+
+	// Prune dedup state for backups the operator has garbage-collected, so
+	// the maps stay bounded by the live population over a multi-day run
+	// instead of growing for the life of the process. Backup names are
+	// timestamp-unique and never reused, so a pruned entry cannot reappear
+	// and be double-counted.
+	pruneAbsent(v.seenCompleted, present)
+	pruneAbsent(v.seenFailed, present)
+	pruneAbsent(v.leakFlagged, present)
+}
+
+// pruneAbsent deletes keys from seen that are not in present.
+func pruneAbsent(seen, present map[string]struct{}) {
+	for name := range seen {
+		if _, ok := present[name]; !ok {
+			delete(seen, name)
+		}
+	}
+}
+
+// recordPhase counts new completed / terminally-failed child backups,
+// deduplicated by name.
+func (v *Verifier) recordPhase(b *previewv1.Backup) {
+	switch {
+	case isCompleted(b):
+		if _, seen := v.seenCompleted[b.Name]; !seen {
+			v.seenCompleted[b.Name] = struct{}{}
+			v.metrics.Completed.Add(1)
+			// The completion path is alive: reset the running stall gap so the
+			// oracle only fires on a sustained run of scheduled-but-never-
+			// completed backups.
+			v.scheduledSinceCompleted = 0
+			v.journal.Info("backup", fmt.Sprintf("backup %q completed", b.Name))
+		}
+	case isTerminalFailure(b):
+		if _, seen := v.seenFailed[b.Name]; !seen {
+			v.seenFailed[b.Name] = struct{}{}
+			v.metrics.Failed.Add(1)
+			v.journal.Warn("backup", fmt.Sprintf("backup %q reached terminal failure phase %q: %s",
+				b.Name, b.Status.Phase, b.Status.Message))
+		}
+	}
+}
+
+// checkRetentionLeak flags a completed backup still present past its
+// retention window. The deadline is derived from the backup's own declared
+// retention (spec.retentionDays, stamped by the ScheduledBackup) plus
+// gcGrace — not the operator-populated status.expiredAt — so the oracle
+// stays black-box (see package doc) and stays correct even when the run's
+// configured retention differs from what an older backup was created under.
+// Falls back to the run config if a backup carries no retention.
+// Flagged once per backup.
+func (v *Verifier) checkRetentionLeak(b *previewv1.Backup, now time.Time) {
+	if !isCompleted(b) || b.Status.StoppedAt == nil {
+		return
+	}
+	if _, done := v.leakFlagged[b.Name]; done {
+		return
+	}
+	retentionDays := v.cfg.RetentionDays
+	if b.Spec.RetentionDays != nil {
+		retentionDays = *b.Spec.RetentionDays
+	}
+	deadline := b.Status.StoppedAt.Time.
+		Add(time.Duration(retentionDays) * 24 * time.Hour).
+		Add(gcGrace)
+	if now.After(deadline) {
+		v.leakFlagged[b.Name] = struct{}{}
+		v.metrics.RetentionLeaks.Add(1)
+		v.journal.Error("backup", fmt.Sprintf(
+			"retention leak: backup %q still present %s past its %d-day retention window (stoppedAt %s)",
+			b.Name, now.Sub(b.Status.StoppedAt.Time).Round(time.Second),
+			retentionDays, b.Status.StoppedAt.Time.Format(time.RFC3339)))
+	}
+}
+
+// isCompleted reports whether a Backup reached the "completed" phase.
+func isCompleted(b *previewv1.Backup) bool {
+	return b != nil && b.Status.Phase == cnpgv1.BackupPhaseCompleted
+}
+
+// isTerminalFailure reports whether a Backup reached a genuine failure
+// phase from which no reconcile will recover it. BackupPhaseSkipped is
+// deliberately excluded: the operator emits it when a backup is
+// intentionally not taken (e.g. the cluster is a non-primary/standby), which
+// is an expected no-op, not a failure of the backup machinery.
+func isTerminalFailure(b *previewv1.Backup) bool {
+	return b != nil && b.Status.Phase == cnpgv1.BackupPhaseFailed
+}
diff --git a/test/longhaul/backup/verifier_test.go b/test/longhaul/backup/verifier_test.go
new file mode 100644
index 000000000..9cc65ce10
--- /dev/null
+++ b/test/longhaul/backup/verifier_test.go
@@ -0,0 +1,376 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package backup
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	previewv1 "github.com/documentdb/documentdb-operator/api/preview"
+	"github.com/documentdb/documentdb-operator/test/longhaul/journal"
+)
+
+// fakeBackupClient is a minimal Client stub for unit tests.
+type fakeBackupClient struct {
+	ensureCalls []ensureArgs
+	ensureErr   error
+	sb          *previewv1.ScheduledBackup
+	sbErr       error
+	children    []previewv1.Backup
+	childErr    error
+}
+
+type ensureArgs struct {
+	name          string
+	schedule      string
+	retentionDays int
+}
+
+func (f *fakeBackupClient) EnsureScheduledBackup(_ context.Context, name, schedule string, retentionDays int) error {
+	f.ensureCalls = append(f.ensureCalls, ensureArgs{name, schedule, retentionDays})
+	return f.ensureErr
+}
+
+func (f *fakeBackupClient) GetScheduledBackup(_ context.Context, _ string) (*previewv1.ScheduledBackup, error) {
+	return f.sb, f.sbErr
+}
+
+func (f *fakeBackupClient) ListScheduledChildBackups(_ context.Context, _ string) ([]previewv1.Backup, error) {
+	return f.children, f.childErr
+}
+
+// mkBackup builds a child Backup with the given phase and stopped time.
+// A zero stopped time is left nil. RetentionDays/ExpiredAt are omitted
+// because the leak oracle derives its deadline from the verifier config,
+// not from the backup's own fields.
+func mkBackup(name string, phase cnpgv1.BackupPhase, stopped time.Time) previewv1.Backup {
+	b := previewv1.Backup{
+		ObjectMeta: metav1.ObjectMeta{Name: name},
+		Status:     previewv1.BackupStatus{Phase: phase},
+	}
+	if !stopped.IsZero() {
+		b.Status.StoppedAt = &metav1.Time{Time: stopped}
+	}
+	return b
+}
+
+func newTestVerifier(client Client) (*Verifier, *Metrics) {
+	m := NewMetrics()
+	v := NewVerifier(client, journal.New(), m, Config{
+		ScheduledBackupName: "cluster-longhaul",
+		Schedule:            "0 */6 * * *",
+		RetentionDays:       1,
+	})
+	return v, m
+}
+
+var _ = Describe("Verifier.Bootstrap", func() {
+	It("ensures the ScheduledBackup with configured parameters", func() {
+		fc := &fakeBackupClient{}
+		v, _ := newTestVerifier(fc)
+		Expect(v.Bootstrap(context.Background())).To(Succeed())
+		Expect(fc.ensureCalls).To(HaveLen(1))
+		Expect(fc.ensureCalls[0]).To(Equal(ensureArgs{"cluster-longhaul", "0 */6 * * *", 1}))
+	})
+
+	It("wraps ensure errors", func() {
+		fc := &fakeBackupClient{ensureErr: errors.New("boom")}
+		v, _ := newTestVerifier(fc)
+		Expect(v.Bootstrap(context.Background())).To(MatchError(ContainSubstring("boom")))
+	})
+})
+
+var _ = Describe("Verifier.checkOnce", func() {
+	now := time.Date(2026, 7, 8, 12, 0, 0, 0, time.UTC)
+
+	It("counts scheduling advances without double counting", func() {
+		fc := &fakeBackupClient{
+			sb: &previewv1.ScheduledBackup{
+				Status: previewv1.ScheduledBackupStatus{
+					LastScheduledTime: &metav1.Time{Time: now.Add(-time.Hour)},
+				},
+			},
+		}
+		v, m := newTestVerifier(fc)
+
+		v.checkOnce(context.Background(), now)
+		Expect(m.Snapshot().Scheduled).To(Equal(int64(1)))
+
+		// Same lastScheduledTime → no new count.
+		v.checkOnce(context.Background(), now)
+		Expect(m.Snapshot().Scheduled).To(Equal(int64(1)))
+
+		// Advance → counted.
+		fc.sb.Status.LastScheduledTime = &metav1.Time{Time: now.Add(-30 * time.Minute)}
+		v.checkOnce(context.Background(), now)
+		Expect(m.Snapshot().Scheduled).To(Equal(int64(2)))
+	})
+
+	It("counts completed and failed child backups once each", func() {
+		stopped := now.Add(-30 * time.Minute)
+		fc := &fakeBackupClient{
+			sb: &previewv1.ScheduledBackup{},
+			children: []previewv1.Backup{
+				mkBackup("b1", cnpgv1.BackupPhaseCompleted, stopped),
+				mkBackup("b2", previewv1.BackupPhaseSkipped, time.Time{}),
+				mkBackup("b3", cnpgv1.BackupPhaseFailed, time.Time{}),
+			},
+		}
+		v, m := newTestVerifier(fc)
+
+		v.checkOnce(context.Background(), now)
+		v.checkOnce(context.Background(), now) // idempotent
+
+		snap := m.Snapshot()
+		Expect(snap.Completed).To(Equal(int64(1)))
+		Expect(snap.Failed).To(Equal(int64(1))) // skipped is not a failure
+		Expect(snap.LastChildCount).To(Equal(int64(3)))
+		Expect(snap.RetentionLeaks).To(BeZero())
+	})
+
+	It("does not flag a fresh completed backup within its retention window", func() {
+		// retentionDays=1; stopped 1h ago → well within the 1-day window.
+		stopped := now.Add(-time.Hour)
+		fc := &fakeBackupClient{
+			sb:       &previewv1.ScheduledBackup{},
+			children: []previewv1.Backup{mkBackup("b1", cnpgv1.BackupPhaseCompleted, stopped)},
+		}
+		v, m := newTestVerifier(fc)
+		v.checkOnce(context.Background(), now)
+		Expect(m.Snapshot().RetentionLeaks).To(BeZero())
+	})
+
+	It("flags a retention leak when a backup outlives its window, once", func() {
+		// retentionDays=1 (verifier config); stopped 48h ago and still present
+		// → well past the 1-day window + grace.
+		stopped := now.Add(-48 * time.Hour)
+		fc := &fakeBackupClient{
+			sb:       &previewv1.ScheduledBackup{},
+			children: []previewv1.Backup{mkBackup("b1", cnpgv1.BackupPhaseCompleted, stopped)},
+		}
+		v, m := newTestVerifier(fc)
+
+		v.checkOnce(context.Background(), now)
+		v.checkOnce(context.Background(), now) // must not double-count
+		Expect(m.Snapshot().RetentionLeaks).To(Equal(int64(1)))
+	})
+
+	It("does not flag a leak within the GC grace window", func() {
+		// stopped just over 1 day ago but inside the 10m GC grace.
+		stopped := now.Add(-24 * time.Hour).Add(-1 * time.Minute)
+		fc := &fakeBackupClient{
+			sb:       &previewv1.ScheduledBackup{},
+			children: []previewv1.Backup{mkBackup("b1", cnpgv1.BackupPhaseCompleted, stopped)},
+		}
+		v, m := newTestVerifier(fc)
+		v.checkOnce(context.Background(), now)
+		Expect(m.Snapshot().RetentionLeaks).To(BeZero())
+	})
+
+	It("ignores incomplete backups for leak detection", func() {
+		// A never-completed backup has no stoppedAt → not a leak candidate.
+		fc := &fakeBackupClient{
+			sb:       &previewv1.ScheduledBackup{},
+			children: []previewv1.Backup{mkBackup("b1", cnpgv1.BackupPhaseRunning, time.Time{})},
+		}
+		v, m := newTestVerifier(fc)
+		v.checkOnce(context.Background(), now)
+		Expect(m.Snapshot().RetentionLeaks).To(BeZero())
+	})
+
+	It("uses the backup's own retention, not the run config, for the deadline", func() {
+		// Run config is retentionDays=1, but this backup was created under a
+		// 7-day policy (e.g. a run started earlier with different params).
+		// At 2 days old it is well within its own window → not a leak.
+		stopped := now.Add(-48 * time.Hour)
+		b := mkBackup("b1", cnpgv1.BackupPhaseCompleted, stopped)
+		rd := 7
+		b.Spec.RetentionDays = &rd
+		fc := &fakeBackupClient{
+			sb:       &previewv1.ScheduledBackup{},
+			children: []previewv1.Backup{b},
+		}
+		v, m := newTestVerifier(fc)
+		v.checkOnce(context.Background(), now)
+		Expect(m.Snapshot().RetentionLeaks).To(BeZero())
+	})
+
+	It("prunes dedup state for garbage-collected backups without double-counting", func() {
+		stopped := now.Add(-30 * time.Minute)
+		fc := &fakeBackupClient{
+			sb:       &previewv1.ScheduledBackup{},
+			children: []previewv1.Backup{mkBackup("b1", cnpgv1.BackupPhaseCompleted, stopped)},
+		}
+		v, m := newTestVerifier(fc)
+
+		v.checkOnce(context.Background(), now)
+		Expect(v.seenCompleted).To(HaveLen(1))
+		Expect(m.Snapshot().Completed).To(Equal(int64(1)))
+
+		// b1 is garbage-collected and replaced by a new backup b2.
+		fc.children = []previewv1.Backup{mkBackup("b2", cnpgv1.BackupPhaseCompleted, stopped)}
+		v.checkOnce(context.Background(), now)
+
+		// Map stays bounded to the live set; b1 pruned, b2 counted once.
+		Expect(v.seenCompleted).To(HaveLen(1))
+		Expect(m.Snapshot().Completed).To(Equal(int64(2)))
+	})
+
+	It("survives transient read errors without panicking", func() {
+		fc := &fakeBackupClient{
+			sbErr:    errors.New("apiserver throttled"),
+			childErr: errors.New("apiserver throttled"),
+		}
+		v, m := newTestVerifier(fc)
+		v.checkOnce(context.Background(), now)
+		Expect(m.Snapshot().HasRetentionLeak()).To(BeFalse())
+	})
+})
+
+var _ = Describe("Verifier.checkScheduling liveness", func() {
+	now := time.Date(2026, 7, 8, 12, 0, 0, 0, time.UTC)
+
+	It("warns once when scheduling stalls past nextScheduledTime + grace", func() {
+		// nextScheduledTime is overdue by more than livenessGrace and no
+		// backup has been scheduled since → the scheduler appears stalled.
+		overdue := now.Add(-livenessGrace).Add(-time.Minute)
+		fc := &fakeBackupClient{
+			sb: &previewv1.ScheduledBackup{
+				Status: previewv1.ScheduledBackupStatus{
+					NextScheduledTime: &metav1.Time{Time: overdue},
+				},
+			},
+		}
+		v, _ := newTestVerifier(fc)
+
+		v.checkOnce(context.Background(), now)
+		Expect(v.stallWarnedFor).To(Equal(overdue))
+		Expect(stallWarnings(v)).To(Equal(1))
+
+		// Same overdue deadline → warn-once, no duplicate.
+		v.checkOnce(context.Background(), now)
+		Expect(stallWarnings(v)).To(Equal(1))
+
+		// A new (still overdue) nextScheduledTime re-arms the warning.
+		reArmed := now.Add(-livenessGrace).Add(-30 * time.Second)
+		fc.sb.Status.NextScheduledTime = &metav1.Time{Time: reArmed}
+		v.checkOnce(context.Background(), now)
+		Expect(v.stallWarnedFor).To(Equal(reArmed))
+		Expect(stallWarnings(v)).To(Equal(2))
+	})
+
+	It("does not warn when the next scheduled time is still within grace", func() {
+		fc := &fakeBackupClient{
+			sb: &previewv1.ScheduledBackup{
+				Status: previewv1.ScheduledBackupStatus{
+					// 1m overdue < 5m grace.
+					NextScheduledTime: &metav1.Time{Time: now.Add(-time.Minute)},
+				},
+			},
+		}
+		v, _ := newTestVerifier(fc)
+		v.checkOnce(context.Background(), now)
+		Expect(stallWarnings(v)).To(BeZero())
+		Expect(v.stallWarnedFor.IsZero()).To(BeTrue())
+	})
+
+	It("does not warn when a backup was scheduled after the deadline", func() {
+		overdue := now.Add(-livenessGrace).Add(-time.Minute)
+		fc := &fakeBackupClient{
+			sb: &previewv1.ScheduledBackup{
+				Status: previewv1.ScheduledBackupStatus{
+					// A backup fired just after nextScheduledTime → not stalled.
+					LastScheduledTime: &metav1.Time{Time: overdue.Add(time.Second)},
+					NextScheduledTime: &metav1.Time{Time: overdue},
+				},
+			},
+		}
+		v, _ := newTestVerifier(fc)
+		v.checkOnce(context.Background(), now)
+		Expect(stallWarnings(v)).To(BeZero())
+	})
+})
+
+var _ = Describe("Verifier completion-stall oracle", func() {
+	now := time.Date(2026, 7, 8, 12, 0, 0, 0, time.UTC)
+
+	It("flags a stall when backups keep scheduling but never complete", func() {
+		fc := &fakeBackupClient{sb: &previewv1.ScheduledBackup{}}
+		v, m := newTestVerifier(fc)
+
+		// Successive schedules, each with only a failing child backup.
+		for i := 1; i <= completionStallThreshold; i++ {
+			t := now.Add(time.Duration(i) * time.Hour)
+			fc.sb.Status.LastScheduledTime = &metav1.Time{Time: t}
+			fc.children = []previewv1.Backup{mkBackup(fmt.Sprintf("f%d", i), cnpgv1.BackupPhaseFailed, time.Time{})}
+			v.checkOnce(context.Background(), t)
+		}
+
+		snap := m.Snapshot()
+		Expect(snap.Completed).To(BeZero())
+		Expect(snap.MaxScheduledWithoutCompletion).To(Equal(int64(completionStallThreshold)))
+		Expect(snap.HasCompletionStall()).To(BeTrue())
+	})
+
+	It("does not flag a stall when completions keep pace with scheduling", func() {
+		fc := &fakeBackupClient{sb: &previewv1.ScheduledBackup{}}
+		v, m := newTestVerifier(fc)
+
+		for i := 1; i <= completionStallThreshold+3; i++ {
+			t := now.Add(time.Duration(i) * time.Hour)
+			fc.sb.Status.LastScheduledTime = &metav1.Time{Time: t}
+			// The freshly scheduled backup completes in the same cycle.
+			fc.children = []previewv1.Backup{mkBackup(fmt.Sprintf("c%d", i), cnpgv1.BackupPhaseCompleted, t)}
+			v.checkOnce(context.Background(), t)
+		}
+
+		snap := m.Snapshot()
+		Expect(snap.Completed).To(Equal(int64(completionStallThreshold + 3)))
+		Expect(snap.HasCompletionStall()).To(BeFalse())
+	})
+
+	It("resets the running gap once a completion recovers the path", func() {
+		fc := &fakeBackupClient{sb: &previewv1.ScheduledBackup{}}
+		v, m := newTestVerifier(fc)
+
+		// Two schedules fail (gap climbs to 2, still under threshold)...
+		for i := 1; i <= 2; i++ {
+			t := now.Add(time.Duration(i) * time.Hour)
+			fc.sb.Status.LastScheduledTime = &metav1.Time{Time: t}
+			fc.children = []previewv1.Backup{mkBackup(fmt.Sprintf("f%d", i), cnpgv1.BackupPhaseFailed, time.Time{})}
+			v.checkOnce(context.Background(), t)
+		}
+		Expect(m.Snapshot().MaxScheduledWithoutCompletion).To(Equal(int64(2)))
+
+		// ...then one completes, resetting the running gap.
+		t := now.Add(3 * time.Hour)
+		fc.sb.Status.LastScheduledTime = &metav1.Time{Time: t}
+		fc.children = []previewv1.Backup{mkBackup("c1", cnpgv1.BackupPhaseCompleted, t)}
+		v.checkOnce(context.Background(), t)
+
+		Expect(v.scheduledSinceCompleted).To(BeZero())
+		Expect(m.Snapshot().HasCompletionStall()).To(BeFalse())
+	})
+})
+
+// stallWarnings counts scheduling-stall warnings recorded on the verifier's
+// journal.
+func stallWarnings(v *Verifier) int {
+	n := 0
+	for _, e := range v.journal.Events() {
+		if e.Level == journal.LevelWarn && strings.Contains(e.Message, "stalled") {
+			n++
+		}
+	}
+	return n
+}
diff --git a/test/longhaul/cmd/longhaul/main.go b/test/longhaul/cmd/longhaul/main.go
index c7f375bd8..5982e8d84 100644
--- a/test/longhaul/cmd/longhaul/main.go
+++ b/test/longhaul/cmd/longhaul/main.go
@@ -16,6 +16,7 @@ import (
 	"os"
 	"time"
 
+	"github.com/documentdb/documentdb-operator/test/longhaul/backup"
 	"github.com/documentdb/documentdb-operator/test/longhaul/config"
 	"github.com/documentdb/documentdb-operator/test/longhaul/journal"
 	"github.com/documentdb/documentdb-operator/test/longhaul/monitor"
@@ -146,12 +147,33 @@ func run(cfg config.Config) int {
 	scheduler := operations.NewScheduler(ops, healthMon, j, cfg.OpCooldown)
 	go scheduler.Run(ctx)
 
+	// Start data-protection verifier (ScheduledBackup + retention). Runs
+	// concurrently with the scheduler by design — backup is deliberately not
+	// isolated from topology/chaos operations.
+	backupMetrics := backup.NewMetrics()
+	if cfg.BackupEnabled {
+		verifier := backup.NewVerifier(clusterClient, j, backupMetrics, backup.Config{
+			ScheduledBackupName: cfg.ClusterName + "-longhaul",
+			Schedule:            cfg.BackupSchedule,
+			RetentionDays:       cfg.BackupRetentionDays,
+		})
+		if err := verifier.Bootstrap(ctx); err != nil {
+			// A bootstrap failure is not fatal to the whole run — the workload
+			// and other operations still provide signal. Surface it loudly.
+			j.Error("backup", fmt.Sprintf("ScheduledBackup bootstrap failed, backup verification disabled: %v", err))
+		} else {
+			go verifier.Run(ctx)
+		}
+	} else {
+		j.Info("backup", "backup verification disabled via config")
+	}
+
 	// Start metrics sampling goroutine (feeds leak detector).
 	go runMetricsSampling(ctx, clusterClient, leakDetector, j)
 
 	// Start periodic checkpoint reporter.
 	summaryFunc := func() report.Summary {
-		return buildSummary(metrics, leakDetector, scheduler, j)
+		return buildSummary(metrics, backupMetrics, leakDetector, scheduler, j)
 	}
 	reporter := report.NewCheckpointReporter(k8sClientset, cfg.Namespace, cfg.ReportInterval, summaryFunc)
 	go reporter.Run(ctx)
@@ -169,7 +191,7 @@ func run(cfg config.Config) int {
 	// here (before os.Exit) so the authoritative verdict reaches the source
 	// of truth that operators consult — the Run() goroutine cannot do this
 	// reliably because os.Exit can kill it mid-Update.
-	summary := buildSummary(metrics, leakDetector, scheduler, j)
+	summary := buildSummary(metrics, backupMetrics, leakDetector, scheduler, j)
 	markdown := report.GenerateMarkdown(summary)
 	fmt.Println("\n" + markdown)
 	reporter.EmitFinal()
@@ -187,30 +209,43 @@ func run(cfg config.Config) int {
 }
 
 // buildSummary constructs a report.Summary from current state.
-func buildSummary(metrics *workload.Metrics, leakDetector *monitor.LeakDetector, scheduler *operations.Scheduler, j *journal.Journal) report.Summary {
+func buildSummary(metrics *workload.Metrics, backupMetrics *backup.Metrics, leakDetector *monitor.LeakDetector, scheduler *operations.Scheduler, j *journal.Journal) report.Summary {
 	snap := metrics.Snapshot()
+	backupSnap := backupMetrics.Snapshot()
 	leakAnalysis := leakDetector.Analyze()
 
 	result := report.ResultPass
 	failReason := ""
 
-	if snap.HasDataLoss() {
-		result = report.ResultFail
-		failReason = fmt.Sprintf("data loss: %d gaps, %d checksum errors",
-			snap.GapsDetected, snap.ChecksumErrors)
-	}
-	if j.HasPolicyViolation() {
+	appendReason := func(msg string) {
 		result = report.ResultFail
 		if failReason != "" {
 			failReason += "; "
 		}
-		failReason += "outage policy violated"
+		failReason += msg
+	}
+
+	if snap.HasDataLoss() {
+		appendReason(fmt.Sprintf("data loss: %d gaps, %d checksum errors",
+			snap.GapsDetected, snap.ChecksumErrors))
+	}
+	if j.HasPolicyViolation() {
+		appendReason("outage policy violated")
+	}
+	if backupSnap.HasRetentionLeak() {
+		appendReason(fmt.Sprintf("backup retention leak: %d expired backups not collected",
+			backupSnap.RetentionLeaks))
+	}
+	if backupSnap.HasCompletionStall() {
+		appendReason(fmt.Sprintf("backup completion stalled: %d backups scheduled with no completion",
+			backupSnap.MaxScheduledWithoutCompletion))
 	}
 
 	return report.Summary{
 		Result:       result,
 		Duration:     snap.Elapsed,
 		Metrics:      snap,
+		Backup:       backupSnap,
 		LeakAnalysis: leakAnalysis,
 		OpsExecuted:  scheduler.OpsExecuted(),
 		Windows:      j.DisruptionWindows(),
diff --git a/test/longhaul/config/config.go b/test/longhaul/config/config.go
index 5665bcd77..6528b23cc 100644
--- a/test/longhaul/config/config.go
+++ b/test/longhaul/config/config.go
@@ -32,6 +32,11 @@ const (
 	// Observability and reporting.
 	EnvReportInterval = "LONGHAUL_REPORT_INTERVAL"
 
+	// Data protection (ScheduledBackup + retention verification).
+	EnvBackupEnabled       = "LONGHAUL_BACKUP_ENABLED"
+	EnvBackupSchedule      = "LONGHAUL_BACKUP_SCHEDULE"
+	EnvBackupRetentionDays = "LONGHAUL_BACKUP_RETENTION_DAYS"
+
 	// Operational toggles.
 	EnvResetData = "LONGHAUL_RESET_DATA"
 )
@@ -73,6 +78,17 @@ type Config struct {
 	// ReportInterval is how often checkpoint reports are generated.
 	ReportInterval time.Duration
 
+	// BackupEnabled controls whether the ScheduledBackup + retention
+	// verifier runs. Default true.
+	BackupEnabled bool
+
+	// BackupSchedule is the cron expression for the canary ScheduledBackup.
+	BackupSchedule string
+
+	// BackupRetentionDays is the retention window applied to child backups
+	// and used to derive the retention-leak deadline.
+	BackupRetentionDays int
+
 	// ResetData controls whether the workload collection is dropped on startup.
 	// Default false so that pod restarts preserve durability history; opt in
 	// for fresh local/dev iterations.
@@ -93,6 +109,10 @@ func DefaultConfig() Config {
 		MinInstances:    1,
 		MaxInstances:    3,
 		ReportInterval:  1 * time.Hour,
+
+		BackupEnabled:       true,
+		BackupSchedule:      "0 */6 * * *",
+		BackupRetentionDays: 1,
 	}
 }
 
@@ -177,6 +197,22 @@ func LoadFromEnv() (Config, error) {
 		cfg.ReportInterval = d
 	}
 
+	if v := strings.TrimSpace(strings.ToLower(os.Getenv(EnvBackupEnabled))); v != "" {
+		cfg.BackupEnabled = v == "true" || v == "1" || v == "yes"
+	}
+
+	if v := os.Getenv(EnvBackupSchedule); v != "" {
+		cfg.BackupSchedule = v
+	}
+
+	if v := os.Getenv(EnvBackupRetentionDays); v != "" {
+		n, err := strconv.Atoi(v)
+		if err != nil {
+			return cfg, fmt.Errorf("invalid %s=%q: %w", EnvBackupRetentionDays, v, err)
+		}
+		cfg.BackupRetentionDays = n
+	}
+
 	if v := strings.TrimSpace(strings.ToLower(os.Getenv(EnvResetData))); v != "" {
 		cfg.ResetData = v == "true" || v == "1" || v == "yes"
 	}
@@ -213,6 +249,14 @@ func (c *Config) Validate() error {
 	if c.MaxInstances < c.MinInstances {
 		return fmt.Errorf("max instances (%d) must be >= min instances (%d)", c.MaxInstances, c.MinInstances)
 	}
+	if c.BackupEnabled {
+		if c.BackupSchedule == "" {
+			return fmt.Errorf("backup schedule must not be empty when backups are enabled")
+		}
+		if c.BackupRetentionDays < 1 {
+			return fmt.Errorf("backup retention days must be at least 1, got %d", c.BackupRetentionDays)
+		}
+	}
 	return nil
 }
 
diff --git a/test/longhaul/config/config_test.go b/test/longhaul/config/config_test.go
index 506803f52..51669b5ed 100644
--- a/test/longhaul/config/config_test.go
+++ b/test/longhaul/config/config_test.go
@@ -35,6 +35,7 @@ var _ = Describe("Config", func() {
 				EnvDocumentDBURI, EnvNumWriters,
 				EnvOpCooldown, EnvRecoveryTimeout, EnvSteadyStateWait,
 				EnvMinInstances, EnvMaxInstances, EnvReportInterval,
+				EnvBackupEnabled, EnvBackupSchedule, EnvBackupRetentionDays,
 			} {
 				GinkgoT().Setenv(k, "")
 			}
@@ -103,6 +104,24 @@ var _ = Describe("Config", func() {
 			Expect(err).NotTo(HaveOccurred())
 			Expect(cfg.DocumentDBURI).To(Equal("mongodb://localhost:27017"))
 		})
+
+		It("parses the backup env knobs", func() {
+			GinkgoT().Setenv(EnvBackupEnabled, "true")
+			GinkgoT().Setenv(EnvBackupSchedule, "0 */6 * * *")
+			GinkgoT().Setenv(EnvBackupRetentionDays, "7")
+			cfg, err := LoadFromEnv()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(cfg.BackupEnabled).To(BeTrue())
+			Expect(cfg.BackupSchedule).To(Equal("0 */6 * * *"))
+			Expect(cfg.BackupRetentionDays).To(Equal(7))
+		})
+
+		It("returns error for invalid BackupRetentionDays", func() {
+			GinkgoT().Setenv(EnvBackupRetentionDays, "abc")
+			_, err := LoadFromEnv()
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring(EnvBackupRetentionDays))
+		})
 	})
 
 	Describe("Validate", func() {
@@ -160,6 +179,29 @@ var _ = Describe("Config", func() {
 			cfg.MaxInstances = 4
 			Expect(cfg.Validate()).To(MatchError(ContainSubstring("must not exceed 3")))
 		})
+
+		It("fails when backups enabled but schedule is empty", func() {
+			cfg := DefaultConfig()
+			cfg.ClusterName = "test"
+			cfg.BackupSchedule = ""
+			Expect(cfg.Validate()).To(MatchError(ContainSubstring("backup schedule")))
+		})
+
+		It("fails when backup retention days is below 1", func() {
+			cfg := DefaultConfig()
+			cfg.ClusterName = "test"
+			cfg.BackupRetentionDays = 0
+			Expect(cfg.Validate()).To(MatchError(ContainSubstring("backup retention days")))
+		})
+
+		It("skips backup validation when backups disabled", func() {
+			cfg := DefaultConfig()
+			cfg.ClusterName = "test"
+			cfg.BackupEnabled = false
+			cfg.BackupSchedule = ""
+			cfg.BackupRetentionDays = 0
+			Expect(cfg.Validate()).To(Succeed())
+		})
 	})
 
 	Describe("IsEnabled", func() {
diff --git a/test/longhaul/deploy/rbac.yaml b/test/longhaul/deploy/rbac.yaml
index f5ed5abcd..0f4395546 100644
--- a/test/longhaul/deploy/rbac.yaml
+++ b/test/longhaul/deploy/rbac.yaml
@@ -29,6 +29,14 @@ rules:
   - apiGroups: ["documentdb.io"]
     resources: ["dbs"]
     verbs: ["get", "list", "patch"]
+  # Manage ScheduledBackups and read their child Backups for the data-protection
+  # verifier (ensure a ScheduledBackup, then watch child Backup CRs).
+  - apiGroups: ["documentdb.io"]
+    resources: ["scheduledbackups"]
+    verbs: ["create", "get", "list", "update"]
+  - apiGroups: ["documentdb.io"]
+    resources: ["backups"]
+    verbs: ["list"]
   # Create/update ConfigMaps for periodic report persistence.
   - apiGroups: [""]
     resources: ["configmaps"]
diff --git a/test/longhaul/deploy/setup.yaml b/test/longhaul/deploy/setup.yaml
index e05115567..8f3f4fe81 100644
--- a/test/longhaul/deploy/setup.yaml
+++ b/test/longhaul/deploy/setup.yaml
@@ -7,7 +7,7 @@
 # It creates the target namespace, credentials secret, and DocumentDB CR.
 #
 # Prerequisites:
-#   1. AKS cluster with k8s >= 1.30
+#   1. AKS cluster with k8s >= 1.35 (operator requires ImageVolume, GA in 1.35)
 #   2. cert-manager installed:
 #      helm repo add jetstack https://charts.jetstack.io && helm repo update
 #      helm install cert-manager jetstack/cert-manager --namespace cert-manager \
diff --git a/test/longhaul/monitor/k8sclient.go b/test/longhaul/monitor/k8sclient.go
index b51ffdd82..f843f960b 100644
--- a/test/longhaul/monitor/k8sclient.go
+++ b/test/longhaul/monitor/k8sclient.go
@@ -11,6 +11,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
@@ -267,3 +269,89 @@ func (k *K8sClusterClient) GetPodMetrics(ctx context.Context) ([]PodMetrics, err
 func (k *K8sClusterClient) MetricsAvailable() bool {
 	return k.metricsAvail.Load()
 }
+
+// scheduledBackupLabel is the label the ScheduledBackup reconciler stamps
+// on every child Backup CR it creates (see
+// preview.ScheduledBackup.CreateBackup). Selecting on it isolates the
+// backups produced by our canary ScheduledBackup from any ad-hoc Backup
+// objects that might exist in the namespace.
+const scheduledBackupLabel = "scheduledbackup"
+
+// EnsureScheduledBackup creates the ScheduledBackup CR for the target
+// cluster, or reconciles an existing one so a run started with a different
+// schedule/retention actually takes effect. Existing child backups and their
+// accumulated history are preserved — the CR is updated in place, never
+// recreated — so only backups created after the update pick up the new spec.
+func (k *K8sClusterClient) EnsureScheduledBackup(ctx context.Context, name, schedule string, retentionDays int) error {
+	existing := &previewv1.ScheduledBackup{}
+	err := k.crClient.Get(ctx, types.NamespacedName{Namespace: k.namespace, Name: name}, existing)
+	if err == nil {
+		return k.reconcileScheduledBackup(ctx, existing, schedule, retentionDays)
+	}
+	if !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to get ScheduledBackup %s/%s: %w", k.namespace, name, err)
+	}
+
+	rd := retentionDays
+	sb := &previewv1.ScheduledBackup{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: k.namespace,
+		},
+		Spec: previewv1.ScheduledBackupSpec{
+			Cluster:       cnpgv1.LocalObjectReference{Name: k.clusterName},
+			Schedule:      schedule,
+			RetentionDays: &rd,
+		},
+	}
+	if err := k.crClient.Create(ctx, sb); err != nil {
+		return fmt.Errorf("failed to create ScheduledBackup %s/%s: %w", k.namespace, name, err)
+	}
+	return nil
+}
+
+// reconcileScheduledBackup updates an existing ScheduledBackup in place when
+// its schedule or retention differs from what this run requested.
+func (k *K8sClusterClient) reconcileScheduledBackup(ctx context.Context, existing *previewv1.ScheduledBackup, schedule string, retentionDays int) error {
+	curRetention := 0
+	if existing.Spec.RetentionDays != nil {
+		curRetention = *existing.Spec.RetentionDays
+	}
+	if existing.Spec.Schedule == schedule && curRetention == retentionDays {
+		return nil
+	}
+
+	log.Printf("[k8sclient] ScheduledBackup %s/%s spec drift; updating (schedule %q->%q, retentionDays %d->%d)",
+		k.namespace, existing.Name, existing.Spec.Schedule, schedule, curRetention, retentionDays)
+
+	rd := retentionDays
+	existing.Spec.Schedule = schedule
+	existing.Spec.RetentionDays = &rd
+	if err := k.crClient.Update(ctx, existing); err != nil {
+		return fmt.Errorf("failed to update ScheduledBackup %s/%s: %w", k.namespace, existing.Name, err)
+	}
+	return nil
+}
+
+// GetScheduledBackup fetches the ScheduledBackup CR by name in the target
+// namespace.
+func (k *K8sClusterClient) GetScheduledBackup(ctx context.Context, name string) (*previewv1.ScheduledBackup, error) {
+	sb := &previewv1.ScheduledBackup{}
+	if err := k.crClient.Get(ctx, types.NamespacedName{Namespace: k.namespace, Name: name}, sb); err != nil {
+		return nil, fmt.Errorf("failed to get ScheduledBackup %s/%s: %w", k.namespace, name, err)
+	}
+	return sb, nil
+}
+
+// ListScheduledChildBackups returns every Backup CR that the named
+// ScheduledBackup produced, identified by the "scheduledbackup" label.
+func (k *K8sClusterClient) ListScheduledChildBackups(ctx context.Context, scheduledBackupName string) ([]previewv1.Backup, error) {
+	var list previewv1.BackupList
+	if err := k.crClient.List(ctx, &list,
+		ctrlclient.InNamespace(k.namespace),
+		ctrlclient.MatchingLabels{scheduledBackupLabel: scheduledBackupName},
+	); err != nil {
+		return nil, fmt.Errorf("failed to list child Backups for ScheduledBackup %s: %w", scheduledBackupName, err)
+	}
+	return list.Items, nil
+}
diff --git a/test/longhaul/report/report.go b/test/longhaul/report/report.go
index e7316a7f7..2aef7661f 100644
--- a/test/longhaul/report/report.go
+++ b/test/longhaul/report/report.go
@@ -8,6 +8,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/documentdb/documentdb-operator/test/longhaul/backup"
 	"github.com/documentdb/documentdb-operator/test/longhaul/journal"
 	"github.com/documentdb/documentdb-operator/test/longhaul/monitor"
 	"github.com/documentdb/documentdb-operator/test/longhaul/workload"
@@ -38,6 +39,11 @@ type Summary struct {
 	// failed, verify passes, gaps, checksum errors).
 	Metrics workload.MetricsSnapshot
 
+	// Backup is a snapshot of the data-protection counters (backups
+	// scheduled/completed/failed, live population, retention leaks). A
+	// retention leak flips Result to FAIL.
+	Backup backup.MetricsSnapshot
+
 	// LeakAnalysis is the operator-pod resource trend (memory/CPU slope over
 	// the run); LeakAnalysis.HasLeak being true does NOT flip Result — it
 	// only emits a warning annotation.
@@ -90,6 +96,21 @@ func GenerateMarkdown(s Summary) string {
 	fmt.Fprintf(&b, "| Checksum Errors | %d |\n", s.Metrics.ChecksumErrors)
 	b.WriteString("\n")
 
+	// Data Protection (ScheduledBackup + retention)
+	b.WriteString("## Data Protection\n\n")
+	b.WriteString("| Metric | Value |\n")
+	b.WriteString("|--------|-------|\n")
+	fmt.Fprintf(&b, "| Backups Scheduled | %d |\n", s.Backup.Scheduled)
+	fmt.Fprintf(&b, "| Backups Completed | %d |\n", s.Backup.Completed)
+	fmt.Fprintf(&b, "| Backups Failed | %d |\n", s.Backup.Failed)
+	fmt.Fprintf(&b, "| Live Backup Count | %d |\n", s.Backup.LastChildCount)
+	fmt.Fprintf(&b, "| Retention Leaks | %d |\n", s.Backup.RetentionLeaks)
+	fmt.Fprintf(&b, "| Max Scheduled Without Completion | %d |\n", s.Backup.MaxScheduledWithoutCompletion)
+	if !s.Backup.LastScheduled.IsZero() {
+		fmt.Fprintf(&b, "| Last Scheduled | %s |\n", s.Backup.LastScheduled.Format(time.RFC3339))
+	}
+	b.WriteString("\n")
+
 	// Disruption Windows
 	if len(s.Windows) > 0 {
 		b.WriteString("## Disruption Windows\n\n")