diff --git a/test/longhaul/README.md b/test/longhaul/README.md
index d87f153f..2333b562 100644
--- a/test/longhaul/README.md
+++ b/test/longhaul/README.md
@@ -133,6 +133,7 @@ All configuration is via environment variables.
 | `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_RESET_DATA` | No | `false` | If `true`, drop the workload collection on startup. Off by default so a Deployment pod restart preserves durability history. |
+| `LONGHAUL_RETAIN_PER_WRITER` | No | `2000000` | Retention window: most-recent verified documents kept per writer before the pruner deletes older ones, bounding disk usage. `0` disables pruning (unbounded growth). |
 
 ## CI Safety
 
diff --git a/test/longhaul/cmd/longhaul/main.go b/test/longhaul/cmd/longhaul/main.go
index c7f375bd..54401385 100644
--- a/test/longhaul/cmd/longhaul/main.go
+++ b/test/longhaul/cmd/longhaul/main.go
@@ -132,9 +132,20 @@ func run(cfg config.Config) int {
 	// Start verifier. A single verifier is sufficient — see StartVerifier
 	// godoc. Writers are passed so the verifier can detect tail loss by
 	// comparing each writer's acked tip against what's in the DB.
-	workload.StartVerifier(ctx, db, writers, metrics, j)
+	verifier := workload.StartVerifier(ctx, db, writers, metrics, j)
 	j.Info("main", "verifier started")
 
+	// Start retention pruner (bounds the workload collection so an unbounded
+	// write test cannot eventually exhaust the PVC). It prunes only documents
+	// below the verifier's confirmed floor, so it never affects the durability
+	// verdict. Disabled when RetainPerWriter == 0.
+	if cfg.RetainPerWriter > 0 {
+		workload.StartPruner(ctx, db.Collection(workload.CollectionName), writers, verifier, cfg.RetainPerWriter, metrics, j)
+		j.Info("main", "retention pruner started")
+	} else {
+		j.Info("main", "retention pruning disabled (LONGHAUL_RETAIN_PER_WRITER=0)")
+	}
+
 	// Configure operations.
 	ops := []operations.Operation{
 		operations.NewScaleUp(clusterClient, healthMon, cfg.MaxInstances, cfg.RecoveryTimeout),
diff --git a/test/longhaul/config/config.go b/test/longhaul/config/config.go
index 5665bcd7..90ef1711 100644
--- a/test/longhaul/config/config.go
+++ b/test/longhaul/config/config.go
@@ -34,8 +34,17 @@ const (
 
 	// Operational toggles.
 	EnvResetData = "LONGHAUL_RESET_DATA"
+
+	// Data retention. Bounds the workload collection so an unbounded write
+	// test does not eventually exhaust the PVC.
+	EnvRetainPerWriter = "LONGHAUL_RETAIN_PER_WRITER"
 )
 
+// DefaultRetainPerWriter is the default number of most-recent documents kept
+// per writer when pruning is enabled. At ~10 writes/sec/writer this retains
+// roughly 55 hours of history per writer while bounding steady-state disk use.
+const DefaultRetainPerWriter = 2_000_000
+
 // Config holds all configuration for a long haul test run.
 type Config struct {
 	// MaxDuration is the maximum test duration. Zero means run until failure.
@@ -77,6 +86,11 @@ type Config struct {
 	// Default false so that pod restarts preserve durability history; opt in
 	// for fresh local/dev iterations.
 	ResetData bool
+
+	// RetainPerWriter is the number of most-recent documents kept per writer.
+	// Older, already-verified documents are pruned to bound disk usage. Zero
+	// disables pruning (unbounded growth — the pre-retention behavior).
+	RetainPerWriter int64
 }
 
 // DefaultConfig returns a Config with safe defaults for local development.
@@ -93,6 +107,7 @@ func DefaultConfig() Config {
 		MinInstances:    1,
 		MaxInstances:    3,
 		ReportInterval:  1 * time.Hour,
+		RetainPerWriter: DefaultRetainPerWriter,
 	}
 }
 
@@ -181,6 +196,14 @@ func LoadFromEnv() (Config, error) {
 		cfg.ResetData = v == "true" || v == "1" || v == "yes"
 	}
 
+	if v := os.Getenv(EnvRetainPerWriter); v != "" {
+		n, err := strconv.ParseInt(v, 10, 64)
+		if err != nil {
+			return cfg, fmt.Errorf("invalid %s=%q: %w", EnvRetainPerWriter, v, err)
+		}
+		cfg.RetainPerWriter = n
+	}
+
 	return cfg, nil
 }
 
@@ -213,6 +236,9 @@ 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.RetainPerWriter < 0 {
+		return fmt.Errorf("retain per writer must not be negative, got %d", c.RetainPerWriter)
+	}
 	return nil
 }
 
