diff --git a/cmd/util/common/checkpoint_test.go b/cmd/util/common/checkpoint_test.go
new file mode 100644
index 00000000000..80fd32c6203
--- /dev/null
+++ b/cmd/util/common/checkpoint_test.go
@@ -0,0 +1,88 @@
+package common
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/onflow/flow-go/ledger/complete/wal"
+	modelbootstrap "github.com/onflow/flow-go/model/bootstrap"
+	"github.com/onflow/flow-go/utils/unittest"
+)
+
+// TestFindLatestCheckpointFilePath verifies that the returned file name matches the version of the
+// latest checkpoint in the directory. V6 and V7 checkpoints coexist in a payloadless triedir, so
+// rendering the wrong version's name would point at a file that does not exist.
+func TestFindLatestCheckpointFilePath(t *testing.T) {
+	v6Root := modelbootstrap.FilenameWALRootCheckpoint
+	v7Root := modelbootstrap.FilenameWALRootCheckpoint + wal.V7FileSuffix
+
+	tests := []struct {
+		name     string
+		files    []string
+		expected string
+	}{
+		{
+			name:     "empty directory falls back to the V6 root checkpoint",
+			files:    nil,
+			expected: v6Root,
+		},
+		{
+			name:     "V6 root checkpoint only",
+			files:    []string{v6Root},
+			expected: v6Root,
+		},
+		{
+			name: "V7 root checkpoint is preferred over the V6 root checkpoint",
+			// this is the state of a payloadless triedir right after bootstrap: the V6 root
+			// checkpoint copied from the bootstrap folder, plus its V7 conversion
+			files:    []string{v6Root, v7Root},
+			expected: v7Root,
+		},
+		{
+			name:     "numbered V6 checkpoint",
+			files:    []string{v6Root, wal.NumberToFilename(10)},
+			expected: wal.NumberToFilename(10),
+		},
+		{
+			name:     "numbered V7 checkpoint",
+			files:    []string{v6Root, v7Root, wal.NumberToFilenameV7(10)},
+			expected: wal.NumberToFilenameV7(10),
+		},
+		{
+			name:     "highest number wins across versions",
+			files:    []string{wal.NumberToFilename(20), wal.NumberToFilenameV7(10)},
+			expected: wal.NumberToFilename(20),
+		},
+		{
+			name:     "V7 wins over V6 at the same number",
+			files:    []string{wal.NumberToFilename(10), wal.NumberToFilenameV7(10)},
+			expected: wal.NumberToFilenameV7(10),
+		},
+		{
+			name: "numbered checkpoints win over the root checkpoint",
+			files: []string{
+				v6Root, v7Root,
+				wal.NumberToFilename(10), wal.NumberToFilenameV7(10),
+				wal.NumberToFilenameV7(11),
+			},
+			expected: wal.NumberToFilenameV7(11),
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			unittest.RunWithTempDir(t, func(dir string) {
+				for _, name := range tc.files {
+					require.NoError(t, os.WriteFile(filepath.Join(dir, name), []byte{}, 0644))
+				}
+
+				checkpointFilePath, err := findLatestCheckpointFilePath(dir)
+				require.NoError(t, err)
+				require.Equal(t, filepath.Join(dir, tc.expected), checkpointFilePath)
+			})
+		})
+	}
+}
diff --git a/ledger/complete/ledger.go b/ledger/complete/ledger.go
index fe2bbd0808d..c19833d0b64 100644
--- a/ledger/complete/ledger.go
+++ b/ledger/complete/ledger.go
@@ -335,15 +335,6 @@ func (l *Ledger) Trie(rootHash ledger.RootHash) (*trie.MTrie, error) {
 	return l.forest.GetTrie(rootHash)
 }
 
