diff --git a/weed/s3api/s3lifecycle/reader/reader.go b/weed/s3api/s3lifecycle/reader/reader.go
index a5eb1c382..e75a224d9 100644
--- a/weed/s3api/s3lifecycle/reader/reader.go
+++ b/weed/s3api/s3lifecycle/reader/reader.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"strings"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -13,14 +14,36 @@ import (
 )
 
 // Event is one in-shard meta-log event delivered to the router.
+//
+// BootstrapVersion is set only by the bucket bootstrapper when it
+// expands a .versions/ directory; the meta-log path leaves it nil.
+// Carries pre-computed sibling state so the router can fire
+// NoncurrentDays / NewerNoncurrent without listing again.
 type Event struct {
-	TsNs      int64
-	Bucket    string
-	Key       string
-	ShardID   int
-	OldEntry  *filer_pb.Entry
-	NewEntry  *filer_pb.Entry
-	NewParent string
+	TsNs             int64
+	Bucket           string
+	Key              string
+	ShardID          int
+	OldEntry         *filer_pb.Entry
+	NewEntry         *filer_pb.Entry
+	NewParent        string
+	BootstrapVersion *BootstrapVersion
+}
+
+// BootstrapVersion is the per-version state computed once per
+// .versions/<key>/ directory at bootstrap time. Key fields shape
+// EvaluateAction: IsLatest gates current vs. noncurrent rules,
+// NoncurrentIndex gates NewerNoncurrentVersions retention,
+// SuccessorModTime sets the noncurrent clock (when this version was
+// replaced).
+type BootstrapVersion struct {
+	LogicalKey       string
+	VersionID        string
+	IsLatest         bool
+	IsDeleteMarker   bool
+	NumVersions      int
+	NoncurrentIndex  int // 0 = newest noncurrent
+	SuccessorModTime time.Time
 }
 
 // IsDelete reports whether this event removes an entry.
diff --git a/weed/s3api/s3lifecycle/router/router.go b/weed/s3api/s3lifecycle/router/router.go
index af1d82ce5..e3ab9abcd 100644
--- a/weed/s3api/s3lifecycle/router/router.go
+++ b/weed/s3api/s3lifecycle/router/router.go
@@ -71,6 +71,13 @@ func Route(ctx context.Context, snap *engine.Snapshot, ev *reader.Event, now tim
 	if len(keys) == 0 {
 		return nil
 	}
+	// Bootstrap-expanded version event: sibling state is pre-computed,
+	// info.Key is the LOGICAL key so rule prefixes match. Skip the
+	// meta-log path's version-folder skip.
+	if ev.BootstrapVersion != nil {
+		return routeBootstrapVersion(snap, ev, keys)
+	}
+
 	versioned := snap.BucketVersioned(ev.Bucket)
 
 	// EXP_DM can fire on two version-folder events: the marker create
@@ -220,6 +227,91 @@ func routeSoleSurvivorMarker(ctx context.Context, snap *engine.Snapshot, ev *rea
 	return matches
 }
 
+// routeBootstrapVersion handles a synthesized event from BucketBootstrapper.
+// The bootstrap walker has already listed .versions/<key>/, sorted siblings
+// newest-first, and stamped each one's IsLatest / NoncurrentIndex /
+// SuccessorModTime. The router only needs to assemble ObjectInfo and run
+// the match loop with the standard kind gates. ev.NewEntry is the version
+// file itself; ev.Key is the version-folder path; the LOGICAL key from
+// BootstrapVersion drives prefix matching and the dispatcher.
+func routeBootstrapVersion(snap *engine.Snapshot, ev *reader.Event, keys []s3lifecycle.ActionKey) []Match {
+	bv := ev.BootstrapVersion
+	entry := ev.NewEntry
+	if entry == nil || entry.Attributes == nil || bv.LogicalKey == "" {
+		return nil
+	}
+	idx := bv.NoncurrentIndex
+	info := &s3lifecycle.ObjectInfo{
+		Key:              bv.LogicalKey,
+		ModTime:          time.Unix(entry.Attributes.Mtime, int64(entry.Attributes.MtimeNs)),
+		Size:             int64(entry.Attributes.FileSize),
+		IsLatest:         bv.IsLatest,
+		IsDeleteMarker:   bv.IsDeleteMarker,
+		NumVersions:      bv.NumVersions,
+		SuccessorModTime: bv.SuccessorModTime,
+	}
+	if !bv.IsLatest {
+		info.NoncurrentIndex = &idx
+	}
+	if tags := extractTags(entry.Extended); len(tags) > 0 {
+		info.Tags = tags
+	}
+	eventTime := time.Unix(0, ev.TsNs)
+	identity := buildIdentityFromEntry(entry)
+	var matches []Match
+	for _, key := range keys {
+		action := snap.Action(key)
+		if action == nil || !action.IsActive() || action.Mode != engine.ModeEventDriven {
+			continue
+		}
+		// ABORT_MPU never applies to a versioned object.
+		if key.ActionKind == s3lifecycle.ActionKindAbortMPU {
+			continue
+		}
+		// Noncurrent rules clock from when this version was replaced
+		// (SuccessorModTime), not from when it was originally written.
+		// Bootstrap populates SuccessorModTime; fall back to ModTime
+		// for the latest version (no successor exists).
+		clock := info.ModTime
+		if !info.IsLatest && !info.SuccessorModTime.IsZero() {
+			clock = info.SuccessorModTime
+		}
+		dueTime := clock.Add(action.Delay)
+		res := s3lifecycle.EvaluateAction(action.Rule, key.ActionKind, info, dueTime)
+		if res.Action == s3lifecycle.ActionNone {
+			continue
+		}
+		// Pin the version_id only for kinds the dispatcher needs to
+		// target by version: noncurrent retention and the marker
+		// itself. EXPIRATION_DAYS / EXPIRATION_DATE on the latest
+		// must NOT carry it — between schedule and dispatch a fresh
+		// PUT can land, and identity-CAS against the original
+		// version's bytes would still pass even though the latest has
+		// moved on. Empty VersionID makes the dispatcher fetch the
+		// current latest, where identity-CAS resolves to STALE_IDENTITY
+		// and bootstrap re-schedules with the new latest's identity.
+		var matchVersionID string
+		switch key.ActionKind {
+		case s3lifecycle.ActionKindNoncurrentDays,
+			s3lifecycle.ActionKindNewerNoncurrent,
+			s3lifecycle.ActionKindExpiredDeleteMarker:
+			matchVersionID = bv.VersionID
+		}
+		matches = append(matches, Match{
+			Key:       key,
+			Action:    action,
+			Result:    res,
+			EventTs:   eventTime,
+			DueTime:   dueTime,
+			Bucket:    ev.Bucket,
+			ObjectKey: bv.LogicalKey,
+			VersionID: matchVersionID,
+			Identity:  identity,
+		})
+	}
+	return matches
+}
+
 // logicalKeyFromVersionPath extracts <logical> from <logical>.versions/<file>.
 // Returns false for a bucket-root marker (path = ".versions/<file>"), since
 // AWS S3 has no concept of an object at the bucket key itself.
diff --git a/weed/s3api/s3lifecycle/router/router_test.go b/weed/s3api/s3lifecycle/router/router_test.go
index 909576f43..76f632e5e 100644
--- a/weed/s3api/s3lifecycle/router/router_test.go
+++ b/weed/s3api/s3lifecycle/router/router_test.go
@@ -828,3 +828,167 @@ func TestRouteVersionedAllVersionFolderPathsSkipped(t *testing.T) {
 	}
 }
 