diff --git a/test/longhaul/config/config_test.go b/test/longhaul/config/config_test.go
index 506803f5..dcf920a2 100644
--- a/test/longhaul/config/config_test.go
+++ b/test/longhaul/config/config_test.go
@@ -23,6 +23,7 @@ var _ = Describe("Config", func() {
 			Expect(cfg.SteadyStateWait).To(Equal(60 * time.Second))
 			Expect(cfg.MinInstances).To(Equal(1))
 			Expect(cfg.MaxInstances).To(Equal(3))
+			Expect(cfg.RetainPerWriter).To(Equal(int64(DefaultRetainPerWriter)))
 		})
 	})
 
@@ -35,6 +36,7 @@ var _ = Describe("Config", func() {
 				EnvDocumentDBURI, EnvNumWriters,
 				EnvOpCooldown, EnvRecoveryTimeout, EnvSteadyStateWait,
 				EnvMinInstances, EnvMaxInstances, EnvReportInterval,
+				EnvResetData, EnvRetainPerWriter,
 			} {
 				GinkgoT().Setenv(k, "")
 			}
@@ -103,6 +105,26 @@ var _ = Describe("Config", func() {
 			Expect(err).NotTo(HaveOccurred())
 			Expect(cfg.DocumentDBURI).To(Equal("mongodb://localhost:27017"))
 		})
+
+		It("parses RetainPerWriter from env", func() {
+			GinkgoT().Setenv(EnvRetainPerWriter, "500000")
+			cfg, err := LoadFromEnv()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(cfg.RetainPerWriter).To(Equal(int64(500_000)))
+		})
+
+		It("parses RetainPerWriter=0 to disable pruning", func() {
+			GinkgoT().Setenv(EnvRetainPerWriter, "0")
+			cfg, err := LoadFromEnv()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(cfg.RetainPerWriter).To(BeZero())
+		})
+
+		It("returns error for invalid RetainPerWriter", func() {
+			GinkgoT().Setenv(EnvRetainPerWriter, "not-a-number")
+			_, err := LoadFromEnv()
+			Expect(err).To(MatchError(ContainSubstring(EnvRetainPerWriter)))
+		})
 	})
 
 	Describe("Validate", func() {
@@ -160,6 +182,13 @@ var _ = Describe("Config", func() {
 			cfg.MaxInstances = 4
 			Expect(cfg.Validate()).To(MatchError(ContainSubstring("must not exceed 3")))
 		})
+
+		It("fails when RetainPerWriter is negative", func() {
+			cfg := DefaultConfig()
+			cfg.ClusterName = "test"
+			cfg.RetainPerWriter = -1
+			Expect(cfg.Validate()).To(MatchError(ContainSubstring("retain per writer")))
+		})
 	})
 
 	Describe("IsEnabled", func() {
diff --git a/test/longhaul/deploy/deployment.yaml b/test/longhaul/deploy/deployment.yaml
index 49d5c09f..835e577f 100644
--- a/test/longhaul/deploy/deployment.yaml
+++ b/test/longhaul/deploy/deployment.yaml
@@ -59,6 +59,12 @@ data:
   # DocumentDB workload collection and longhaul-report ConfigMap checkpoints.
   # Override to "true" only for local/dev iterations.
   LONGHAUL_RESET_DATA: "false"
+  # Retention: keep only the most recent N documents per writer so the endurance
+  # write load does not grow the PVC without bound (a fixed disk always fills
+  # eventually otherwise). Pruning removes only already-verified history, so it
+  # never affects the durability verdict. At ~10 writes/sec/writer, 2,000,000
+  # docs is roughly 55h of per-writer history. Set to "0" to disable pruning.
+  LONGHAUL_RETAIN_PER_WRITER: "2000000"
   # Replica range for scale operations.
   # Bounds for spec.instancesPerNode (CRD allows 1-3). Default Min=1, Max=3
   # exercises full scale-up/down cycle. Set Min=Max to disable scale ops.
diff --git a/test/longhaul/workload/pruner.go b/test/longhaul/workload/pruner.go
new file mode 100644
index 00000000..71fee7d0
--- /dev/null
+++ b/test/longhaul/workload/pruner.go
@@ -0,0 +1,158 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package workload
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/documentdb/documentdb-operator/test/longhaul/journal"
+
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+)
+
+const (
+	// pruneInterval is how often the pruner trims old documents. A long-haul
+	// run writes ~10 docs/sec/writer, so a few thousand rows accumulate per
+	// writer between cycles — a small, index-backed DeleteMany each time.
+	pruneInterval = 5 * time.Minute
+)
+
+// floorProvider reports the highest fully-verified seq per writer. *Verifier
+// satisfies this; the pruner only ever deletes strictly below the floor, so it
+// can never remove a document the verifier has not already accounted for.
+type floorProvider interface {
+	ConfirmedFloor(writerID string) int64
+}
+
+// pruneBackend abstracts the delete so Pruner can be unit-tested without a
+// live collection. Production uses docdbPruneBackend.
+type pruneBackend interface {
+	// deleteThrough removes all documents for writerID with seq <= throughSeq
+	// and returns the number deleted.
+	deleteThrough(ctx context.Context, writerID string, throughSeq int64) (int64, error)
+}
+
+// docdbPruneBackend adapts *mongo.Collection to pruneBackend. The delete filter
+// rides the existing unique (writer_id, seq) index, so it is a bounded range
+// delete rather than a collection scan.
+type docdbPruneBackend struct {
+	coll *mongo.Collection
+}
+
+func (m docdbPruneBackend) deleteThrough(ctx context.Context, writerID string, throughSeq int64) (int64, error) {
+	res, err := m.coll.DeleteMany(ctx, bson.D{
+		{Key: "writer_id", Value: writerID},
+		{Key: "seq", Value: bson.D{{Key: "$lte", Value: throughSeq}}},
+	})
+	if err != nil {
+		return 0, err
+	}
+	return res.DeletedCount, nil
+}
+
+// Pruner bounds the workload collection by keeping only the most recent
+// retainPerWriter documents per writer. It deletes strictly below the
+// verifier's confirmed floor, so pruning never affects the durability verdict:
+// every removed document was already scanned, and the verifier's startup
+// DB-min seed keeps a post-restart scan from misreading the pruned prefix as a
+// gap.
+//
+// Pruning freezes permanently once the verifier reports any data loss (gap or
+// checksum mismatch): the run is already FAIL, so the collection is left intact
+// to preserve the corrupt document and its neighbours for post-mortem debugging.
+type Pruner struct {
+	writers         []*Writer
+	floor           floorProvider
+	backend         pruneBackend
+	retainPerWriter int64
+	metrics         *Metrics
+	journal         *journal.Journal
+
+	// frozen is set once a data-loss failure freezes pruning, so the
+	// "pruning frozen" warning is logged exactly once. Only touched from the
+	// pruner goroutine.
+	frozen bool
+}
+
+// NewPruner constructs a Pruner. retainPerWriter must be > 0; callers gate on
+// that (0 disables pruning entirely) before constructing.
+func NewPruner(coll *mongo.Collection, writers []*Writer, floor floorProvider, retainPerWriter int64, metrics *Metrics, j *journal.Journal) *Pruner {
+	return &Pruner{
+		writers:         writers,
+		floor:           floor,
+		backend:         docdbPruneBackend{coll: coll},
+		retainPerWriter: retainPerWriter,
+		metrics:         metrics,
+		journal:         j,
+	}
+}
+
+// Run starts the prune loop. It blocks until the context is cancelled.
+func (p *Pruner) Run(ctx context.Context) {
+	p.journal.Info("pruner", fmt.Sprintf("pruner started (retain %d docs/writer)", p.retainPerWriter))
+	defer p.journal.Info("pruner", "pruner stopped")
+
+	ticker := time.NewTicker(pruneInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			p.pruneAll(ctx)
+		}
+	}
+}
+
+func (p *Pruner) pruneAll(ctx context.Context) {
+	// Freeze pruning the moment the durability oracle detects a failure. Once a
+	// gap or checksum mismatch is seen the run is already FAIL, so bounding disk
+	// no longer matters — instead we preserve the collection exactly as it was
+	// at the failure so the corrupt document (and its neighbours) stay on disk
+	// for post-mortem debugging. HasDataLoss() is monotonic (the counters never
+	// reset within a run), so once frozen the pruner stays frozen.
+	if p.metrics.Snapshot().HasDataLoss() {
+		if !p.frozen {
+			p.frozen = true
+			p.journal.Warn("pruner", "data loss detected; pruning frozen to preserve evidence for debugging")
+		}
+		return
+	}
+	for _, w := range p.writers {
+		p.pruneWriter(ctx, w.id)
+	}
+}
+
+// pruneWriter deletes the writer's documents older than the retention window.
+// The delete boundary is (confirmed floor - retainPerWriter): everything at or
+// below it is both fully verified and outside the retained tail, so removing it
+// is safe and bounds the writer's footprint to ~retainPerWriter documents.
+func (p *Pruner) pruneWriter(ctx context.Context, writerID string) {
+	floor := p.floor.ConfirmedFloor(writerID)
+	throughSeq := floor - p.retainPerWriter
+	if throughSeq < 1 {
+		// Not enough verified history yet to prune anything.
+		return
+	}
+
+	deleted, err := p.backend.deleteThrough(ctx, writerID, throughSeq)
+	if err != nil {
+		p.journal.Warn("pruner", fmt.Sprintf("prune failed for writer %s: %v", writerID, err))
+		return
+	}
+	if deleted > 0 {
+		p.journal.Info("pruner", fmt.Sprintf("pruned %d docs for writer %s (seq <= %d)", deleted, writerID, throughSeq))
+	}
+}
+
+// StartPruner launches a single pruner goroutine and returns it.
+func StartPruner(ctx context.Context, coll *mongo.Collection, writers []*Writer, floor floorProvider, retainPerWriter int64, metrics *Metrics, j *journal.Journal) *Pruner {
+	p := NewPruner(coll, writers, floor, retainPerWriter, metrics, j)
+	go p.Run(ctx)
+	return p
+}
diff --git a/test/longhaul/workload/pruner_test.go b/test/longhaul/workload/pruner_test.go
new file mode 100644
index 00000000..9eb9bc68
--- /dev/null
+++ b/test/longhaul/workload/pruner_test.go
@@ -0,0 +1,176 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package workload
+
+import (
+	"context"
+	"errors"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/documentdb/documentdb-operator/test/longhaul/journal"
+)
+
+// fakeFloor is a controllable floorProvider.
+type fakeFloor map[string]int64
+
+func (f fakeFloor) ConfirmedFloor(writerID string) int64 { return f[writerID] }
+
+// fakePruneBackend records deleteThrough calls and returns scripted results.
+type fakePruneBackend struct {
+	// deletedThrough[writerID] is the highest throughSeq requested.
+	calls []pruneCall
+	// perWriterDeleted maps writerID -> count returned by deleteThrough.
+	perWriterDeleted map[string]int64
+	err              error
+}
+
+type pruneCall struct {
+	writerID   string
+	throughSeq int64
+}
+
+func (f *fakePruneBackend) deleteThrough(_ context.Context, writerID string, throughSeq int64) (int64, error) {
+	f.calls = append(f.calls, pruneCall{writerID: writerID, throughSeq: throughSeq})
+	if f.err != nil {
+		return 0, f.err
+	}
+	if f.perWriterDeleted != nil {
+		return f.perWriterDeleted[writerID], nil
+	}
+	return 0, nil
+}
+
+func newTestPruner(b pruneBackend, floor floorProvider, retain int64, m *Metrics) *Pruner {
+	return &Pruner{
+		writers:         []*Writer{{id: "w000"}, {id: "w001"}},
+		floor:           floor,
+		backend:         b,
+		retainPerWriter: retain,
+		metrics:         m,
+		journal:         journal.New(),
+	}
+}
+
+var _ = Describe("Pruner.pruneWriter", func() {
+	It("deletes through (floor - retainPerWriter) when there is enough history", func() {
+		b := &fakePruneBackend{perWriterDeleted: map[string]int64{"w000": 500}}
+		m := NewMetrics()
+		p := newTestPruner(b, fakeFloor{"w000": 10_000}, 2_000, m)
+
+		p.pruneWriter(context.Background(), "w000")
+
+		Expect(b.calls).To(HaveLen(1))
+		Expect(b.calls[0]).To(Equal(pruneCall{writerID: "w000", throughSeq: 8_000}))
+	})
+
+	It("does not delete when the floor is below the retention window", func() {
+		b := &fakePruneBackend{}
+		m := NewMetrics()
+		// floor 1500, retain 2000 => throughSeq = -500 < 1 => no delete.
+		p := newTestPruner(b, fakeFloor{"w000": 1_500}, 2_000, m)
+
+		p.pruneWriter(context.Background(), "w000")
+
+		Expect(b.calls).To(BeEmpty())
+	})
+
+	It("does not delete when the floor is exactly at the retention window", func() {
+		b := &fakePruneBackend{}
+		m := NewMetrics()
+		// floor 2000, retain 2000 => throughSeq = 0 < 1 => no delete.
+		p := newTestPruner(b, fakeFloor{"w000": 2_000}, 2_000, m)
+
+		p.pruneWriter(context.Background(), "w000")
+
+		Expect(b.calls).To(BeEmpty())
+	})
+
+	It("prunes the first document once the floor advances one past the window", func() {
+		b := &fakePruneBackend{perWriterDeleted: map[string]int64{"w000": 1}}
+		m := NewMetrics()
+		// floor 2001, retain 2000 => throughSeq = 1 => delete seq <= 1.
+		p := newTestPruner(b, fakeFloor{"w000": 2_001}, 2_000, m)
+
+		p.pruneWriter(context.Background(), "w000")
+
+		Expect(b.calls).To(HaveLen(1))
+		Expect(b.calls[0].throughSeq).To(Equal(int64(1)))
+	})
+
+	It("swallows backend errors without issuing further deletes", func() {
+		b := &fakePruneBackend{err: errors.New("delete failed")}
+		m := NewMetrics()
+		p := newTestPruner(b, fakeFloor{"w000": 10_000}, 2_000, m)
+
+		p.pruneWriter(context.Background(), "w000")
+
+		Expect(b.calls).To(HaveLen(1))
+	})
+
+	It("never deletes at or above the confirmed floor (safety invariant)", func() {
+		b := &fakePruneBackend{perWriterDeleted: map[string]int64{"w000": 3}}
+		m := NewMetrics()
+		floor := int64(5_000)
+		p := newTestPruner(b, fakeFloor{"w000": floor}, 1_000, m)
+
+		p.pruneWriter(context.Background(), "w000")
+
+		Expect(b.calls).To(HaveLen(1))
+		Expect(b.calls[0].throughSeq).To(BeNumerically("<", floor),
+			"pruner must only delete strictly below the verifier's confirmed floor")
+	})
+})
+
+var _ = Describe("Pruner.pruneAll", func() {
+	It("prunes every writer using its own floor", func() {
+		b := &fakePruneBackend{perWriterDeleted: map[string]int64{"w000": 10, "w001": 20}}
+		m := NewMetrics()
+		p := newTestPruner(b, fakeFloor{"w000": 5_000, "w001": 9_000}, 1_000, m)
+
+		p.pruneAll(context.Background())
+
+		Expect(b.calls).To(ConsistOf(
+			pruneCall{writerID: "w000", throughSeq: 4_000},
+			pruneCall{writerID: "w001", throughSeq: 8_000},
+		))
+	})
+
+	It("freezes pruning when a checksum error has been detected (preserve evidence)", func() {
+		b := &fakePruneBackend{perWriterDeleted: map[string]int64{"w000": 10, "w001": 20}}
+		m := NewMetrics()
+		p := newTestPruner(b, fakeFloor{"w000": 5_000, "w001": 9_000}, 1_000, m)
+		m.ChecksumErrors.Add(1)
+
+		p.pruneAll(context.Background())
+
+		Expect(b.calls).To(BeEmpty(), "no deletes must be issued once data loss is detected")
+		Expect(p.frozen).To(BeTrue())
+	})
+
+	It("freezes pruning when a gap has been detected", func() {
+		b := &fakePruneBackend{perWriterDeleted: map[string]int64{"w000": 10}}
+		m := NewMetrics()
+		p := newTestPruner(b, fakeFloor{"w000": 5_000}, 1_000, m)
+		m.VerifyGapsDetected.Add(1)
+
+		p.pruneAll(context.Background())
+
+		Expect(b.calls).To(BeEmpty())
+	})
+
+	It("stays frozen on later cycles even though the failure counters are unchanged", func() {
+		b := &fakePruneBackend{perWriterDeleted: map[string]int64{"w000": 10}}
+		m := NewMetrics()
+		p := newTestPruner(b, fakeFloor{"w000": 5_000}, 1_000, m)
+		m.ChecksumErrors.Add(1)
+
+		p.pruneAll(context.Background())
+		p.pruneAll(context.Background())
+
+		Expect(b.calls).To(BeEmpty())
+		Expect(p.frozen).To(BeTrue())
+	})
+})
diff --git a/test/longhaul/workload/verifier.go b/test/longhaul/workload/verifier.go
index 10addf7d..9f19dc09 100644
--- a/test/longhaul/workload/verifier.go
+++ b/test/longhaul/workload/verifier.go
@@ -5,7 +5,9 @@ package workload
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"sync/atomic"
 	"time"
 
 	"github.com/documentdb/documentdb-operator/test/longhaul/journal"