-// Checkpointer returns a checkpointer instance
-func (l *Ledger) Checkpointer() (*realWAL.Checkpointer, error) {
-	checkpointer, err := l.wal.NewCheckpointer()
-	if err != nil {
-		return nil, fmt.Errorf("cannot create checkpointer for compactor: %w", err)
-	}
-	return checkpointer, nil
-}
-
 func (l *Ledger) MigrateAt(
 	state ledger.State,
 	migration ledger.Migration,
diff --git a/ledger/complete/ledger_with_compactor.go b/ledger/complete/ledger_with_compactor.go
index 7a07d65c8e1..5fe106012c8 100644
--- a/ledger/complete/ledger_with_compactor.go
+++ b/ledger/complete/ledger_with_compactor.go
@@ -34,8 +34,6 @@ func NewLedgerWithCompactor(
 	logger zerolog.Logger,
 	pathFinderVersion uint8,
 ) (*LedgerWithCompactor, error) {
-	logger = logger.With().Str("ledger_mod", "complete").Logger()
-
 	// Create the ledger
 	l, err := NewLedger(diskWAL, ledgerCapacity, metrics, logger, pathFinderVersion)
 	if err != nil {
diff --git a/ledger/complete/payloadless/flattener_test.go b/ledger/complete/payloadless/flattener_test.go
new file mode 100644
index 00000000000..651697afdee
--- /dev/null
+++ b/ledger/complete/payloadless/flattener_test.go
@@ -0,0 +1,139 @@
+package payloadless
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/onflow/flow-go/ledger"
+	"github.com/onflow/flow-go/ledger/common/hash"
+	"github.com/onflow/flow-go/ledger/common/testutils"
+)
+
+// noChildNodes is the getNode callback for reading leaf nodes, which never reference children.
+func noChildNodes(t *testing.T) func(nodeIndex uint64) (*Node, error) {
+	return func(nodeIndex uint64) (*Node, error) {
+		require.FailNow(t, "leaf node must not resolve child nodes", "index %d", nodeIndex)
+		return nil, nil
+	}
+}
+
+// TestEncodeDecodeLeafNodeWithLeafHash covers the `leafHashPresent` encoding: an allocated
+// register's leaf keeps its leaf hash across a round trip, and the encoding is 100 bytes.
+func TestEncodeDecodeLeafNodeWithLeafHash(t *testing.T) {
+	path := testutils.PathByUint8(7)
+	leaf := NewLeaf(path, []byte("register value"), 256)
+	require.NotNil(t, leaf.LeafHash(), "sanity check: an allocated register leaf has a leaf hash")
+
+	scratch := make([]byte, 1024)
+	encoded := EncodeNode(leaf, 0, 0, scratch)
+
+	// node type (1) + height (2) + hash (32) + path (32) + leaf hash flag (1) + leaf hash (32)
+	require.Len(t, encoded, 100)
+	require.Equal(t, leafHashPresent, encoded[encNodeTypeSize+encHeightSize+encHashSize+encPathSize])
+
+	decoded, err := ReadNode(bytes.NewReader(encoded), make([]byte, 1024), noChildNodes(t))
+	require.NoError(t, err)
+	require.Equal(t, leaf.Height(), decoded.Height())
+	require.Equal(t, leaf.Hash(), decoded.Hash())
+	require.Equal(t, *leaf.Path(), *decoded.Path())
+	require.NotNil(t, decoded.LeafHash())
+	require.Equal(t, *leaf.LeafHash(), *decoded.LeafHash())
+}
+
+// TestEncodeDecodeLeafNodeWithoutLeafHash covers the `leafHashAbsent` encoding, which is the one
+// on-disk mechanism V7 adds over V6. A leaf for an unallocated register has no leaf hash, so the
+// flag byte is the only thing recording its absence, and the encoding is 32 bytes shorter.
+func TestEncodeDecodeLeafNodeWithoutLeafHash(t *testing.T) {
+	path := testutils.PathByUint8(7)
+
+	// An unallocated register (empty value) yields a default leaf, whose leaf hash is nil.
+	leaf := NewLeaf(path, nil, 256)
+	require.True(t, leaf.IsLeaf())
+	require.Nil(t, leaf.LeafHash(), "sanity check: an unallocated register leaf has no leaf hash")
+
+	scratch := make([]byte, 1024)
+	encoded := EncodeNode(leaf, 0, 0, scratch)
+
+	// node type (1) + height (2) + hash (32) + path (32) + leaf hash flag (1), and no leaf hash
+	require.Len(t, encoded, 68)
+	require.Equal(t, leafHashAbsent, encoded[encNodeTypeSize+encHeightSize+encHashSize+encPathSize])
+
+	decoded, err := ReadNode(bytes.NewReader(encoded), make([]byte, 1024), noChildNodes(t))
+	require.NoError(t, err)
+	require.Equal(t, leaf.Height(), decoded.Height())
+	require.Equal(t, leaf.Hash(), decoded.Hash())
+	require.Equal(t, *leaf.Path(), *decoded.Path())
+	require.Nil(t, decoded.LeafHash(), "absent leaf hash must decode back to nil")
+}
+
+// TestReadNodeRejectsInvalidLeafHashFlag verifies that a leaf hash flag other than
+// `leafHashAbsent` or `leafHashPresent` is reported as an error rather than silently
+// interpreted, so a corrupted checkpoint cannot be read as a valid trie.
+func TestReadNodeRejectsInvalidLeafHashFlag(t *testing.T) {
+	leaf := NewLeaf(testutils.PathByUint8(7), []byte("register value"), 256)
+
+	encoded := EncodeNode(leaf, 0, 0, make([]byte, 1024))
+	flagPos := encNodeTypeSize + encHeightSize + encHashSize + encPathSize
+
+	for _, flag := range []byte{2, 0xff} {
+		corrupted := make([]byte, len(encoded))
+		copy(corrupted, encoded)
+		corrupted[flagPos] = flag
+
+		_, err := ReadNode(bytes.NewReader(corrupted), make([]byte, 1024), noChildNodes(t))
+		require.Error(t, err)
+		require.ErrorContains(t, err, "invalid leaf hash flag")
+	}
+}
+
+// TestReadNodeRejectsTruncatedLeafHash verifies that a leaf whose flag promises a leaf hash but
+// whose bytes are cut short is reported as an error.
+func TestReadNodeRejectsTruncatedLeafHash(t *testing.T) {
+	leaf := NewLeaf(testutils.PathByUint8(7), []byte("register value"), 256)
+
+	encoded := EncodeNode(leaf, 0, 0, make([]byte, 1024))
+	// drop the last byte of the leaf hash
+	truncated := encoded[:len(encoded)-1]
+
+	_, err := ReadNode(bytes.NewReader(truncated), make([]byte, 1024), noChildNodes(t))
+	require.Error(t, err)
+	require.ErrorContains(t, err, "cannot read leaf hash")
+}
+
+// TestEncodeDecodeLeafNodeSmallScratch verifies both leaf encodings are correct when the scratch
+// buffer is too small to hold the node, i.e. when the encoder and decoder allocate instead.
+func TestEncodeDecodeLeafNodeSmallScratch(t *testing.T) {
+	leaves := map[string]*Node{
+		"leaf hash present": NewLeaf(testutils.PathByUint8(7), []byte("register value"), 256),
+		"leaf hash absent":  NewLeaf(testutils.PathByUint8(7), nil, 256),
+	}
+
+	for name, leaf := range leaves {
+		t.Run(name, func(t *testing.T) {
+			encoded := EncodeNode(leaf, 0, 0, nil)
+
+			decoded, err := ReadNode(bytes.NewReader(encoded), nil, noChildNodes(t))
+			require.NoError(t, err)
+			require.Equal(t, leaf.Hash(), decoded.Hash())
+			require.Equal(t, leaf.LeafHash() == nil, decoded.LeafHash() == nil)
+		})
+	}
+}
+
+// TestEncodeDecodeLeafNodeWithZeroLeafHash guards against a flag-free encoding: an all-zero leaf
+// hash is a legitimate value and must not be confused with an absent one.
+func TestEncodeDecodeLeafNodeWithZeroLeafHash(t *testing.T) {
+	var zeroLeafHash hash.Hash
+	path := testutils.PathByUint8(7)
+	leaf := NewNode(256, nil, nil, path, &zeroLeafHash, ledger.GetDefaultHashForHeight(0))
+
+	encoded := EncodeNode(leaf, 0, 0, make([]byte, 1024))
+	require.Len(t, encoded, 100)
+
+	decoded, err := ReadNode(bytes.NewReader(encoded), make([]byte, 1024), noChildNodes(t))
+	require.NoError(t, err)
+	require.NotNil(t, decoded.LeafHash(), "a zero leaf hash is present, not absent")
+	require.Equal(t, zeroLeafHash, *decoded.LeafHash())
+}
diff --git a/ledger/complete/payloadless_compactor.go b/ledger/complete/payloadless_compactor.go
index 19ee2f7046c..b668a95945c 100644
--- a/ledger/complete/payloadless_compactor.go
+++ b/ledger/complete/payloadless_compactor.go
@@ -239,58 +239,6 @@ Loop:
 	}
 }
 