+
+func bootstrapVersionEntry(versionID string, mtime time.Time, isDeleteMarker bool) *filer_pb.Entry {
+	ext := map[string][]byte{
+		s3_constants.ExtVersionIdKey: []byte(versionID),
+	}
+	if isDeleteMarker {
+		ext[s3_constants.ExtDeleteMarkerKey] = []byte("true")
+	}
+	mtimeUnix := mtime.Unix()
+	return &filer_pb.Entry{
+		Name: "v_" + versionID,
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime:   mtimeUnix,
+			MtimeNs: int32(mtime.UnixNano() - mtimeUnix*int64(1e9)),
+		},
+		Extended: ext,
+	}
+}
+
+func TestRouteBootstrapVersionLatestExpirationDaysFires(t *testing.T) {
+	// Bootstrap-emitted event for the LATEST version of a versioned
+	// object. ExpirationDays should fire (creates a delete marker at
+	// dispatch). ObjectKey is the LOGICAL key. VersionID must be EMPTY
+	// for EXPIRATION_DAYS so the dispatcher fetches the current latest:
+	// if a fresh PUT landed between schedule and dispatch, identity-CAS
+	// against the original version's bytes would pass even though the
+	// latest has moved on.
+	rule := &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	old := now.AddDate(0, 0, -2)
+	entry := bootstrapVersionEntry("v-current", old, false)
+	ev := &reader.Event{
+		Bucket:   "bk",
+		Key:      "logs/foo" + s3_constants.VersionsFolder + "/v_v-current",
+		NewEntry: entry,
+		BootstrapVersion: &reader.BootstrapVersion{
+			LogicalKey:  "logs/foo",
+			VersionID:   "v-current",
+			IsLatest:    true,
+			NumVersions: 1,
+		},
+	}
+	matches := Route(context.Background(), snap, ev, now, nil)
+	if len(matches) != 1 {
+		t.Fatalf("want 1 match (ExpirationDays on latest), got %v", matches)
+	}
+	if matches[0].ObjectKey != "logs/foo" {
+		t.Fatalf("ObjectKey=%q, want logs/foo", matches[0].ObjectKey)
+	}
+	if matches[0].VersionID != "" {
+		t.Fatalf("VersionID=%q, want empty for EXPIRATION_DAYS", matches[0].VersionID)
+	}
+}
+
+func TestRouteBootstrapVersionNoncurrentDaysFires(t *testing.T) {
+	// Bootstrap-emitted event for a NONCURRENT version. NoncurrentDays
+	// uses SuccessorModTime as the clock — when this version was
+	// replaced by the next-newer sibling.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	successor := now.AddDate(0, 0, -3) // replaced 3 days ago
+	old := now.AddDate(0, 0, -10)      // mtime older still
+	entry := bootstrapVersionEntry("v-old", old, false)
+	ev := &reader.Event{
+		Bucket:   "bk",
+		Key:      "logs/foo" + s3_constants.VersionsFolder + "/v_v-old",
+		NewEntry: entry,
+		BootstrapVersion: &reader.BootstrapVersion{
+			LogicalKey:       "logs/foo",
+			VersionID:        "v-old",
+			IsLatest:         false,
+			NumVersions:      2,
+			NoncurrentIndex:  0,
+			SuccessorModTime: successor,
+		},
+	}
+	matches := Route(context.Background(), snap, ev, now, nil)
+	if len(matches) != 1 {
+		t.Fatalf("want 1 match (NoncurrentDays), got %v", matches)
+	}
+	if matches[0].VersionID != "v-old" {
+		t.Fatalf("VersionID=%q, want v-old", matches[0].VersionID)
+	}
+}
+
+func TestRouteBootstrapVersionNoncurrentRespectsNewerNoncurrentVersions(t *testing.T) {
+	// NewerNoncurrentVersions=2 keeps the two newest noncurrents safe.
+	// A version at NoncurrentIndex=0 (newest noncurrent) must NOT fire;
+	// index=2 (third-newest) MUST fire.
+	rule := &s3lifecycle.Rule{
+		ID:                              "r",
+		Status:                          s3lifecycle.StatusEnabled,
+		NoncurrentVersionExpirationDays: 1,
+		NewerNoncurrentVersions:         2,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	successor := now.AddDate(0, 0, -3)
+	old := now.AddDate(0, 0, -10)
+
+	mk := func(idx int) *reader.Event {
+		return &reader.Event{
+			Bucket:   "bk",
+			Key:      "logs/foo" + s3_constants.VersionsFolder + "/v_old" + string(rune('0'+idx)),
+			NewEntry: bootstrapVersionEntry("old"+string(rune('0'+idx)), old, false),
+			BootstrapVersion: &reader.BootstrapVersion{
+				LogicalKey:       "logs/foo",
+				VersionID:        "old" + string(rune('0'+idx)),
+				IsLatest:         false,
+				NumVersions:      4,
+				NoncurrentIndex:  idx,
+				SuccessorModTime: successor,
+			},
+		}
+	}
+
+	if got := Route(context.Background(), snap, mk(0), now, nil); len(got) != 0 {
+		t.Fatalf("noncurrent rank 0 must be retained, got %v", got)
+	}
+	if got := Route(context.Background(), snap, mk(1), now, nil); len(got) != 0 {
+		t.Fatalf("noncurrent rank 1 must be retained, got %v", got)
+	}
+	if got := Route(context.Background(), snap, mk(2), now, nil); len(got) != 1 {
+		t.Fatalf("noncurrent rank 2 must fire, got %v", got)
+	}
+}
+
+func TestRouteBootstrapVersionAbortMPUNeverEmittedForVersion(t *testing.T) {
+	// AbortIncompleteMultipartUpload only applies to MPU init dirs, not
+	// versioned object versions. Even if the bucket has the rule, a
+	// bootstrap version event must not produce an ABORT_MPU match.
+	rule := &s3lifecycle.Rule{
+		ID:                          "r",
+		Status:                      s3lifecycle.StatusEnabled,
+		AbortMPUDaysAfterInitiation: 1,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	old := now.AddDate(0, 0, -10)
+	ev := &reader.Event{
+		Bucket:   "bk",
+		Key:      "logs/foo" + s3_constants.VersionsFolder + "/v_x",
+		NewEntry: bootstrapVersionEntry("x", old, false),
+		BootstrapVersion: &reader.BootstrapVersion{
+			LogicalKey:  "logs/foo",
+			VersionID:   "x",
+			IsLatest:    true,
+			NumVersions: 1,
+		},
+	}
+	if got := Route(context.Background(), snap, ev, now, nil); len(got) != 0 {
+		t.Fatalf("bootstrap version event must not produce ABORT_MPU match, got %v", got)
+	}
+}
diff --git a/weed/s3api/s3lifecycle/scheduler/bootstrap.go b/weed/s3api/s3lifecycle/scheduler/bootstrap.go
index ee66bf993..61558e01c 100644
--- a/weed/s3api/s3lifecycle/scheduler/bootstrap.go
+++ b/weed/s3api/s3lifecycle/scheduler/bootstrap.go
@@ -3,14 +3,17 @@ package scheduler
 import (
 	"context"
 	"fmt"
+	"sort"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
 // EventInjector is the bootstrap-side hook into the dispatcher pipeline.
@@ -20,6 +23,39 @@ type EventInjector interface {
 	InjectEvent(ctx context.Context, ev *reader.Event) error
 }
 
+// listPageSize is the page size for paginated directory listings during
+// the bucket walk. The filer caps SeaweedList(..., limit=0) at
+// DirListingLimit (1000 by default) per call, so a single-page list
+// would silently truncate large directories — a correctness bug for
+// noncurrent retention since older versions past the page boundary
+// would never reach the rank/sort math. var so tests can shrink it
+// without producing thousands of entries.
+var listPageSize uint32 = 1024
+
+// listAll issues paginated SeaweedList calls until the listing is
+// exhausted, invoking fn for every entry. Pagination uses
+// startFrom = lastEntryName (exclusive) to advance.
+func listAll(ctx context.Context, client filer_pb.SeaweedFilerClient, dir string, fn func(*filer_pb.Entry) error) error {
+	startFrom := ""
+	for {
+		var pageCount uint32
+		var lastName string
+		if err := filer_pb.SeaweedList(ctx, client, dir, "", func(e *filer_pb.Entry, _ bool) error {
+			pageCount++
+			if e != nil {
+				lastName = e.Name
+			}
+			return fn(e)
+		}, startFrom, false, listPageSize); err != nil {
+			return err
+		}
+		if pageCount < listPageSize {
+			return nil
+		}
+		startFrom = lastName
+	}
+}
+
 // BucketBootstrapper backfills already-existing entries when a freshly-PUT
 // rule's bucket appears in the engine. The reader-driven path only sees
 // meta-log events created after the rule lands; without this walk,
@@ -75,7 +111,26 @@ func (b *BucketBootstrapper) walkBucket(ctx context.Context, bucket string) {
 	root := strings.TrimSuffix(b.BucketsPath, "/") + "/" + bucket
 	glog.V(0).Infof("lifecycle bootstrap: starting walk for bucket %s (root=%s)", bucket, root)
 	count := 0
-	if err := walkBucketDir(ctx, b.FilerClient, root, root, func(entry *filer_pb.Entry, key string) error {
+	// skipBare records bucket-relative bare-key paths that
+	// expandVersionsDir already routed as the null version. Without it
+	// the walker's regular emission would also fire for the bare entry
+	// — in a versioned bucket buildObjectInfo classifies it as
+	// IsLatest=true, NumVersions=0, and ExpirationDays would create a
+	// stray delete marker that hides the real latest.
+	skipBare := map[string]bool{}
+	var cb func(entry *filer_pb.Entry, key string) error
+	cb = func(entry *filer_pb.Entry, key string) error {
+		if isVersionsDir(entry) {
+			n, err := b.expandVersionsDir(ctx, bucket, root, key, entry, cb, skipBare)
+			count += n
+			return err
+		}
+		if !entry.IsDirectory && skipBare[key] {
+			return nil
+		}
+		if entry.IsDirectoryKeyObject() && skipBare[key] {
+			return nil
+		}
 		ev := &reader.Event{
 			// TsNs=0 sentinel: dispatcher.advance treats <=0 as no-op,
 			// so the reader's persisted cursor isn't ratcheted forward
@@ -88,7 +143,8 @@ func (b *BucketBootstrapper) walkBucket(ctx context.Context, bucket string) {
 		}
 		count++
 		return b.Injector.InjectEvent(ctx, ev)
-	}); err != nil {
+	}
+	if err := walkBucketDir(ctx, b.FilerClient, root, root, cb); err != nil {
 		if ctx.Err() == nil {
 			glog.V(0).Infof("lifecycle bootstrap %s: %v", bucket, err)
 		}
@@ -97,42 +153,244 @@ func (b *BucketBootstrapper) walkBucket(ctx context.Context, bucket string) {
 	glog.V(0).Infof("lifecycle bootstrap: bucket %s injected %d entries", bucket, count)
 }
 
-// walkBucketDir lists every file under dir recursively and invokes cb
-// with the filer entry plus its bucket-relative key. MPU init dirs at
-// .uploads/<id> are emitted as a single (directory-shaped) entry so the
-// router's MPU detection fires; deeper directories recurse.
-func walkBucketDir(ctx context.Context, client filer_pb.SeaweedFilerClient, dir, bucketRoot string, cb func(entry *filer_pb.Entry, key string) error) error {
+// versionItem is the per-sibling state expandVersionsDir builds: the
+// filer entry plus its version_id (or "null" for the bare null version
+// living outside .versions/). isExplicitNull marks bare entries the
+// suspended-versioning write path tagged with ExtVersionIdKey="null"
+// (s3api_object_handlers_put.go); we trust those as latest when the
+// .versions/ pointer is missing. A pre-versioning bare object has no
+// such marker, so a missing pointer there is a race window with a new
+// version write and we keep the newest-sibling fallback.
+type versionItem struct {
+	entry          *filer_pb.Entry
+	versionID      string
+	bareKey        string // bucket-relative path; non-empty only for the null version
+	isExplicitNull bool
+}
+
+// expandVersionsDir lists <root>/<key>/ and, when the children look like
+// SeaweedFS version files, injects one reader.Event per version with
+// BootstrapVersion populated. The bare logical key (the "null" version,
+// living outside .versions/) is included as a sibling so:
+//   - pre-versioning objects with a newer .versions/ history fire
+//     NoncurrentDays as id="null"
+//   - suspended-bucket writes (which clear the .versions/ latest pointer)
+//     correctly classify null as the current version while every
+//     .versions/ child becomes noncurrent
+//
+// versionsKey is the bucket-relative path of the .versions/ directory
+// (e.g. "logs/foo.versions"). When the bare null version is included,
+// its bucket-relative path is added to skipBare so the walker's regular
+// emission for the same entry is suppressed.
+//
+// When no child has ExtVersionIdKey the directory is a coincidentally-
+// named user folder; recurse via fallback (the bucket walk's own cb).
+func (b *BucketBootstrapper) expandVersionsDir(ctx context.Context, bucket, root, versionsKey string, versionsEntry *filer_pb.Entry, fallback func(*filer_pb.Entry, string) error, skipBare map[string]bool) (int, error) {
+	logical := strings.TrimSuffix(versionsKey, s3_constants.VersionsFolder)
+	if logical == "" {
+		return 0, nil
+	}
+	versionsDir := strings.TrimSuffix(b.BucketsPath, "/") + "/" + bucket + "/" + versionsKey
+	// Collect file children only. Subdirectories under .versions/ would
+	// corrupt sort/rank math; the disambiguation pass below also wants
+	// to see only file-shaped children. Paginate so a hot key with
+	// thousands of versions doesn't truncate at DirListingLimit.
 	var children []*filer_pb.Entry
-	if err := filer_pb.SeaweedList(ctx, client, dir, "", func(e *filer_pb.Entry, _ bool) error {
-		children = append(children, e)
+	if err := listAll(ctx, b.FilerClient, versionsDir, func(e *filer_pb.Entry) error {
+		if e != nil && e.Attributes != nil && !e.IsDirectory {
+			children = append(children, e)
+		}
 		return nil
-	}, "", false, 0); err != nil {
+	}); err != nil {
+		return 0, fmt.Errorf("list %s: %w", versionsDir, err)
+	}
+	items := make([]versionItem, 0, len(children)+1)
+	for _, e := range children {
+		if id, ok := e.Extended[s3_constants.ExtVersionIdKey]; ok && len(id) > 0 {
+			items = append(items, versionItem{entry: e, versionID: string(id)})
+		}
+	}
+	if len(items) == 0 {
+		// Coincidentally-named user folder (or an empty .versions
+		// container). fallback is the bucket walk's own cb so nested
+		// .versions/ entries inside still expand.
+		if fallback == nil {
+			return 0, nil
+		}
+		if err := walkBucketDir(ctx, b.FilerClient, versionsDir, root, fallback); err != nil {
+			return 0, err
+		}
+		return 0, nil
+	}
+	// Look up the bare null version. SeaweedFS keeps it at the logical
+	// path for pre-versioning objects and for suspended-bucket writes.
+	// Both shapes count: regular file (PUT'd object) and explicit S3
+	// directory-key marker (object name ends in /).
+	if nullEntry, nullKey, explicit, ok := b.lookupNullVersion(ctx, bucket, logical); ok {
+		items = append(items, versionItem{
+			entry:          nullEntry,
+			versionID:      "null",
+			bareKey:        nullKey,
+			isExplicitNull: explicit,
+		})
+	}
+	// Sort newest-first: primary by mtime ns, fallback by version_id
+	// (CompareVersionIds returns <0 when first arg is newer). PUTs only
+	// set second-level Mtime, so collisions in the same second are
+	// resolved by the canonical version-id ordering used elsewhere.
+	sort.SliceStable(items, func(i, j int) bool {
+		mi := items[i].entry.Attributes.Mtime*int64(1e9) + int64(items[i].entry.Attributes.MtimeNs)
+		mj := items[j].entry.Attributes.Mtime*int64(1e9) + int64(items[j].entry.Attributes.MtimeNs)
+		if mi != mj {
+			return mi > mj
+		}
+		return s3lifecycle.CompareVersionIds(items[i].versionID, items[j].versionID) < 0
+	})
+	// Resolve latest position.
+	//   1. Pointer names a real id -> that wins (in-order or backdated).
+	//   2. Pointer absent + items[0] is an EXPLICIT null (suspended write
+	//      cleared the pointer and tagged the bare object as null, AND
+	//      the bare object is newest by mtime) -> null is latest.
+	//   3. Pointer absent in any other shape: fall back to newest
+	//      sibling. Catches the post-suspended re-enable race window —
+	//      a fresh .versions/<v1> write whose pointer update hasn't
+	//      landed yet outranks the older suspended-null bare object.
+	latestID := string(versionsEntry.Extended[s3_constants.ExtLatestVersionIdKey])
+	latestPos := 0
+	if latestID != "" {
+		for i, it := range items {
+			if it.versionID == latestID {
+				latestPos = i
+				break
+			}
+		}
+	} else if len(items) > 0 && items[0].versionID == "null" && items[0].isExplicitNull {
+		latestPos = 0
+	}
+	count := 0
+	for i, it := range items {
+		var successor time.Time
+		if i > 0 {
+			prev := items[i-1].entry.Attributes
+			successor = time.Unix(prev.Mtime, int64(prev.MtimeNs))
+		}
+		bv := &reader.BootstrapVersion{
+			LogicalKey:       logical,
+			VersionID:        it.versionID,
+			IsLatest:         i == latestPos,
+			IsDeleteMarker:   string(it.entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true",
+			NumVersions:      len(items),
+			SuccessorModTime: successor,
+		}
+		if !bv.IsLatest {
+			rank := i
+			if i > latestPos {
+				rank = i - 1
+			}
+			bv.NoncurrentIndex = rank
+		}
+		// Event Key for bookkeeping: real version files keep the
+		// .versions/<file> path; the null version uses its bare path
+		// so the dispatcher's identity check resolves to the same
+		// entry the walker would have emitted.
+		evKey := versionsKey + "/" + it.entry.Name
+		if it.versionID == "null" {
+			evKey = it.bareKey
+		}
+		ev := &reader.Event{
+			TsNs:             0,
+			Bucket:           bucket,
+			Key:              evKey,
+			ShardID:          s3lifecycle.ShardID(bucket, logical),
+			NewEntry:         it.entry,
+			BootstrapVersion: bv,
+		}
+		if err := b.Injector.InjectEvent(ctx, ev); err != nil {
+			return count, err
+		}
+		if it.versionID == "null" && skipBare != nil {
+			skipBare[it.bareKey] = true
+		}
+		count++
+	}
+	return count, nil
+}
+
+// lookupNullVersion returns the bare-key entry that represents the null
+// version of logical, if any. Both regular files and S3 directory-key
+// markers (an empty directory entry with Mime set) qualify. The
+// explicit return reports whether the entry's Extended map carries
+// ExtVersionIdKey == "null" — the marker the suspended-versioning
+// write path applies (s3api_object_handlers_put.go). bucketRelKey is
+// the bucket-relative path the walker would otherwise emit, so the
+// caller can suppress the duplicate.
+func (b *BucketBootstrapper) lookupNullVersion(ctx context.Context, bucket, logical string) (entry *filer_pb.Entry, bucketRelKey string, explicit bool, ok bool) {
+	bucketPath := strings.TrimSuffix(b.BucketsPath, "/") + "/" + bucket
+	parent, name := util.NewFullPath(bucketPath, logical).DirAndName()
+	resp, err := filer_pb.LookupEntry(ctx, b.FilerClient, &filer_pb.LookupDirectoryEntryRequest{
+		Directory: parent,
+		Name:      name,
+	})
+	if err != nil || resp == nil || resp.Entry == nil {
+		return nil, "", false, false
+	}
+	e := resp.Entry
+	if e.IsDirectory && !e.IsDirectoryKeyObject() {
+		return nil, "", false, false
+	}
+	if id, hasID := e.Extended[s3_constants.ExtVersionIdKey]; hasID && string(id) == "null" {
+		explicit = true
+	}
+	return e, strings.TrimPrefix(parent+"/"+name, bucketPath+"/"), explicit, true
+}
+
+// walkBucketDir streams entries under dir and invokes cb. Two kinds of
+// directories are emitted whole rather than recursed into:
+//   - .uploads/<id> MPU init dirs (router fires ABORT_MPU off the dir entry)
+//   - <key>.versions/ directories (caller expands them into per-version
+//     events; recursing here would emit individual version files without
+//     the sibling state needed for NoncurrentDays / NewerNoncurrent)
+//
+// .versions/ dirs are processed before everything else at each level so
+// the cb's expandVersionsDir call can record the bare null-version key
+// in the walk-shared skip set before the same level emits the bare entry.
+// Two streaming passes (rather than buffering the whole directory) trade
+// a second listing for bounded memory on flat buckets with millions of
+// entries.
+func walkBucketDir(ctx context.Context, client filer_pb.SeaweedFilerClient, dir, bucketRoot string, cb func(entry *filer_pb.Entry, key string) error) error {
+	// Pass 1: .versions/ dirs only.
+	if err := listAll(ctx, client, dir, func(e *filer_pb.Entry) error {
+		if e == nil || e.Attributes == nil {
+			return nil
+		}
+		if !e.IsDirectory || !isVersionsDir(e) {
+			return nil
+		}
+		full := dir + "/" + e.Name
+		key := strings.TrimPrefix(full, bucketRoot+"/")
+		return cb(e, key)
+	}); err != nil {
 		return fmt.Errorf("list %s: %w", dir, err)
 	}
-	for _, entry := range children {
-		if entry == nil || entry.Attributes == nil {
-			continue
+	// Pass 2: everything else. Bare entries whose name was claimed by
+	// a sibling .versions/ expansion are dropped by the cb's skip-set.
+	return listAll(ctx, client, dir, func(e *filer_pb.Entry) error {
+		if e == nil || e.Attributes == nil {
+			return nil
 		}
-		full := dir + "/" + entry.Name
+		if e.IsDirectory && isVersionsDir(e) {
+			return nil
+		}
+		full := dir + "/" + e.Name
 		key := strings.TrimPrefix(full, bucketRoot+"/")
-
-		if entry.IsDirectory {
-			if isMPUInitDir(key, entry) {
-				if err := cb(entry, key); err != nil {
-					return err
-				}
-				continue
+		if e.IsDirectory {
+			if isMPUInitDir(key, e) {
+				return cb(e, key)
 			}
-			if err := walkBucketDir(ctx, client, full, bucketRoot, cb); err != nil {
-				return err
-			}
-			continue
+			return walkBucketDir(ctx, client, full, bucketRoot, cb)
 		}
-		if err := cb(entry, key); err != nil {
-			return err
-		}
-	}
-	return nil
+		return cb(e, key)
+	})
 }
 
 // isMPUInitDir mirrors router.mpuInitInfo: a directory at .uploads/<id>
@@ -152,3 +410,16 @@ func isMPUInitDir(key string, entry *filer_pb.Entry) bool {
 	return ok && len(v) > 0
 }
 
+// isVersionsDir matches `<x>.versions/` by name suffix. We can't gate on
+// ExtLatestVersionIdKey here: createDeleteMarker writes the version file
+// before updating the parent's Extended pointer, so a walk that races
+// with that update would see the directory without the pointer and
+// recurse into raw version files, losing the sibling state needed for
+// noncurrent rules. expandVersionsDir handles disambiguation by
+// inspecting children for ExtVersionIdKey; coincidentally-named
+// directories that aren't real .versions storage fall through to a
+// regular recursion.
+func isVersionsDir(entry *filer_pb.Entry) bool {
+	return entry.IsDirectory && strings.HasSuffix(entry.Name, s3_constants.VersionsFolder)
+}
+
diff --git a/weed/s3api/s3lifecycle/scheduler/bootstrap_test.go b/weed/s3api/s3lifecycle/scheduler/bootstrap_test.go
index 30c674108..fc99d608e 100644
--- a/weed/s3api/s3lifecycle/scheduler/bootstrap_test.go
+++ b/weed/s3api/s3lifecycle/scheduler/bootstrap_test.go
@@ -3,7 +3,9 @@ package scheduler
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
+	"sort"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -52,8 +54,8 @@ func (s *fakeListStream) SendMsg(any) error { return nil }
 func (s *fakeListStream) RecvMsg(any) error { return nil }
 
 // fakeFilerClient embeds SeaweedFilerClient (nil interface) and overrides
-// only ListEntries. Calling any other method panics, which is fine for
-// these tests.
+// ListEntries + LookupDirectoryEntry. Calling any other method panics,
+// which is fine for these tests.
 type fakeFilerClient struct {
 	filer_pb.SeaweedFilerClient
 
@@ -63,15 +65,49 @@ type fakeFilerClient struct {
 	listedN  int32                        // atomic counter for cross-goroutine reads
 }
 
+func (c *fakeFilerClient) LookupDirectoryEntry(ctx context.Context, in *filer_pb.LookupDirectoryEntryRequest, opts ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	for _, e := range c.tree[in.Directory] {
+		if e != nil && e.Name == in.Name {
+			return &filer_pb.LookupDirectoryEntryResponse{Entry: e}, nil
+		}
+	}
+	return nil, filer_pb.ErrNotFound
+}
+
 func (c *fakeFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) {
 	c.mu.Lock()
 	c.listed = append(c.listed, in.Directory)
 	c.mu.Unlock()
 	atomic.AddInt32(&c.listedN, 1)
 
-	children := c.tree[in.Directory]
-	responses := make([]*filer_pb.ListEntriesResponse, 0, len(children))
-	for _, e := range children {
+	// Mirror the filer: sort children by name, honor StartFromFileName
+	// (exclusive unless InclusiveStartFrom), and cap at Limit. listAll's
+	// pagination loop depends on these semantics to advance correctly.
+	src := c.tree[in.Directory]
+	filtered := make([]*filer_pb.Entry, 0, len(src))
+	for _, e := range src {
+		if e == nil {
+			continue
+		}
+		if in.StartFromFileName != "" {
+			if in.InclusiveStartFrom {
+				if e.Name < in.StartFromFileName {
+					continue
+				}
+			} else if e.Name <= in.StartFromFileName {
+				continue
+			}
+		}
+		filtered = append(filtered, e)
+	}
+	sort.SliceStable(filtered, func(i, j int) bool { return filtered[i].Name < filtered[j].Name })
+	if in.Limit > 0 && uint32(len(filtered)) > in.Limit {
+		filtered = filtered[:in.Limit]
+	}
+	responses := make([]*filer_pb.ListEntriesResponse, 0, len(filtered))
+	for _, e := range filtered {
 		responses = append(responses, &filer_pb.ListEntriesResponse{Entry: e})
 	}
 	return &fakeListStream{responses: responses, ctx: ctx}, nil
@@ -345,13 +381,18 @@ func TestBucketBootstrapper_KickOffNew_LaunchesPerBucket(t *testing.T) {
 
 	b.KickOffNew(context.Background(), []string{"bucketA", "bucketB"})
 
-	// Both walks must hit ListEntries once each (empty trees -> no recursion).
+	// Each walk lists the bucket root twice (pass 1: .versions/, pass 2:
+	// everything else); 2 buckets * 2 passes = 4 listings total.
 	waitFor(t, func() bool {
-		return atomic.LoadInt32(&client.listedN) >= 2
-	}, "both bucket walks to start")
+		return atomic.LoadInt32(&client.listedN) >= 4
+	}, "both bucket walks to complete")
 
 	listed := client.listedCopy()
-	assert.ElementsMatch(t, []string{"/buckets/bucketA", "/buckets/bucketB"}, listed)
+	seen := map[string]bool{}
+	for _, d := range listed {
+		seen[d] = true
+	}
+	assert.Equal(t, map[string]bool{"/buckets/bucketA": true, "/buckets/bucketB": true}, seen)
 
 	b.mu.Lock()
 	defer b.mu.Unlock()
@@ -370,31 +411,34 @@ func TestBucketBootstrapper_KickOffNew_SkipsAlreadyKnown(t *testing.T) {
 	}
 
 	b.KickOffNew(context.Background(), []string{"bucketA", "bucketB"})
+	// Each walk does two ListEntries calls (pass 1: .versions/, pass 2:
+	// everything else). 2 buckets * 2 passes = 4 listings.
 	waitFor(t, func() bool {
-		return atomic.LoadInt32(&client.listedN) >= 2
+		return atomic.LoadInt32(&client.listedN) >= 4
 	}, "first wave to complete")
 
 	firstWave := atomic.LoadInt32(&client.listedN)
 
 	// Second call: bucketA is already known, bucketC is new. Only one
-	// new walk should fire.
+	// new walk should fire (2 listings).
 	b.KickOffNew(context.Background(), []string{"bucketA", "bucketC"})
 	waitFor(t, func() bool {
-		return atomic.LoadInt32(&client.listedN) >= firstWave+1
-	}, "bucketC walk to start")
+		return atomic.LoadInt32(&client.listedN) >= firstWave+2
+	}, "bucketC walk to complete")
 
 	// Give a moment for any spurious bucketA walk to also tick.
 	time.Sleep(20 * time.Millisecond)
 
 	listed := client.listedCopy()
-	// Count distinct buckets walked.
+	// Each bucket walks once across both calls; the walk does two
+	// listings of the bucket root (pass 1 + pass 2).
 	bucketCount := map[string]int{}
 	for _, d := range listed {
 		bucketCount[d]++
 	}
-	assert.Equal(t, 1, bucketCount["/buckets/bucketA"], "bucketA must be walked exactly once across both calls")
-	assert.Equal(t, 1, bucketCount["/buckets/bucketB"])
-	assert.Equal(t, 1, bucketCount["/buckets/bucketC"])
+	assert.Equal(t, 2, bucketCount["/buckets/bucketA"], "bucketA must be walked exactly once (=2 list calls) across both calls")
+	assert.Equal(t, 2, bucketCount["/buckets/bucketB"])
+	assert.Equal(t, 2, bucketCount["/buckets/bucketC"])
 
 	b.mu.Lock()
 	defer b.mu.Unlock()
@@ -435,3 +479,624 @@ func TestBucketBootstrapper_KickOffNew_EmptyBucketListIsNoop(t *testing.T) {
 	assert.Equal(t, int32(0), atomic.LoadInt32(&client.listedN))
 	assert.Empty(t, inj.snapshot())
 }
+
+// versionFile builds a version-file entry with the given mtime, version_id,
+// and optional delete-marker flag.
+func versionFile(versionID string, mtime time.Time, isMarker bool) *filer_pb.Entry {
+	ext := map[string][]byte{
+		s3_constants.ExtVersionIdKey: []byte(versionID),
+	}
+	if isMarker {
+		ext[s3_constants.ExtDeleteMarkerKey] = []byte("true")
+	}
+	return &filer_pb.Entry{
+		Name: "v_" + versionID,
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime: mtime.Unix(),
+		},
+		Extended: ext,
+	}
+}
+
+func TestWalkBucketDir_VersionsDirEmittedOnceAndNotRecursed(t *testing.T) {
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v2"),
+	})
+	now := time.Now()
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot: {versionsDir},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", now.Add(-2*time.Hour), false),
+				versionFile("v2", now, false),
+			},
+		},
+	}
+	var seen []string
+	err := walkBucketDir(context.Background(), client, testBucketRoot, testBucketRoot, func(_ *filer_pb.Entry, key string) error {
+		seen = append(seen, key)
+		return nil
+	})
+	require.NoError(t, err)
+	assert.Equal(t, []string{"foo" + s3_constants.VersionsFolder}, seen)
+	assert.NotContains(t, client.listedCopy(), testBucketRoot+"/foo"+s3_constants.VersionsFolder)
+}
+
+func TestWalkBucketDir_VersionsDirEmittedRegardlessOfLatestPointer(t *testing.T) {
+	// walkBucketDir matches <x>.versions/ purely on the name suffix —
+	// gating on ExtLatestVersionIdKey would lose the race window where
+	// the version file exists before the parent's metadata update lands.
+	// expandVersionsDir handles disambiguation by inspecting children.
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, nil)
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot: {versionsDir},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {fileEntry("inner.txt")},
+		},
+	}
+	var seen []string
+	err := walkBucketDir(context.Background(), client, testBucketRoot, testBucketRoot, func(_ *filer_pb.Entry, key string) error {
+		seen = append(seen, key)
+		return nil
+	})
+	require.NoError(t, err)
+	assert.Equal(t, []string{"foo" + s3_constants.VersionsFolder}, seen)
+}
+
+func TestExpandVersionsDir_CoincidentallyNamedFolderRecursesViaFallback(t *testing.T) {
+	// A user-created folder happening to end in .versions/ has children
+	// without ExtVersionIdKey. expandVersionsDir must recurse via the
+	// fallback callback so inner files still emit normal events.
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, nil)
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				fileEntry("inner.txt"),
+				dirEntry("sub", nil),
+			},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder + "/sub": {fileEntry("deep.txt")},
+		},
+	}
+	var seen []string
+	cb := func(_ *filer_pb.Entry, key string) error {
+		seen = append(seen, key)
+		return nil
+	}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets"}
+	count, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, cb, nil)
+	require.NoError(t, err)
+	assert.Equal(t, 0, count, "fallback path returns 0 — events go through cb")
+	assert.ElementsMatch(t, []string{"foo" + s3_constants.VersionsFolder + "/inner.txt", "foo" + s3_constants.VersionsFolder + "/sub/deep.txt"}, seen)
+}
+
+func TestExpandVersionsDir_RaceWithMissingPointerStillExpands(t *testing.T) {
+	// Real .versions container whose parent metadata update hasn't
+	// landed yet (no ExtLatestVersionIdKey on the dir). Children DO
+	// carry ExtVersionIdKey. expandVersionsDir must still emit version
+	// events; missing-pointer fallback (newest-by-mtime as latest)
+	// covers retention safety.
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, nil) // no Extended at all
+	now := time.Now()
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", now.Add(-2*time.Hour), false),
+				versionFile("v2", now.Add(-1*time.Hour), false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	count, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, nil)
+	require.NoError(t, err)
+	assert.Equal(t, 2, count)
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	assert.True(t, byID["v2"].IsLatest, "newest by mtime is latest when pointer missing")
+}
+
+func TestExpandVersionsDir_LatestAndNoncurrentsByMtime(t *testing.T) {
+	now := time.Now()
+	v1mt := now.Add(-3 * time.Hour)
+	v2mt := now.Add(-2 * time.Hour)
+	v3mt := now.Add(-1 * time.Hour)
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v3"),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", v1mt, false),
+				versionFile("v2", v2mt, false),
+				versionFile("v3", v3mt, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+
+	count, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, nil)
+	require.NoError(t, err)
+	assert.Equal(t, 3, count)
+
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		require.NotNil(t, ev.BootstrapVersion)
+		assert.Equal(t, "foo", ev.BootstrapVersion.LogicalKey)
+		assert.Equal(t, 3, ev.BootstrapVersion.NumVersions)
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	require.Contains(t, byID, "v1")
+	require.Contains(t, byID, "v2")
+	require.Contains(t, byID, "v3")
+
+	assert.True(t, byID["v3"].IsLatest)
+	assert.True(t, byID["v3"].SuccessorModTime.IsZero(), "newest sibling has no successor")
+
+	assert.False(t, byID["v2"].IsLatest)
+	assert.Equal(t, 0, byID["v2"].NoncurrentIndex, "newest noncurrent")
+	assert.Equal(t, v3mt.Unix(), byID["v2"].SuccessorModTime.Unix())
+
+	assert.False(t, byID["v1"].IsLatest)
+	assert.Equal(t, 1, byID["v1"].NoncurrentIndex)
+	assert.Equal(t, v2mt.Unix(), byID["v1"].SuccessorModTime.Unix())
+}
+
+func TestExpandVersionsDir_LatestPointerOutOfOrderByMtime(t *testing.T) {
+	// Backdated PUT scenario: latest pointer names v1 but v1's mtime is
+	// OLDER than v2's. After newest-first sort the order is [v2, v1] so
+	// latestPos == 1, exercising the rank-skip path for the noncurrent.
+	now := time.Now()
+	v1mt := now.Add(-3 * time.Hour)
+	v2mt := now.Add(-1 * time.Hour)
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v1"),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", v1mt, false),
+				versionFile("v2", v2mt, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	_, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, nil)
+	require.NoError(t, err)
+
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	assert.True(t, byID["v1"].IsLatest)
+	assert.False(t, byID["v2"].IsLatest)
+	assert.Equal(t, 0, byID["v2"].NoncurrentIndex, "v2 is the only noncurrent → rank 0")
+}
+
+func TestExpandVersionsDir_MissingLatestPointerFallsBackToNewest(t *testing.T) {
+	// No latest pointer (rare race window): treat the newest sibling
+	// by mtime as latest so retention isn't unsafe.
+	now := time.Now()
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", now.Add(-2*time.Hour), false),
+				versionFile("v2", now.Add(-1*time.Hour), false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	_, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, nil)
+	require.NoError(t, err)
+
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	assert.True(t, byID["v2"].IsLatest, "newest by mtime is latest when pointer missing")
+	assert.False(t, byID["v1"].IsLatest)
+}
+
+func TestExpandVersionsDir_DeleteMarkerFlagPropagated(t *testing.T) {
+	now := time.Now()
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v-marker"),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v-marker", now, true),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	_, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, nil)
+	require.NoError(t, err)
+
+	events := inj.snapshot()
+	require.Len(t, events, 1)
+	assert.True(t, events[0].BootstrapVersion.IsDeleteMarker)
+	assert.True(t, events[0].BootstrapVersion.IsLatest)
+}
+
+func bareFile(name string, mtime time.Time) *filer_pb.Entry {
+	return &filer_pb.Entry{
+		Name: name,
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime: mtime.Unix(),
+		},
+	}
+}
+
+// suspendedNullFile mirrors the suspended-versioning write path: the
+// bare entry carries ExtVersionIdKey="null" so bootstrap can tell it
+// apart from a pre-versioning bare object during a pointer-missing
+// race window.
+func suspendedNullFile(name string, mtime time.Time) *filer_pb.Entry {
+	return &filer_pb.Entry{
+		Name: name,
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime: mtime.Unix(),
+		},
+		Extended: map[string][]byte{
+			s3_constants.ExtVersionIdKey: []byte("null"),
+		},
+	}
+}
+
+func TestExpandVersionsDir_PreVersioningNullIsNoncurrent(t *testing.T) {
+	// Object existed pre-versioning as the bare key. Versioning was
+	// enabled and a newer version v1 was PUT under .versions/. The
+	// .versions/ latest pointer names v1, so null is noncurrent.
+	now := time.Now()
+	v1mt := now.Add(-1 * time.Hour)
+	nullMt := now.Add(-3 * time.Hour)
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v1"),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot: {bareFile("foo", nullMt), versionsDir},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", v1mt, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	skipBare := map[string]bool{}
+	count, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, skipBare)
+	require.NoError(t, err)
+	assert.Equal(t, 2, count)
+
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	assert.True(t, byID["v1"].IsLatest)
+	assert.False(t, byID["null"].IsLatest)
+	assert.Equal(t, 0, byID["null"].NoncurrentIndex)
+	assert.Equal(t, 2, byID["null"].NumVersions)
+	assert.True(t, skipBare["foo"], "bare-key skip recorded")
+}
+
+func TestExpandVersionsDir_SuspendedNullIsCurrent(t *testing.T) {
+	// Suspended-bucket scenario: a write to the null version cleared the
+	// .versions/ latest pointer AND tagged the bare entry with
+	// ExtVersionIdKey="null". Older real versions remain in .versions/.
+	// Null must be IsLatest=true; .versions/ children become noncurrent.
+	now := time.Now()
+	v1mt := now.Add(-3 * time.Hour)
+	v2mt := now.Add(-2 * time.Hour)
+	nullMt := now.Add(-1 * time.Hour)
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{}) // pointer cleared
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot: {suspendedNullFile("foo", nullMt), versionsDir},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", v1mt, false),
+				versionFile("v2", v2mt, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	skipBare := map[string]bool{}
+	_, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, skipBare)
+	require.NoError(t, err)
+
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	assert.True(t, byID["null"].IsLatest, "pointer cleared + null exists -> null is latest")
+	assert.False(t, byID["v1"].IsLatest)
+	assert.False(t, byID["v2"].IsLatest)
+	assert.True(t, skipBare["foo"])
+}
+
+func TestExpandVersionsDir_NullVersionDirectoryKeyMarker(t *testing.T) {
+	// Directory-key marker (object name ends in /): the bare entry is a
+	// directory with Mime set. Treat as null version.
+	now := time.Now()
+	v1mt := now.Add(-1 * time.Hour)
+	dirMarker := &filer_pb.Entry{
+		Name:        "foo",
+		IsDirectory: true,
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime: now.Add(-3 * time.Hour).Unix(),
+			Mime:  "application/x-directory",
+		},
+	}
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v1"),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot: {dirMarker, versionsDir},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", v1mt, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	skipBare := map[string]bool{}
+	_, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, skipBare)
+	require.NoError(t, err)
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	require.Contains(t, byID, "null", "directory-key marker counts as null version")
+	assert.True(t, skipBare["foo"])
+}
+
+func TestWalkBucketDir_VersionsDirOrderingClaimsNullSibling(t *testing.T) {
+	// End-to-end through walkBucket: bare foo + foo.versions/ are
+	// siblings in the same directory. The two-pass walker processes
+	// the .versions/ first; expandVersionsDir claims "foo" as null;
+	// the second pass sees skipBare["foo"]==true and emits no regular
+	// event for the bare entry.
+	now := time.Now()
+	v1mt := now.Add(-1 * time.Hour)
+	nullMt := now.Add(-3 * time.Hour)
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v1"),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot: {bareFile("foo", nullMt), versionsDir},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", v1mt, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	b.KickOffNew(context.Background(), []string{"b1"})
+	// give the goroutine time to finish
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) && len(inj.snapshot()) < 2 {
+		time.Sleep(10 * time.Millisecond)
+	}
+	events := inj.snapshot()
+
+	// Exactly two events: v1 (latest) and null (noncurrent). NO third
+	// regular event for the bare "foo" entry.
+	assert.Len(t, events, 2)
+	versionIDs := []string{}
+	for _, ev := range events {
+		require.NotNil(t, ev.BootstrapVersion, "all events must be BootstrapVersion-tagged")
+		versionIDs = append(versionIDs, ev.BootstrapVersion.VersionID)
+	}
+	assert.ElementsMatch(t, []string{"v1", "null"}, versionIDs)
+}
+
+func TestExpandVersionsDir_VersionIDTiebreakOnSameSecondMtime(t *testing.T) {
+	// Two versions written in the same second: Mtime ties. The
+	// CompareVersionIds tiebreak puts the version_id with newer
+	// canonical ordering first. Use new-format IDs (inverted timestamps)
+	// so smaller string sorts as newer.
+	now := time.Now().Truncate(time.Second)
+	idNewer := "8000000000000000aaaaaaaaaaaaaaaa"
+	idOlder := "9000000000000000bbbbbbbbbbbbbbbb"
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte(idNewer),
+	})
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile(idOlder, now, false),
+				versionFile(idNewer, now, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	_, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, nil)
+	require.NoError(t, err)
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	assert.True(t, byID[idNewer].IsLatest, "newer canonical id wins the tiebreak")
+	assert.False(t, byID[idOlder].IsLatest)
+	assert.Equal(t, 0, byID[idOlder].NoncurrentIndex)
+}
+
+func TestExpandVersionsDir_PreVersioningNullDuringPointerRaceFallsBackToNewest(t *testing.T) {
+	// Pre-versioning bare object existed when versioning was enabled.
+	// A new version v1 was just written under .versions/<file> but the
+	// parent's ExtLatestVersionIdKey update has not landed yet. The
+	// bare entry has NO ExtVersionIdKey marker — distinguishing it from
+	// a suspended-bucket write. Bootstrap must treat v1 as latest (the
+	// newest sibling) and the implicit null as noncurrent, so the null
+	// expiration is scheduled this run instead of waiting for a future
+	// bootstrap.
+	now := time.Now()
+	v1mt := now.Add(-1 * time.Hour) // newer
+	nullMt := now.Add(-3 * time.Hour)
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{}) // pointer not yet written
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot: {bareFile("foo", nullMt), versionsDir},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v1", v1mt, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	skipBare := map[string]bool{}
+	_, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, skipBare)
+	require.NoError(t, err)
+
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	assert.True(t, byID["v1"].IsLatest, "newest sibling wins when null is implicit")
+	assert.False(t, byID["null"].IsLatest, "implicit null is noncurrent during pointer-missing race")
+}
+
+func TestExpandVersionsDir_SuspendedThenReEnabledNullIsNoncurrent(t *testing.T) {
+	// Bucket was suspended: bare entry was written with
+	// ExtVersionIdKey="null" and the .versions/ pointer cleared.
+	// Versioning was re-enabled and a fresh PUT created
+	// .versions/<v-new> with newer mtime, but the pointer-update for
+	// that new version hasn't landed yet. Bootstrap running in this
+	// window must keep v-new as latest (it's newest by mtime); the
+	// explicit null is noncurrent. Promoting the older null to latest
+	// just because it's explicit would skip current-version expiration
+	// of v-new and never schedule the null's noncurrent retention.
+	now := time.Now()
+	nullMt := now.Add(-3 * time.Hour) // OLDER bare-null
+	vNewMt := now.Add(-1 * time.Hour) // newer real version
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{}) // pointer not yet written
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot: {suspendedNullFile("foo", nullMt), versionsDir},
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: {
+				versionFile("v-new", vNewMt, false),
+			},
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	skipBare := map[string]bool{}
+	_, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, skipBare)
+	require.NoError(t, err)
+
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	assert.True(t, byID["v-new"].IsLatest, "newest sibling wins even when an older explicit null exists")
+	assert.False(t, byID["null"].IsLatest)
+	assert.Equal(t, 0, byID["null"].NoncurrentIndex)
+}
+
+func TestExpandVersionsDir_PaginatesBeyondListingLimit(t *testing.T) {
+	// The filer caps SeaweedList(..., limit=0) at DirListingLimit per
+	// call. Expanding a hot key with more versions than that limit
+	// would silently truncate, so the rank/sort math would be wrong
+	// past the boundary. listAll paginates via StartFromFileName.
+	// Shrink listPageSize so the test doesn't need thousands of entries.
+	prevPageSize := listPageSize
+	listPageSize = 2
+	t.Cleanup(func() { listPageSize = prevPageSize })
+
+	now := time.Now().Truncate(time.Second)
+	const total = 7
+	versionsDir := dirEntry("foo"+s3_constants.VersionsFolder, map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey: []byte("v0"),
+	})
+	versions := make([]*filer_pb.Entry, 0, total)
+	for i := 0; i < total; i++ {
+		// v0 is newest; v6 is oldest. Names are v_v00 .. v_v06 so the
+		// sort-by-name in the fake matches the sort-by-mtime here.
+		versions = append(versions, versionFile(fmt.Sprintf("v%02d", i), now.Add(-time.Duration(i)*time.Hour), false))
+	}
+	client := &fakeFilerClient{
+		tree: map[string][]*filer_pb.Entry{
+			testBucketRoot + "/foo" + s3_constants.VersionsFolder: versions,
+		},
+	}
+	inj := &recordingInjector{}
+	b := &BucketBootstrapper{FilerClient: client, BucketsPath: "/buckets", Injector: inj}
+	count, err := b.expandVersionsDir(context.Background(), "b1", testBucketRoot, "foo"+s3_constants.VersionsFolder, versionsDir, nil, nil)
+	require.NoError(t, err)
+	assert.Equal(t, total, count, "every page must reach the injector")
+
+	listed := client.listedCopy()
+	calls := 0
+	for _, d := range listed {
+		if d == testBucketRoot+"/foo"+s3_constants.VersionsFolder {
+			calls++
+		}
+	}
+	// Pages of 2 over 7 items = 4 calls (2+2+2+1). Loop exits once
+	// page count < listPageSize on the 4th call.
+	assert.Equal(t, 4, calls, "must paginate via StartFromFileName")
+
+	byID := map[string]*reader.BootstrapVersion{}
+	for _, ev := range inj.snapshot() {
+		byID[ev.BootstrapVersion.VersionID] = ev.BootstrapVersion
+	}
+	require.Len(t, byID, total, "no version dropped at a page boundary")
+	for i := 0; i < total; i++ {
+		bv := byID[fmt.Sprintf("v%02d", i)]
+		require.NotNil(t, bv)
+		assert.Equal(t, total, bv.NumVersions, "NumVersions reflects every page")
+	}
+	// v0 is the latest pointer target and the newest by mtime.
+	assert.True(t, byID["v00"].IsLatest)
+	// v6 is the oldest noncurrent.
+	assert.Equal(t, total-2, byID["v06"].NoncurrentIndex)
+}
+
+func TestWalkBucketDir_PaginatesBeyondListingLimit(t *testing.T) {
+	// Same correctness story for the bucket-level walk: hot buckets
+	// with thousands of objects must not silently truncate.
+	prevPageSize := listPageSize
+	listPageSize = 2
+	t.Cleanup(func() { listPageSize = prevPageSize })
+
+	const total = 5
+	rootChildren := make([]*filer_pb.Entry, 0, total)
+	for i := 0; i < total; i++ {
+		rootChildren = append(rootChildren, fileEntry(fmt.Sprintf("k%02d", i)))
+	}
+	client := &fakeFilerClient{tree: map[string][]*filer_pb.Entry{testBucketRoot: rootChildren}}
+	var seen []string
+	err := walkBucketDir(context.Background(), client, testBucketRoot, testBucketRoot, func(_ *filer_pb.Entry, key string) error {
+		seen = append(seen, key)
+		return nil
+	})
+	require.NoError(t, err)
+
+	want := make([]string, 0, total)
+	for i := 0; i < total; i++ {
+		want = append(want, fmt.Sprintf("k%02d", i))
+	}
+	assert.ElementsMatch(t, want, seen, "every page processed")
+
+	listed := client.listedCopy()
+	calls := 0
+	for _, d := range listed {
+		if d == testBucketRoot {
+			calls++
+		}
+	}
+	// Each directory is streamed twice (pass 1: .versions/, pass 2:
+	// everything else) to keep memory bounded on flat buckets. 5
+	// entries at page 2 = 3 paginated calls per pass = 6 total.
+	assert.Equal(t, 6, calls)
+}
diff --git a/weed/s3api/s3lifecycle/version_time.go b/weed/s3api/s3lifecycle/version_time.go
index a9d2e9ae2..5af924b7d 100644
--- a/weed/s3api/s3lifecycle/version_time.go
+++ b/weed/s3api/s3lifecycle/version_time.go
@@ -1,6 +1,76 @@
 package s3lifecycle
 