@@ -41,6 +43,17 @@ type Verifier struct {
 	// a late-arriving fill at a missing seq is not re-checked.
 	// Only mutated from the verifier goroutine, so no lock is needed.
 	nextSeq map[string]int64
+
+	// seeded[writerID] records whether the verifier has performed the one-time
+	// startup seed of expectedSeq from the collection's current minimum seq.
+	// Only touched from the verifier goroutine.
+	seeded map[string]bool
+
+	// floor[writerID] is the highest fully-verified seq for that writer
+	// (nextSeq-1), published as an atomic so the retention pruner can read it
+	// concurrently without racing the verifier goroutine. Everything at or
+	// below the floor has been accounted for and is safe to prune.
+	floor map[string]*atomic.Int64
 }
 
 // NewVerifier creates a verifier. writers is the set of writers whose tips
@@ -49,13 +62,31 @@ type Verifier struct {
 func NewVerifier(db *mongo.Database, writers []*Writer, metrics *Metrics, j *journal.Journal) *Verifier {
 	coll := db.Collection(CollectionName, options.Collection().
 		SetReadConcern(readconcern.Majority()))
+	floor := make(map[string]*atomic.Int64, len(writers))
+	for _, w := range writers {
+		floor[w.id] = &atomic.Int64{}
+	}
 	return &Verifier{
 		metrics:    metrics,
 		journal:    j,
 		collection: coll,
 		writers:    writers,
 		nextSeq:    make(map[string]int64),
+		seeded:     make(map[string]bool),
+		floor:      floor,
+	}
+}
+
+// ConfirmedFloor returns the highest seq the verifier has fully accounted for
+// for writerID (0 if none yet). Every seq at or below this value has been
+// scanned, so a pruner may safely delete documents below it without the
+// verifier ever re-reading them (steady state) or misreading them as gaps
+// (restart, thanks to the startup DB-min seed). Safe for concurrent reads.
+func (v *Verifier) ConfirmedFloor(writerID string) int64 {
+	if a, ok := v.floor[writerID]; ok {
+		return a.Load()
 	}
+	return 0
 }
 
 // Run starts the verifier loop. It blocks until the context is cancelled.
@@ -205,8 +236,35 @@ func (v *Verifier) verifyWriter(ctx context.Context, w *Writer) {
 	if expectedSeq == 0 {
 		expectedSeq = 1
 	}
+
+	// One-time startup seed: retention pruning deletes already-verified
+	// documents below the confirmed floor, so after a pod restart (which
+	// resets the in-memory nextSeq to 1) a naive scan from seq 1 would see the
+	// pruned prefix as one enormous gap — a false data-loss verdict. Advance
+	// expectedSeq to the collection's current minimum seq so the verifier only
+	// audits documents that still exist. Real gaps above the surviving floor
+	// are still detected. Done once per writer; steady-state cycles never
+	// re-read below nextSeq.
+	if !v.seeded[writerID] {
+		// Only mark seeded on success: a transient minSeq error (network blip,
+		// majority-read timeout) must retry next cycle, otherwise the seed is
+		// skipped forever and the very next scan runs from seq 1 against a
+		// pruned collection — turning the surviving prefix into one giant false
+		// gap. minSeq returning (0, nil) for an empty collection is a success.
+		newExpected, ok, err := seedExpectedSeq(expectedSeq, func() (int64, error) {
+			return v.minSeq(ctx, writerID)
+		})
+		if err != nil {
+			v.journal.Warn("verifier", fmt.Sprintf("min-seq seed failed for writer %s (will retry): %v", writerID, err))
+		} else {
+			expectedSeq = newExpected
+			v.seeded[writerID] = ok
+		}
+	}
+
 	if maxSeq < expectedSeq {
 		// Nothing new committed since last cycle.
+		v.publishFloor(writerID, expectedSeq-1)
 		return
 	}
 
@@ -223,6 +281,50 @@ func (v *Verifier) verifyWriter(ctx context.Context, w *Writer) {
 	}
 
 	v.nextSeq[writerID] = r.newExpectedSeq
+	v.publishFloor(writerID, r.newExpectedSeq-1)
+}
+
+// publishFloor records the highest fully-verified seq for writerID so the
+// retention pruner can read it concurrently. No-op for writers not registered
+// at construction (e.g. unit tests that pass nil writers).
+func (v *Verifier) publishFloor(writerID string, floor int64) {
+	if a, ok := v.floor[writerID]; ok {
+		a.Store(floor)
+	}
+}
+
+// seedExpectedSeq performs the one-time startup min-seq seed decision for a
+// single writer. lookup resolves the collection's current minimum seq for that
+// writer (v.minSeq in production). It returns the possibly-advanced expectedSeq
+// and whether the seed succeeded; the caller must only mark the writer seeded
+// when ok is true, so a transient lookup error retries on the next cycle rather
+// than being skipped forever (which would scan a pruned collection from seq 1
+// and register a false gap). An empty collection — lookup returning (0, nil) —
+// counts as success and leaves expectedSeq unchanged.
+func seedExpectedSeq(expectedSeq int64, lookup func() (int64, error)) (int64, bool, error) {
+	minSeq, err := lookup()
+	if err != nil {
+		return expectedSeq, false, err
+	}
+	if minSeq > expectedSeq {
+		expectedSeq = minSeq
+	}
+	return expectedSeq, true, nil
+}
+
+// minSeq returns the lowest seq currently stored for writerID, or 0 if the
+// writer has no documents.
+func (v *Verifier) minSeq(ctx context.Context, writerID string) (int64, error) {
+	opts := options.FindOne().SetSort(bson.D{{Key: "seq", Value: 1}})
+	var doc WriteDocument
+	err := v.collection.FindOne(ctx, bson.M{"writer_id": writerID}, opts).Decode(&doc)
+	if err != nil {
+		if errors.Is(err, mongo.ErrNoDocuments) {
+			return 0, nil
+		}
+		return 0, err
+	}
+	return doc.Seq, nil
 }
 
 // scanWriter streams all docs for writerID with seq in [expectedSeq, maxSeq]
diff --git a/test/longhaul/workload/verifier_test.go b/test/longhaul/workload/verifier_test.go
index 799d4e0c..431fcfd0 100644
--- a/test/longhaul/workload/verifier_test.go
+++ b/test/longhaul/workload/verifier_test.go
@@ -4,6 +4,9 @@
 package workload
 
 import (
+	"errors"
+	"sync/atomic"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
@@ -45,10 +48,95 @@ var _ = Describe("Verifier", func() {
 		// what the verifier actually does. Document the boundary.
 		Skip("verifyAll requires a *mongo.Database; covered by long-haul integration runs")
 	})
+
+	Describe("ConfirmedFloor", func() {
+		It("returns 0 for a writer registered but not yet verified", func() {
+			v := &Verifier{floor: map[string]*atomic.Int64{"w000": {}}}
+			Expect(v.ConfirmedFloor("w000")).To(BeZero())
+		})
+
+		It("reflects the published floor and is isolated per writer", func() {
+			v := &Verifier{floor: map[string]*atomic.Int64{
+				"w000": {}, "w001": {},
+			}}
+			v.publishFloor("w000", 4_200)
+			Expect(v.ConfirmedFloor("w000")).To(Equal(int64(4_200)))
+			Expect(v.ConfirmedFloor("w001")).To(BeZero())
+			Expect(v.ConfirmedFloor("unknown")).To(BeZero(),
+				"unregistered writers read as 0 so the pruner never deletes for them")
+		})
+	})
+
+	Describe("startup min-seq seed (seedExpectedSeq)", func() {
+		// The seed is the restart-after-pruning safety net: retention pruning
+		// deletes already-verified docs below the floor, so a restarted pod
+		// (in-memory nextSeq reset to 1) must advance expectedSeq to the
+		// collection's surviving minimum, or the pruned prefix reads as one
+		// giant false gap. These tests lock the two invariants that matter:
+		// (1) a successful seed advances past the pruned prefix, and
+		// (2) a transient lookup error does NOT mark the writer seeded, so it
+		//     retries next cycle instead of being skipped forever.
+
+		It("advances expectedSeq to the surviving minimum after pruning (restart happy path)", func() {
+			// Fresh verifier => expectedSeq starts at 1. The collection was
+			// pruned so its lowest surviving seq is 5_000.
+			got, ok, err := seedExpectedSeq(1, func() (int64, error) { return 5_000, nil })
+			Expect(err).NotTo(HaveOccurred())
+			Expect(ok).To(BeTrue())
+			Expect(got).To(Equal(int64(5_000)),
+				"scan must start at the surviving minimum, not seq 1, so the pruned prefix is not a false gap")
+		})
+
+		It("leaves expectedSeq unchanged when the minimum is at or below it", func() {
+			// Un-pruned / already-ahead resume point: min-seq must not rewind us.
+			got, ok, err := seedExpectedSeq(200, func() (int64, error) { return 1, nil })
+			Expect(err).NotTo(HaveOccurred())
+			Expect(ok).To(BeTrue())
+			Expect(got).To(Equal(int64(200)))
+		})
+
+		It("treats an empty collection (0, nil) as a successful seed", func() {
+			got, ok, err := seedExpectedSeq(1, func() (int64, error) { return 0, nil })
+			Expect(err).NotTo(HaveOccurred())
+			Expect(ok).To(BeTrue(), "an empty collection is a valid state, not a retry-worthy error")
+			Expect(got).To(Equal(int64(1)))
+		})
+
+		It("does NOT mark seeded on a transient error, and retries successfully next cycle", func() {
+			// This is the regression the Major review comment called out: a
+			// transient minSeq error used to mark the writer seeded, permanently
+			// skipping the seed and producing a false data-loss FAIL.
+			calls := 0
+			lookup := func() (int64, error) {
+				calls++
+				if calls == 1 {
+					return 0, errors.New("majority-read timeout")
+				}
+				return 5_000, nil
+			}
+
+			// Simulate the call-site contract: only mark seeded when ok is true.
+			expectedSeq := int64(1)
+			seeded := false
+
+			got, ok, err := seedExpectedSeq(expectedSeq, lookup)
+			Expect(err).To(HaveOccurred())
+			Expect(ok).To(BeFalse())
+			Expect(got).To(Equal(int64(1)), "a failed seed must not move expectedSeq")
+			seeded = seeded || ok
+			Expect(seeded).To(BeFalse(), "writer must remain un-seeded so it retries next cycle")
+
+			// Next cycle: the seed is retried (because still not seeded) and now
+			// succeeds, advancing past the pruned prefix.
+			got, ok, err = seedExpectedSeq(got, lookup)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(ok).To(BeTrue())
+			Expect(got).To(Equal(int64(5_000)))
+			Expect(calls).To(Equal(2), "the transient error must trigger exactly one retry")
+		})
+	})
 })
 
-// makeDoc constructs a valid WriteDocument whose checksum matches; tests that
-// want a checksum mismatch override Checksum directly.
 func makeDoc(writerID string, seq int64) WriteDocument {
 	payload := "p"
 	return WriteDocument{