-// processTrieUpdate writes the WAL record, tracks the active segment, hands
-// the newly-built trie to the queue, and signals when enough segments have
-// rolled over to checkpoint. Mirrors [Compactor.processTrieUpdate].
-func (c *PayloadlessCompactor) processTrieUpdate(
-	update *WALPayloadlessTrieUpdate,
-	trieQueue *realWAL.PayloadlessTrieQueue,
-	activeSegmentNum int,
-	nextCheckpointNum int,
-) (_activeSegmentNum int, checkpointNum int, checkpointTries []*payloadless.MTrie) {
-
-	segmentNum, skipped, updateErr := c.wal.RecordUpdate(update.Update)
-	update.ResultCh <- updateErr
-
-	defer func() {
-		// Receive the freshly-built trie from the ledger goroutine and stage it.
-		trie := <-update.TrieCh
-		if trie == nil {
-			c.logger.Error().Msg("payloadless compactor failed to get updated trie")
-			return
-		}
-		trieQueue.Push(trie)
-	}()
-
-	if activeSegmentNum == -1 {
-		return segmentNum, -1, nil
-	}
-
-	if updateErr != nil || skipped || segmentNum == activeSegmentNum {
-		return activeSegmentNum, -1, nil
-	}
-
-	// segmentNum > activeSegmentNum — a segment just rolled over.
-
-	if segmentNum != activeSegmentNum+1 {
-		c.logger.Error().Msgf("payloadless compactor got unexpected new segment %d, want %d", segmentNum, activeSegmentNum+1)
-	}
-
-	prevSegmentNum := activeSegmentNum
-	activeSegmentNum = segmentNum
-
-	c.logger.Info().Msgf("finish writing segment file %v, payloadless trie update writing to segment %v; checkpoint triggers at segment %v",
-		prevSegmentNum, activeSegmentNum, nextCheckpointNum)
-
-	if nextCheckpointNum > prevSegmentNum {
-		return activeSegmentNum, -1, nil
-	}
-
-	// nextCheckpointNum == prevSegmentNum — enough segments accumulated.
-	tries := trieQueue.Tries()
-	return activeSegmentNum, nextCheckpointNum, tries
-}
-
 // checkpoint serializes a V7 checkpoint, then prunes older V7 files per the
 // retention policy, and notifies observers.
 func (c *PayloadlessCompactor) checkpoint(ctx context.Context, tries []*payloadless.MTrie, checkpointNum int) error {
@@ -370,6 +318,58 @@ func cleanupCheckpointsV7(checkpointer *realWAL.Checkpointer, checkpointsToKeep
 	return nil
 }
 
+// processTrieUpdate writes the WAL record, tracks the active segment, hands
+// the newly-built trie to the queue, and signals when enough segments have
+// rolled over to checkpoint. Mirrors [Compactor.processTrieUpdate].
+func (c *PayloadlessCompactor) processTrieUpdate(
+	update *WALPayloadlessTrieUpdate,
+	trieQueue *realWAL.PayloadlessTrieQueue,
+	activeSegmentNum int,
+	nextCheckpointNum int,
+) (_activeSegmentNum int, checkpointNum int, checkpointTries []*payloadless.MTrie) {
+
+	segmentNum, skipped, updateErr := c.wal.RecordUpdate(update.Update)
+	update.ResultCh <- updateErr
+
+	defer func() {
+		// Receive the freshly-built trie from the ledger goroutine and stage it.
+		trie := <-update.TrieCh
+		if trie == nil {
+			c.logger.Error().Msg("payloadless compactor failed to get updated trie")
+			return
+		}
+		trieQueue.Push(trie)
+	}()
+
+	if activeSegmentNum == -1 {
+		return segmentNum, -1, nil
+	}
+
+	if updateErr != nil || skipped || segmentNum == activeSegmentNum {
+		return activeSegmentNum, -1, nil
+	}
+
+	// segmentNum > activeSegmentNum — a segment just rolled over.
+
+	if segmentNum != activeSegmentNum+1 {
+		c.logger.Error().Msgf("payloadless compactor got unexpected new segment %d, want %d", segmentNum, activeSegmentNum+1)
+	}
+
+	prevSegmentNum := activeSegmentNum
+	activeSegmentNum = segmentNum
+
+	c.logger.Info().Msgf("finish writing segment file %v, payloadless trie update writing to segment %v; checkpoint triggers at segment %v",
+		prevSegmentNum, activeSegmentNum, nextCheckpointNum)
+
+	if nextCheckpointNum > prevSegmentNum {
+		return activeSegmentNum, -1, nil
+	}
+
+	// nextCheckpointNum == prevSegmentNum — enough segments accumulated.
+	tries := trieQueue.Tries()
+	return activeSegmentNum, nextCheckpointNum, tries
+}
+
 // latestV7CheckpointNum returns the highest V7 checkpoint number on disk,
 // or -1 if none exist or listing fails (with the error logged).
 func latestV7CheckpointNum(checkpointer *realWAL.Checkpointer, logger zerolog.Logger) int {
diff --git a/ledger/complete/wal/checkpointer.go b/ledger/complete/wal/checkpointer.go
index f162cbc8ef8..397b0fe16e8 100644
--- a/ledger/complete/wal/checkpointer.go
+++ b/ledger/complete/wal/checkpointer.go
@@ -886,6 +886,67 @@ func (c *Checkpointer) LoadRootCheckpointV7() ([]*payloadless.MTrie, error) {
 	return OpenAndReadCheckpointV7(c.dir, fileName, c.wal.log)
 }
 
+// LoadLatestCheckpointV7 loads the most recent usable V7 (payloadless) checkpoint from
+// the WAL directory and returns its tries together with the number of the loaded
+// checkpoint.
+//
+// It tries the newest numbered V7 checkpoint first, falling back to older ones if
+// a checkpoint file fails to load. This mirrors the V6 checkpoint selection in
+// [DiskWAL.replay]. The returned `loadedCheckpoint` is the number of the numbered
+// checkpoint that was loaded, used by callers to determine the first WAL segment
+// to replay.
+//
+// When no numbered V7 checkpoint is usable, it falls back to the V7 root
+// checkpoint (converted from the V6 root.checkpoint during bootstrap), if present.
+// In that case, and when no V7 checkpoint of either kind exists, `loadedCheckpoint`
+// is -1, signalling that all segments must be replayed on top of the returned
+// tries (which is the empty slice when no checkpoint exists at all).
+//
+// No error returns are expected during normal operation.
+func (c *Checkpointer) LoadLatestCheckpointV7() (tries []*payloadless.MTrie, loadedCheckpoint int, err error) {
+	checkpoints, err := c.CheckpointsV7()
+	if err != nil {
+		return nil, -1, fmt.Errorf("cannot list V7 checkpoints: %w", err)
+	}
+
+	// Try the newest V7 checkpoint first, falling back to older ones if a file
+	// fails to load. This mirrors the V6 checkpoint selection in [DiskWAL.replay].
+	for i := len(checkpoints) - 1; i >= 0; i-- {
+		num := checkpoints[i]
+		name := NumberToFilenameV7(num)
+		tries, err := OpenAndReadCheckpointV7(c.dir, name, c.wal.log)
+		if err != nil {
+			c.wal.log.Warn().Int("checkpoint", num).Err(err).
+				Msg("V7 checkpoint loading failed; falling back to older checkpoint")
+			continue
+		}
+		c.wal.log.Info().Int("checkpoint", num).Int("trie_count", len(tries)).
+			Msg("loaded V7 checkpoint")
+		return tries, num, nil
+	}
+
+	// No numbered V7 checkpoint loaded: fall back to the V7 root checkpoint, if
+	// present. This is the payloadless analog of the root-checkpoint branch in
+	// [DiskWAL.replay]; like that branch it does not advance the replay start
+	// (loadedCheckpoint stays -1), so all segments are replayed on top of the
+	// root state.
+	hasV7Root, err := c.HasRootCheckpointV7()
+	if err != nil {
+		return nil, -1, fmt.Errorf("cannot check for V7 root checkpoint: %w", err)
+	}
+	if hasV7Root {
+		tries, err := c.LoadRootCheckpointV7()
+		if err != nil {
+			return nil, -1, fmt.Errorf("failed to load V7 root checkpoint: %w", err)
+		}
+		c.wal.log.Info().Int("trie_count", len(tries)).
+			Msg("loaded V7 root checkpoint")
+		return tries, -1, nil
+	}
+
+	return nil, -1, nil
+}
+
 func (c *Checkpointer) HasRootCheckpoint() (bool, error) {
 	return HasRootCheckpoint(c.dir)
 }
@@ -916,6 +977,13 @@ func HasRootCheckpointV7(dir string) (bool, error) {
 	}
 }
 
+// RootCheckpointFilenameV7 returns the filename (not the full path) of the V7
+// (payloadless) root checkpoint: [bootstrap.FilenameWALRootCheckpoint] with the
+// [V7FileSuffix] appended.
+func RootCheckpointFilenameV7() string {
+	return bootstrap.FilenameWALRootCheckpoint + V7FileSuffix
+}
+
 // RemoveCheckpoint deletes both the V6 and the V7 part files for the given checkpoint number.
 // Deleting a version that isn't present is not an error, so this reports a failure whenever
 // either deletion fails.
diff --git a/ledger/complete/wal/payloadless_replay_test.go b/ledger/complete/wal/payloadless_replay_test.go
new file mode 100644
index 00000000000..9be58d6f7ff
--- /dev/null
+++ b/ledger/complete/wal/payloadless_replay_test.go
@@ -0,0 +1,121 @@
+package wal
+
+import (
+	"os"
+	"path"
+	"testing"
+
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
+
+	"github.com/onflow/flow-go/ledger"
+	"github.com/onflow/flow-go/ledger/complete/mtrie"
+	"github.com/onflow/flow-go/ledger/complete/payloadless"
+	"github.com/onflow/flow-go/model/bootstrap"
+	"github.com/onflow/flow-go/module/metrics"
+	"github.com/onflow/flow-go/utils/unittest"
+)
+
+// TestReplayOnPayloadlessForest_IgnoresV6RootCheckpoint is a regression test for
+// the case where a payloadless node boots with both a V7 root checkpoint (the
+// real seed) and a V6 root.checkpoint present in the trie dir. The forest must
+// be seeded from the V7 checkpoint, and the V6 root.checkpoint must NOT be read.
+//
+// To prove the V6 file is never touched, a corrupt root.checkpoint is placed
+// alongside the V7 checkpoint: the previous implementation routed payloadless
+// segment replay through [DiskWAL.replay], which falls back to loading the V6
+// root checkpoint when replaying from segment 0 — that fallback would fail on
+// the corrupt file. With the fix, the V6 file is ignored and replay succeeds.
+func TestReplayOnPayloadlessForest_IgnoresV6RootCheckpoint(t *testing.T) {
+	unittest.RunWithTempDir(t, func(dir string) {
+		logger := zerolog.Nop()
+
+		// Build a V7 root checkpoint from a simple trie and write it as the
+		// payloadless root checkpoint (root.checkpoint.v7).
+		v6Tries := createSimpleTrie(t)
+		rootHash := v6Tries[0].RootHash()
+		v7Tries, err := FromV6Tries(v6Tries)
+		require.NoError(t, err)
+		require.NoError(t, StoreCheckpointV7Concurrently(v7Tries, dir, RootCheckpointFilenameV7(), logger))
+
+		// Place a corrupt V6 root checkpoint next to the V7 one. If the
+		// payloadless replay path attempts to load it, the load fails — which is
+		// exactly the regression this test guards against.
+		junkPath := path.Join(dir, bootstrap.FilenameWALRootCheckpoint)
+		require.NoError(t, os.WriteFile(junkPath, []byte("not a valid v6 checkpoint"), 0644))
+
+		w, err := NewDiskWAL(logger, nil, metrics.NewNoopCollector(), dir, 10, pathByteSize, segmentSize)
+		require.NoError(t, err)
+		defer func() { <-w.Done() }()
+
+		forest, err := payloadless.NewForest(100, &metrics.NoopCollector{}, nil)
+		require.NoError(t, err)
+
+		err = w.ReplayOnPayloadlessForest(forest)
+		require.NoError(t, err, "replay must seed from V7 and must not load the V6 root checkpoint")
+
+		require.True(t, forest.HasTrie(rootHash), "forest must be seeded from the V7 root checkpoint")
+	})
+}
+
+// TestReplayOnPayloadlessForest_ReplaysWALSegments verifies that after seeding
+// the forest from the V7 root checkpoint, WAL segment records that are newer
+// than the checkpoint are still replayed onto the payloadless forest. This
+// guards against the segment-replay refactor accidentally skipping segments.
+func TestReplayOnPayloadlessForest_ReplaysWALSegments(t *testing.T) {
+	unittest.RunWithTempDir(t, func(dir string) {
+		logger := zerolog.Nop()
+
+		// Seed state: a full forest with an initial update, captured as the V7
+		// root checkpoint.
+		fullForest, err := mtrie.NewForest(100, &metrics.NoopCollector{}, nil)
+		require.NoError(t, err)
+
+		paths0, payloads0 := randNPathPayloads(10)
+		seed := &ledger.TrieUpdate{
+			RootHash: fullForest.GetEmptyRootHash(),
+			Paths:    paths0,
+			Payloads: toPayloadPtrs(payloads0),
+		}
+		root0, err := fullForest.Update(seed)
+		require.NoError(t, err)
+
+		v6Tries, err := fullForest.GetTries()
+		require.NoError(t, err)
+		v7Tries, err := FromV6Tries(v6Tries)
+		require.NoError(t, err)
+		require.NoError(t, StoreCheckpointV7Concurrently(v7Tries, dir, RootCheckpointFilenameV7(), logger))
+
+		// A second update, built on root0, recorded into the WAL but NOT in the
+		// checkpoint. Replay must apply it to reach root1.
+		paths1, payloads1 := randNPathPayloads(10)
+		update1 := &ledger.TrieUpdate{
+			RootHash: root0,
+			Paths:    paths1,
+			Payloads: toPayloadPtrs(payloads1),
+		}
+		root1, err := fullForest.Update(update1)
+		require.NoError(t, err)
+
+		// Record update1 into the WAL, then close to flush the segment to disk.
+		recordWAL, err := NewDiskWAL(logger, nil, metrics.NewNoopCollector(), dir, 10, pathByteSize, segmentSize)
+		require.NoError(t, err)
+		_, _, err = recordWAL.RecordUpdate(update1)
+		require.NoError(t, err)
+		<-recordWAL.Done()
+
+		// Replay on a fresh WAL: seed from V7 (root0), then replay the WAL
+		// segment carrying update1 to reach root1.
+		w, err := NewDiskWAL(logger, nil, metrics.NewNoopCollector(), dir, 10, pathByteSize, segmentSize)
+		require.NoError(t, err)
+		defer func() { <-w.Done() }()
+
+		forest, err := payloadless.NewForest(100, &metrics.NoopCollector{}, nil)
+		require.NoError(t, err)
+
+		require.NoError(t, w.ReplayOnPayloadlessForest(forest))
+
+		require.True(t, forest.HasTrie(root0), "forest must contain the V7 checkpoint root")
+		require.True(t, forest.HasTrie(root1), "forest must contain the root produced by replaying the WAL segment")
+	})
+}
diff --git a/ledger/complete/wal/wal.go b/ledger/complete/wal/wal.go
index b8d24d93dee..2cbcd78e594 100644
--- a/ledger/complete/wal/wal.go
+++ b/ledger/complete/wal/wal.go
@@ -153,52 +153,13 @@ func (w *DiskWAL) ReplayOnPayloadlessForest(forest *payloadless.Forest) error {
 		return fmt.Errorf("cannot create checkpointer: %w", err)
 	}
 
-	checkpoints, err := checkpointer.CheckpointsV7()
+	tries, loadedCheckpoint, err := checkpointer.LoadLatestCheckpointV7()
 	if err != nil {
-		return fmt.Errorf("cannot list V7 checkpoints: %w", err)
+		return fmt.Errorf("cannot load latest V7 checkpoint: %w", err)
 	}
 
-	// Try the newest V7 checkpoint first, falling back to older ones if a file
-	// fails to load. This mirrors the V6 checkpoint selection in [DiskWAL.replay].
-	loadedCheckpoint := -1
-	for i := len(checkpoints) - 1; i >= 0; i-- {
-		num := checkpoints[i]
-		name := NumberToFilenameV7(num)
-		tries, err := OpenAndReadCheckpointV7(checkpointer.Dir(), name, w.log)
-		if err != nil {
-			w.log.Warn().Int("checkpoint", num).Err(err).
-				Msg("V7 checkpoint loading failed; falling back to older checkpoint")
-			continue
-		}
-		if err := forest.AddTries(tries); err != nil {
-			return fmt.Errorf("failed to seed payloadless forest from V7 checkpoint %s: %w", name, err)
-		}
-		w.log.Info().Int("checkpoint", num).Int("trie_count", len(tries)).
-			Msg("payloadless forest seeded from V7 checkpoint")
-		loadedCheckpoint = num
-		break
-	}
-
-	// No numbered V7 checkpoint loaded: fall back to the V7 root checkpoint, if
-	// present. This is the payloadless analog of the root-checkpoint branch in
-	// [DiskWAL.replay]; like that branch it does not advance the replay start, so
-	// all segments are replayed on top of the root state.
-	if loadedCheckpoint == -1 {
-		hasV7Root, err := checkpointer.HasRootCheckpointV7()
-		if err != nil {
-			return fmt.Errorf("cannot check for V7 root checkpoint: %w", err)
-		}
-		if hasV7Root {
-			tries, err := checkpointer.LoadRootCheckpointV7()
-			if err != nil {
-				return fmt.Errorf("failed to load V7 root checkpoint: %w", err)
-			}
-			if err := forest.AddTries(tries); err != nil {
-				return fmt.Errorf("failed to seed payloadless forest from V7 root checkpoint: %w", err)
-			}
-			w.log.Info().Int("trie_count", len(tries)).
-				Msg("payloadless forest seeded from V7 root checkpoint")
-		}
+	if err := forest.AddTries(tries); err != nil {
+		return fmt.Errorf("failed to seed payloadless forest from V7 checkpoint: %w", err)
 	}
 
 	return w.replaySegmentsForPayloadlessForest(forest, loadedCheckpoint)
@@ -235,14 +196,19 @@ func (w *DiskWAL) replaySegmentsForPayloadlessForest(
 		// V7 checkpoint already covers everything on disk.
 		return nil
 	}
-	err = w.replay(from, lastSeg,
-		func(tries []*trie.MTrie) error { return nil }, // unused when useCheckpoints=false
+	// Replay only the WAL segment records onto the forest. Unlike
+	// [DiskWAL.replay], this deliberately does NOT fall back to loading the V6
+	// root checkpoint when `from` is 0: the payloadless forest is already seeded
+	// from the V7 checkpoint by the caller ([DiskWAL.ReplayOnPayloadlessForest]),
+	// and the V6 root checkpoint is not loadable into a payloadless forest.
+	// Routing through replay would read (and immediately discard) the entire V6
+	// root checkpoint, a wasteful full-forest load at boot.
+	err = w.replaySegments(from, lastSeg,
 		func(update *ledger.TrieUpdate) error {
 			_, err := forest.Update(update)
 			return err
 		},
 		func(rootHash ledger.RootHash) error { return nil },
-		false, // useCheckpoints
 	)
 	if err != nil {
 		return fmt.Errorf("could not replay WAL segments [%v:%v] for payloadless forest: %w", from, lastSeg, err)
@@ -413,9 +379,31 @@ func (w *DiskWAL) replay(
 		Int("loaded_checkpoint", loadedCheckpoint).
 		Msgf("replaying segments from %d to %d", startSegment, to)
 
+	err = w.replaySegments(startSegment, to, updateFn, deleteFn)
+	if err != nil {
+		return err
+	}
+
+	w.log.Info().Msgf("finished loading checkpoint and replaying WAL from %d to %d", from, to)
+
+	return nil
+}
+
+// replaySegments reads the WAL segment records in the range [from, to] and
+// applies each record to the provided handlers, dispatching WALUpdate records
+// to `updateFn` and WALDelete records to `deleteFn`. It performs NO checkpoint
+// loading: the caller is responsible for seeding any starting state before
+// calling this.
+//
+// No error returns are expected during normal operation.
+func (w *DiskWAL) replaySegments(
+	from, to int,
+	updateFn func(update *ledger.TrieUpdate) error,
+	deleteFn func(rootHash ledger.RootHash) error,
+) error {
 	sr, err := prometheusWAL.NewSegmentsRangeReader(w.log, prometheusWAL.SegmentRange{
 		Dir:   w.wal.Dir(),
-		First: startSegment,
+		First: from,
 		Last:  to,
 	})
 	if err != nil {
@@ -445,14 +433,14 @@ func (w *DiskWAL) replay(
 				return fmt.Errorf("error while processing LedgerWAL deletion: %w", err)
 			}
 		}
-
-		err = reader.Err()
-		if err != nil {
-			return fmt.Errorf("cannot read LedgerWAL: %w", err)
-		}
 	}
 
-	w.log.Info().Msgf("finished loading checkpoint and replaying WAL from %d to %d", from, to)
+	// reader.Next() returns false both on clean EOF and on a read error, so the error
+	// must be checked after the loop to detect a corrupt or truncated final record.
+	err = reader.Err()
+	if err != nil {
+		return fmt.Errorf("cannot read LedgerWAL: %w", err)
+	}
 
 	return nil
 }