+import (
+	"strconv"
+)
+
 // versionIdFormatThreshold distinguishes old vs new format version IDs.
 // New format (inverted timestamps) produces values above this threshold;
 // old format (raw timestamps) produces values below it.
 const versionIdFormatThreshold = 0x4000000000000000
+
+// CompareVersionIds returns negative if a is newer than b, positive if b
+// is newer, 0 if equal. Mirrors compareVersionIds in s3api_version_id.go
+// (kept duplicated to avoid the s3api -> s3lifecycle import cycle). Used
+// as a tiebreak when version mtimes collide at second resolution.
+func CompareVersionIds(a, b string) int {
+	if a == b {
+		return 0
+	}
+	if a == "null" {
+		return 1 // null sorts last
+	}
+	if b == "null" {
+		return -1
+	}
+	aIsNew := isNewFormatVersionId(a)
+	bIsNew := isNewFormatVersionId(b)
+	if aIsNew == bIsNew {
+		// Same format. New format (inverted timestamps) sorts smaller=newer
+		// lexicographically; old format sorts smaller=older.
+		if aIsNew {
+			if a < b {
+				return -1
+			}
+			return 1
+		}
+		if a < b {
+			return 1
+		}
+		return -1
+	}
+	at := getVersionTimestamp(a)
+	bt := getVersionTimestamp(b)
+	if at > bt {
+		return -1
+	}
+	if at < bt {
+		return 1
+	}
+	return 0
+}
+
+func isNewFormatVersionId(versionId string) bool {
+	if len(versionId) < 16 || versionId == "null" {
+		return false
+	}
+	t, err := strconv.ParseUint(versionId[:16], 16, 64)
+	if err != nil {
+		return false
+	}
+	return t > versionIdFormatThreshold
+}
+
+func getVersionTimestamp(versionId string) int64 {
+	if len(versionId) < 16 || versionId == "null" {
+		return 0
+	}
+	t, err := strconv.ParseUint(versionId[:16], 16, 64)
+	if err != nil {
+		return 0
+	}
+	if t > versionIdFormatThreshold {
+		return int64(^uint64(0)>>1) - int64(t)
+	}
+	return int64(t)
+}