diff --git a/cmd/util/cmd/checkpoint-convert-v6/cmd.go b/cmd/util/cmd/checkpoint-convert-v6/cmd.go
new file mode 100644
index 00000000000..4bce42ce904
--- /dev/null
+++ b/cmd/util/cmd/checkpoint-convert-v6/cmd.go
@@ -0,0 +1,133 @@
+package checkpoint_convert_v6
+
+import (
+	"path/filepath"
+	"strings"
+
+	"github.com/rs/zerolog/log"
+	"github.com/spf13/cobra"
+
+	"github.com/onflow/flow-go/ledger/complete/wal"
+)
+
+var (
+	flagCheckpointDir  string
+	flagCheckpoint     string
+	flagExecutionDir   string
+	flagOutputDir      string
+	flagOutput         string
+	flagNWorker        uint
+	flagPrevCheckpoint int
+	flagWALFrom        int
+	flagWALTo          int
+)
+
+// Cmd reconstructs a full V6 checkpoint from a V7 (payloadless) checkpoint by
+// re-sourcing every leaf's payload from a previous full V6 checkpoint plus the
+// WAL segments written since.
+var Cmd = &cobra.Command{
+	Use:   "checkpoint-convert-v6",
+	Short: "Reconstruct a V6 checkpoint from a V7 (payloadless) checkpoint.",
+	Long: `Reconstruct a full V6 checkpoint from a V7 (payloadless) checkpoint.
+
+A V7 checkpoint stores only a leaf hash per register, not the payload, so it
+cannot be turned back into a V6 checkpoint on its own. This command recovers each
+payload from two sources in the execution directory:
+
+  - the previous full V6 checkpoint (registers not updated since), and
+  - the WAL segments written between that checkpoint and the V7 checkpoint
+    (registers that were updated).
+
+For each V7 leaf it finds the payload whose HashLeaf(path, value) matches the
+stored leaf hash. By default the previous checkpoint and the WAL range are
+auto-discovered: the previous full checkpoint is the latest V6 checkpoint with a
+number lower than the V7 checkpoint's number N, and the WAL range is (M, N].
+Both can be overridden with flags.
+
+The checkpoint is processed one subtrie partition at a time so only one
+partition's payloads are held in memory; --nworker partitions are processed
+concurrently (valid range [1, 16]), trading peak memory for speed.
+
+NOTE: the per-trie regSize (AllocatedRegSize) field is metrics-only and is not
+stored in a V7 checkpoint; it is written as 0 in the reconstructed checkpoint.
+This does not affect trie root hashes.`,
+	Run: run,
+}
+
+func init() {
+	Cmd.Flags().StringVar(&flagCheckpointDir, "checkpoint-dir", "",
+		"directory containing the V7 checkpoint files (required)")
+	_ = Cmd.MarkFlagRequired("checkpoint-dir")
+
+	Cmd.Flags().StringVar(&flagCheckpoint, "checkpoint", "",
+		"V7 checkpoint header filename, e.g. \"checkpoint.00000100.v7\" (required)")
+	_ = Cmd.MarkFlagRequired("checkpoint")
+
+	Cmd.Flags().StringVar(&flagExecutionDir, "execution-dir", "",
+		"ledger WAL directory holding the previous V6 checkpoint and WAL segments (required)")
+	_ = Cmd.MarkFlagRequired("execution-dir")
+
+	Cmd.Flags().StringVar(&flagOutputDir, "output-dir", "",
+		"directory to write the reconstructed V6 checkpoint files to (required)")
+	_ = Cmd.MarkFlagRequired("output-dir")
+
+	Cmd.Flags().StringVar(&flagOutput, "output", "",
+		"V6 output filename. Default: input filename with the \".v7\" suffix removed.")
+
+	Cmd.Flags().UintVar(&flagNWorker, "nworker", 1,
+		"number of subtrie partitions to process in parallel (valid range [1, 16])")
+
+	Cmd.Flags().IntVar(&flagPrevCheckpoint, "prev-checkpoint", -1,
+		"override the previous full V6 checkpoint number to source payloads from (default: auto-discover)")
+
+	Cmd.Flags().IntVar(&flagWALFrom, "wal-from", -1,
+		"override the first WAL segment number to replay (default: previous checkpoint + 1)")
+
+	Cmd.Flags().IntVar(&flagWALTo, "wal-to", -1,
+		"override the last WAL segment number to replay (default: V7 checkpoint number)")
+}
+
+func run(*cobra.Command, []string) {
+	outputFile := flagOutput
+	if outputFile == "" {
+		outputFile = defaultV6Filename(flagCheckpoint)
+	}
+
+	log.Info().
+		Str("checkpoint_dir", flagCheckpointDir).
+		Str("checkpoint", flagCheckpoint).
+		Str("execution_dir", flagExecutionDir).
+		Str("output_dir", flagOutputDir).
+		Str("output", outputFile).
+		Uint("nworker", flagNWorker).
+		Int("prev_checkpoint", flagPrevCheckpoint).
+		Int("wal_from", flagWALFrom).
+		Int("wal_to", flagWALTo).
+		Msg("reconstructing V6 checkpoint from V7")
+
+	err := wal.ConvertCheckpointV7ToV6(
+		flagCheckpointDir,
+		flagCheckpoint,
+		flagExecutionDir,
+		flagPrevCheckpoint,
+		flagWALFrom,
+		flagWALTo,
+		flagOutputDir,
+		outputFile,
+		log.Logger,
+		flagNWorker,
+	)
+	if err != nil {
+		log.Fatal().Err(err).Msg("checkpoint conversion failed")
+	}
+
+	log.Info().
+		Str("output", filepath.Join(flagOutputDir, outputFile)).
+		Msg("✅ V7→V6 checkpoint reconstruction completed successfully")
+}
+
+// defaultV6Filename returns the default V6 output filename for a given V7
+// checkpoint filename: strip the ".v7" suffix if present.
+func defaultV6Filename(v7Name string) string {
+	return strings.TrimSuffix(v7Name, wal.V7FileSuffix)
+}
diff --git a/cmd/util/cmd/root.go b/cmd/util/cmd/root.go
index b6156fbff9e..5514af7197e 100644
--- a/cmd/util/cmd/root.go
+++ b/cmd/util/cmd/root.go
@@ -15,6 +15,7 @@ import (
 	bootstrap_execution_state_payloads "github.com/onflow/flow-go/cmd/util/cmd/bootstrap-execution-state-payloads"
 	check_storage "github.com/onflow/flow-go/cmd/util/cmd/check-storage"
 	checkpoint_collect_stats "github.com/onflow/flow-go/cmd/util/cmd/checkpoint-collect-stats"
+	checkpoint_convert_v6 "github.com/onflow/flow-go/cmd/util/cmd/checkpoint-convert-v6"
 	checkpoint_convert_v7 "github.com/onflow/flow-go/cmd/util/cmd/checkpoint-convert-v7"
 	checkpoint_iterate_nodes "github.com/onflow/flow-go/cmd/util/cmd/checkpoint-iterate-nodes"
 	checkpoint_list_tries "github.com/onflow/flow-go/cmd/util/cmd/checkpoint-list-tries"
@@ -108,6 +109,7 @@ func addCommands() {
 	rootCmd.AddCommand(export.Cmd)
 	rootCmd.AddCommand(checkpoint_list_tries.Cmd)
 	rootCmd.AddCommand(checkpoint_collect_stats.Cmd)
+	rootCmd.AddCommand(checkpoint_convert_v6.Cmd)
 	rootCmd.AddCommand(checkpoint_convert_v7.Cmd)
 	rootCmd.AddCommand(checkpoint_iterate_nodes.Cmd)
 	rootCmd.AddCommand(checkpoint_verify_hash.Cmd)
diff --git a/ledger/complete/wal/checkpoint_v6_convert.go b/ledger/complete/wal/checkpoint_v6_convert.go
new file mode 100644
index 00000000000..6b96aadb5ec
--- /dev/null
+++ b/ledger/complete/wal/checkpoint_v6_convert.go
@@ -0,0 +1,345 @@
+package wal
+
+import (
+	"fmt"
+	"os"
+
+	prometheusWAL "github.com/onflow/wal/wal"
+	"github.com/rs/zerolog"
+
+	"github.com/onflow/flow-go/model/bootstrap"
+)
+
+// leaf-hash presence flag values in the V7 (payloadless) leaf encoding. They
+// mirror the (unexported) leafHashAbsent / leafHashPresent constants in
+// ledger/complete/payloadless/flattener.go; they are duplicated here because the
+// streaming converter operates on the raw byte stream rather than through the
+// payloadless flattener.
+const (
+	leafHashAbsentFlag  = byte(0)
+	leafHashPresentFlag = byte(1)
+)
+
+// ConvertCheckpointV7ToV6 reconstructs a full V6 checkpoint from a V7
+// (payloadless) checkpoint by re-sourcing every leaf's payload from a previous
+// full V6 checkpoint plus the WAL segments written between that previous
+// checkpoint and the V7 checkpoint.
+//
+// Rationale: a V7 checkpoint stores only a leaf hash per register, not the
+// payload, so it cannot be turned back into a V6 checkpoint on its own. However,
+// every payload referenced by the V7 checkpoint must exist either in the previous
+// full checkpoint (if the register was not updated since) or in one of the WAL
+// segments written since (if it was). By computing HashLeaf(path, value) for each
+// candidate payload from those two sources and matching it against the leaf hash
+// stored in the V7 checkpoint, the original payload is recovered.
+//
+// Inputs:
+//   - (v7Dir, v7File): the V7 checkpoint header file to convert, e.g.
+//     "checkpoint.00000100.v7". Its number N is parsed from the filename.
+//   - execDir: the standard ledger WAL directory holding both the previous V6
+//     checkpoint part files and the numbered WAL segment files.
+//   - prevCheckpointNum: the previous full V6 checkpoint number M to source
+//     unchanged payloads from. If negative, it is auto-discovered as the latest
+//     V6 checkpoint in execDir with number strictly less than N. If no such
+//     numbered checkpoint exists, it falls back to the V6 root checkpoint, in
+//     which case the full WAL range [0, N] is replayed.
+//   - walFrom, walTo: the inclusive WAL segment range to source updated payloads
+//     from. If negative, they default to (M+1, N] — i.e. all updates applied
+//     after the previous checkpoint up to and including the V7 checkpoint state.
+//   - (outputDir, outputFile): where to write the reconstructed V6 checkpoint.
+//
+// Memory: the checkpoint is processed one subtrie partition (first path nibble)
+// at a time. For partition i, only that partition's payloads are held in memory
+// while its V6 subtrie part file is written, then released before the next
+// partition. nWorker partitions are processed concurrently (valid range
+// [1, subtrieCount]), trading peak memory for speed.
+//
+// Limitation: the per-trie regSize (AllocatedRegSize) field is metrics-only and
+// is dropped by the V7 format; it is NOT reconstructed and is written as 0 in
+// every trie root record. A warning is logged. This does not affect trie root
+// hashes or any consensus-critical state; it only affects the LatestTrieRegSize
+// metric until the node rebuilds tries.
+//
+// The output filename must NOT carry the V7 suffix and no output part file may
+// already exist; otherwise the call is rejected. On any failure, partially
+// written output files are removed.
+//
+// No error returns are expected during normal operation; all error returns
+// indicate malformed input, missing source data, an unmatched leaf hash, a
+// clobbering output, or an IO failure.
+func ConvertCheckpointV7ToV6(
+	v7Dir string,
+	v7File string,
+	execDir string,
+	prevCheckpointNum int,
+	walFrom int,
+	walTo int,
+	outputDir string,
+	outputFile string,
+	logger zerolog.Logger,
+	nWorker uint,
+) error {
+	err := convertCheckpointV7ToV6(
+		v7Dir, v7File, execDir, prevCheckpointNum, walFrom, walTo, outputDir, outputFile, logger, nWorker)
+	if err != nil {
+		cleanupErr := deleteCheckpointFiles(outputDir, outputFile)
+		if cleanupErr != nil {
+			return fmt.Errorf("fail to cleanup temp file %s, after running into error: %w", cleanupErr, err)
+		}
+		return err
+	}
+	return nil
+}
+
+func convertCheckpointV7ToV6(
+	v7Dir string,
+	v7File string,
+	execDir string,
+	prevCheckpointNum int,
+	walFrom int,
+	walTo int,
+	outputDir string,
+	outputFile string,
+	logger zerolog.Logger,
+	nWorker uint,
+) error {
+	if nWorker == 0 || nWorker > subtrieCount {
+		return fmt.Errorf("invalid nWorker %v, valid range is [1, %v]", nWorker, subtrieCount)
+	}
+
+	// The output is a V6 checkpoint, so it must not use the V7 suffix.
+	if err := requireV6Filename(outputFile); err != nil {
+		return err
+	}
+
+	// Validate V7 input exists and read its part-file checksums.
+	v7Header := filePathCheckpointHeader(v7Dir, v7File)
+	if _, err := os.Stat(v7Header); err != nil {
+		return fmt.Errorf("V7 checkpoint header not found at %s: %w", v7Header, err)
+	}
+	v7SubtrieChecksums, v7TopTrieChecksum, err := readCheckpointHeaderV7(v7Header, logger)
+	if err != nil {
+		return fmt.Errorf("could not read V7 checkpoint header: %w", err)
+	}
+	if err := allPartFileExist(v7Dir, v7File, len(v7SubtrieChecksums)); err != nil {
+		return fmt.Errorf("V7 part files incomplete for %s/%s: %w", v7Dir, v7File, err)
+	}
+
+	// Determine the V7 checkpoint number N from its filename.
+	v7Info, ok := parseCheckpointFilename(v7File)
+	if !ok || v7Info.Version != VersionV7 {
+		return fmt.Errorf("could not parse V7 checkpoint number from filename %q", v7File)
+	}
+	n := v7Info.Number
+
+	// Resolve the previous full V6 checkpoint (M) to source unchanged payloads.
+	// prevNum is -1 when the resolved source is the V6 root checkpoint, which
+	// seeds WAL segment 0 (see resolveWALRange).
+	prevNum, prevFile, err := resolvePrevCheckpoint(execDir, n, prevCheckpointNum)
+	if err != nil {
+		return err
+	}
+	prevHeaderPath := filePathCheckpointHeader(execDir, prevFile)
+	if _, err := os.Stat(prevHeaderPath); err != nil {
+		return fmt.Errorf("previous V6 checkpoint header not found at %s: %w", prevHeaderPath, err)
+	}
+	prevSubtrieChecksums, prevTopTrieChecksum, err := readCheckpointHeader(prevHeaderPath, logger)
+	if err != nil {
+		return fmt.Errorf("could not read previous V6 checkpoint header: %w", err)
+	}
+	if err := allPartFileExist(execDir, prevFile, len(prevSubtrieChecksums)); err != nil {
+		return fmt.Errorf("previous V6 part files incomplete for %s/%s: %w", execDir, prevFile, err)
+	}
+
+	// Resolve and validate the WAL segment range (M, N] to source updated payloads.
+	from, to, err := resolveWALRange(execDir, prevNum, n, walFrom, walTo)
+	if err != nil {
+		return err
+	}
+
+	// Validate V6 output is not present (any of the part files).
+	v6Existing, err := findCheckpointPartFiles(outputDir, outputFile)
+	if err != nil {
+		return fmt.Errorf("could not check existing V6 output files: %w", err)
+	}
+	if len(v6Existing) != 0 {
+		return fmt.Errorf("V6 output already exists: %v", v6Existing)
+	}
+
+	// Remove any leftover temp part files from a previously interrupted conversion.
+	if err := removeStaleTempFiles(outputDir, outputFile, logger); err != nil {
+		return fmt.Errorf("could not remove stale temp files: %w", err)
+	}
+
+	logger.Info().
+		Str("v7_dir", v7Dir).
+		Str("v7_file", v7File).
+		Str("exec_dir", execDir).
+		Int("v7_number", n).
+		Int("prev_checkpoint", prevNum).
+		Int("wal_from", from).
+		Int("wal_to", to).
+		Str("output_dir", outputDir).
+		Str("output_file", outputFile).
+		Uint("nworker", nWorker).
+		Msg("starting streaming V7→V6 checkpoint conversion")
+
+	src := payloadSource{
+		execDir:              execDir,
+		prevFile:             prevFile,
+		prevSubtrieChecksums: prevSubtrieChecksums,
+		prevTopTrieChecksum:  prevTopTrieChecksum,
+		walFrom:              from,
+		walTo:                to,
+	}
+
+	// The top-trie part file may contain leaf nodes for registers that sit above
+	// the subtrie split (rare; only when a register is alone in a top-level
+	// subtree). Their payloads can belong to any partition, so they are sourced
+	// separately. This pre-scan is cheap and is usually empty for dense state.
+	topPool, err := buildTopTriePayloadPool(v7Dir, v7File, v7TopTrieChecksum, src, logger)
+	if err != nil {
+		return fmt.Errorf("could not build top-trie payload pool: %w", err)
+	}
+
+	// Convert the 16 subtrie part files concurrently, recomputing each checksum.
+	newSubtrieChecksums, err := convertSubTriesV7ToV6Concurrently(
+		v7Dir, v7File, outputDir, outputFile, v7SubtrieChecksums, src, logger, nWorker)
+	if err != nil {
+		return fmt.Errorf("could not convert subtrie files: %w", err)
+	}
+
+	// Convert the top-trie part file.
+	newTopTrieChecksum, err := convertTopTrieFileV7ToV6(
+		v7Dir, v7File, outputDir, outputFile, v7TopTrieChecksum, topPool, logger)
+	if err != nil {
+		return fmt.Errorf("could not convert top-trie file: %w", err)
+	}
+
+	// Write the V6 header referencing the freshly computed checksums.
+	if err := storeCheckpointHeader(newSubtrieChecksums, newTopTrieChecksum, outputDir, outputFile, logger); err != nil {
+		return fmt.Errorf("could not write V6 checkpoint header: %w", err)
+	}
+
+	// Sanity check: the reconstructed V6 root hashes must equal the V7 root hashes
+	// (they are carried over verbatim, so this validates the written file is
+	// well-formed and consistent).
+	if err := verifyRootHashesMatch(v7Dir, v7File, outputDir, outputFile, logger); err != nil {
+		return fmt.Errorf("root hash verification failed: %w", err)
+	}
+
+	logger.Info().Msg("stream V7→V6 checkpoint conversion complete")
+	return nil
+}
+
+// payloadSource describes where reconstructed payloads are sourced from: the
+// previous full V6 checkpoint and the WAL segment range written since.
+type payloadSource struct {
+	execDir              string
+	prevFile             string
+	prevSubtrieChecksums []uint32
+	prevTopTrieChecksum  uint32
+	walFrom              int
+	walTo                int
+}
+
+// resolvePrevCheckpoint returns the previous full V6 checkpoint to source
+// unchanged payloads from, as both its number and its on-disk filename.
+//
+// If override is non-negative it is used directly (and must be < n). Otherwise
+// the latest numbered V6 checkpoint with number < n in execDir is used. If no
+// such numbered checkpoint exists, it falls back to the V6 root checkpoint
+// (bootstrap.FilenameWALRootCheckpoint), which is the bootstrap full checkpoint
+// seeding WAL segment 0: sourcing from it requires replaying the full WAL range
+// [0, n] (resolveWALRange derives walFrom = prevNum+1 = 0 from the returned
+// prevNum of -1).
+//
+// The returned prevNum is the resolved checkpoint number, or -1 when the source
+// is the root checkpoint.
+//
+// No error returns are expected during normal operation.
+func resolvePrevCheckpoint(execDir string, n int, override int) (prevNum int, prevFile string, err error) {
+	if override >= 0 {
+		if override >= n {
+			return 0, "", fmt.Errorf("previous checkpoint %d must be less than V7 checkpoint %d", override, n)
+		}
+		return override, NumberToFilename(override), nil
+	}
+
+	nums, _, err := ListV6Checkpoints(execDir)
+	if err != nil {
+		return 0, "", fmt.Errorf("could not list V6 checkpoints in %s: %w", execDir, err)
+	}
+	prev := -1
+	for _, num := range nums {
+		if num < n && num > prev {
+			prev = num
+		}
+	}
+	if prev >= 0 {
+		return prev, NumberToFilename(prev), nil
+	}
+
+	// No numbered V6 checkpoint below n. Fall back to the V6 root checkpoint if
+	// present: it is a full checkpoint that, together with the WAL replayed from
+	// segment 0, can source every payload.
+	hasRoot, err := HasRootCheckpoint(execDir)
+	if err != nil {
+		return 0, "", fmt.Errorf("could not check for V6 root checkpoint in %s: %w", execDir, err)
+	}
+	if hasRoot {
+		return -1, bootstrap.FilenameWALRootCheckpoint, nil
+	}
+
+	return 0, "", fmt.Errorf("no previous V6 checkpoint with number < %d and no V6 root checkpoint found in %s; "+
+		"a full checkpoint is required to source unchanged payloads", n, execDir)
+}
+
+// resolveWALRange returns the inclusive WAL segment range to replay. When
+// overrideFrom / overrideTo are negative they default to (prevNum, n] = [prevNum+1, n].
+// The resolved range is validated against the segments available in execDir.
+// A returned range with from > to indicates no WAL replay is needed.
+//
+// No error returns are expected during normal operation.
+func resolveWALRange(execDir string, prevNum int, n int, overrideFrom int, overrideTo int) (from int, to int, err error) {
+	from = prevNum + 1
+	if overrideFrom >= 0 {
+		from = overrideFrom
+	}
+	to = n
+	if overrideTo >= 0 {
+		to = overrideTo
+	}
+
+	if from > to {
+		// No updates between the previous checkpoint and the V7 checkpoint.
+		return from, to, nil
+	}
+
+	first, last, err := prometheusWAL.Segments(execDir)
+	if err != nil {
+		return 0, 0, fmt.Errorf("could not list WAL segments in %s: %w", execDir, err)
+	}
+	if first < 0 {
+		return 0, 0, fmt.Errorf("no WAL segments found in %s but range [%d, %d] is required", execDir, from, to)
+	}
+	if from < first || to > last {
+		return 0, 0, fmt.Errorf("required WAL segment range [%d, %d] is not fully available; "+
+			"segments present are [%d, %d]", from, to, first, last)
+	}
+	return from, to, nil
+}
+
+// requireV6Filename rejects an output filename that is empty or carries the V7
+// suffix, since the reconstructed output is a V6 checkpoint.
+//
+// Expected error returns during normal operation: none.
+func requireV6Filename(fileName string) error {
+	if fileName == "" {
+		return fmt.Errorf("V6 output filename is empty")
+	}
+	if len(fileName) > len(V7FileSuffix) && fileName[len(fileName)-len(V7FileSuffix):] == V7FileSuffix {
+		return fmt.Errorf("V6 output filename %q must not end with %q", fileName, V7FileSuffix)
+	}
+	return nil
+}
diff --git a/ledger/complete/wal/checkpoint_v6_convert_stream.go b/ledger/complete/wal/checkpoint_v6_convert_stream.go
new file mode 100644
index 00000000000..82519181dca
--- /dev/null
+++ b/ledger/complete/wal/checkpoint_v6_convert_stream.go
@@ -0,0 +1,888 @@
+package wal
+
+import (
+	"bufio"
+	"encoding/binary"
+	"fmt"
+	"io"
+	"os"
+
+	prometheusWAL "github.com/onflow/wal/wal"
+	"github.com/rs/zerolog"
+
+	"github.com/onflow/flow-go/ledger"
+	"github.com/onflow/flow-go/ledger/common/hash"
+	"github.com/onflow/flow-go/ledger/complete/mtrie/flattener"
+	"github.com/onflow/flow-go/ledger/complete/mtrie/node"
+	"github.com/onflow/flow-go/ledger/complete/payloadless"
+)
+
+type v7ToV6SubtrieResult struct {
+	index    int
+	checksum uint32
+	err      error
+}
+
+// convertSubTriesV7ToV6Concurrently streams all subtrieCount subtrie part files
+// through the V7→V6 conversion using up to nWorker goroutines, and returns the
+// recomputed per-file checksums in subtrie-index order.
+//
+// Each worker builds the payload pool for its partition, converts that subtrie
+// part file, then releases the pool before taking the next job, so peak memory
+// is bounded by nWorker partitions' payloads.
+//
+// No error returns are expected during normal operation.
+func convertSubTriesV7ToV6Concurrently(
+	v7Dir string,
+	v7File string,
+	outputDir string,
+	outputFile string,
+	v7SubtrieChecksums []uint32,
+	src payloadSource,
+	logger zerolog.Logger,
+	nWorker uint,
+) ([]uint32, error) {
+	jobs := make(chan int, subtrieCount)
+	for i := 0; i < subtrieCount; i++ {
+		jobs <- i
+	}
+	close(jobs)
+
+	// Buffered to subtrieCount so workers never block on send, even if the
+	// collector returns early after the first error.
+	results := make(chan v7ToV6SubtrieResult, subtrieCount)
+
+	for w := 0; w < int(nWorker); w++ {
+		go func() {
+			for i := range jobs {
+				sum, err := convertSubTrieFileV7ToV6(
+					v7Dir, v7File, outputDir, outputFile, i, v7SubtrieChecksums[i], src, logger)
+				results <- v7ToV6SubtrieResult{index: i, checksum: sum, err: err}
+			}
+		}()
+	}
+
+	checksums := make([]uint32, subtrieCount)
+	for k := 0; k < subtrieCount; k++ {
+		r := <-results
+		if r.err != nil {
+			return nil, fmt.Errorf("fail to convert %v-th subtrie: %w", r.index, r.err)
+		}
+		checksums[r.index] = r.checksum
+	}
+	return checksums, nil
+}
+
+// convertSubTrieFileV7ToV6 builds the payload pool for partition `index`, then
+// streams the V7 subtrie part file at that index, writing the reconstructed V6
+// subtrie part file, and returns the recomputed checksum.
+//
+// expectedSum is the checksum recorded in the V7 header for this subtrie; it is
+// verified against the checksum embedded in the V7 subtrie file before conversion.
+//
+// No error returns are expected during normal operation.
+func convertSubTrieFileV7ToV6(
+	v7Dir string,
+	v7File string,
+	outputDir string,
+	outputFile string,
+	index int,
+	expectedSum uint32,
+	src payloadSource,
+	logger zerolog.Logger,
+) (checksum uint32, errToReturn error) {
+	// Build the leaf-hash → payload pool for this partition. It is released when
+	// this function returns, before the worker takes the next partition, so peak
+	// memory is bounded by nWorker partitions' payloads.
+	pool, err := buildPartitionPayloadPool(index, src, logger)
+	if err != nil {
+		return 0, fmt.Errorf("could not build payload pool for partition %d: %w", index, err)
+	}
+
+	inPath, _, err := filePathSubTries(v7Dir, v7File, index)
+	if err != nil {
+		return 0, err
+	}
+
+	inFile, err := os.Open(inPath)
+	if err != nil {
+		return 0, fmt.Errorf("could not open V7 subtrie file %v: %w", inPath, err)
+	}
+	defer func() {
+		errToReturn = closeAndMergeError(inFile, errToReturn)
+	}()
+
+	nodeCount, embeddedSum, err := readSubTriesFooter(inFile)
+	if err != nil {
+		return 0, fmt.Errorf("could not read V7 subtrie footer: %w", err)
+	}
+	if embeddedSum != expectedSum {
+		return 0, fmt.Errorf("mismatch checksum in V7 subtrie file %v: header has %v, file has %v",
+			index, expectedSum, embeddedSum)
+	}
+
+	if _, err := inFile.Seek(0, io.SeekStart); err != nil {
+		return 0, fmt.Errorf("could not seek to start of V7 subtrie file: %w", err)
+	}
+	if err := validateFileHeader(MagicBytesCheckpointSubtrie, VersionV7, inFile); err != nil {
+		return 0, fmt.Errorf("invalid V7 subtrie file header: %w", err)
+	}
+	reader := bufio.NewReaderSize(inFile, defaultBufioReadSize)
+
+	closable, err := createWriterForSubtrie(outputDir, outputFile, logger, index)
+	if err != nil {
+		return 0, fmt.Errorf("could not create writer for subtrie: %w", err)
+	}
+	defer func() {
+		errToReturn = closeAndMergeError(closable, errToReturn)
+	}()
+
+	writer := NewCRC32Writer(closable)
+	if _, err := writer.Write(encodeVersion(MagicBytesCheckpointSubtrie, VersionV6)); err != nil {
+		return 0, fmt.Errorf("cannot write version into subtrie file: %w", err)
+	}
+
+	logging := logProgress(fmt.Sprintf("converting %v-th sub trie (V7→V6)", index), int(nodeCount), logger)
+	conv := newV7ToV6NodeConverter(pool)
+	for i := uint64(0); i < nodeCount; i++ {
+		if err := conv.convertNode(reader, writer); err != nil {
+			return 0, fmt.Errorf("cannot convert node %d of subtrie %d: %w", i, index, err)
+		}
+		logging(i)
+	}
+
+	sum, err := storeSubtrieFooter(nodeCount, writer)
+	if err != nil {
+		return 0, fmt.Errorf("could not store subtrie footer: %w", err)
+	}
+	return sum, nil
+}
+
+// convertTopTrieFileV7ToV6 streams the V7 top-trie part file, converting its
+// top-level nodes (leaves sourced from topPool) and re-encoding each trie root
+// record to add back V6's register-size field (written as 0, see below), and
+// returns the recomputed checksum.
+//
+// regSize is metrics-only and not recoverable from a V7 checkpoint, so it is
+// written as 0; a warning is logged once.
+//
+// expectedSum is the top-trie checksum recorded in the V7 header; it is verified
+// against the checksum embedded in the V7 top-trie file before conversion.
+//
+// No error returns are expected during normal operation.
+func convertTopTrieFileV7ToV6(
+	v7Dir string,
+	v7File string,
+	outputDir string,
+	outputFile string,
+	expectedSum uint32,
+	topPool map[hash.Hash]*ledger.Payload,
+	logger zerolog.Logger,
+) (checksum uint32, errToReturn error) {
+	inPath, _ := filePathTopTries(v7Dir, v7File)
+
+	inFile, err := os.Open(inPath)
+	if err != nil {
+		return 0, fmt.Errorf("could not open V7 top-trie file %v: %w", inPath, err)
+	}
+	defer func() {
+		errToReturn = closeAndMergeError(inFile, errToReturn)
+	}()
+
+	topLevelNodesCount, triesCount, embeddedSum, err := readTopTriesFooter(inFile)
+	if err != nil {
+		return 0, fmt.Errorf("could not read V7 top-trie footer: %w", err)
+	}
+	if embeddedSum != expectedSum {
+		return 0, fmt.Errorf("mismatch V7 top-trie checksum: header has %v, file has %v",
+			expectedSum, embeddedSum)
+	}
+
+	if _, err := inFile.Seek(0, io.SeekStart); err != nil {
+		return 0, fmt.Errorf("could not seek to start of V7 top-trie file: %w", err)
+	}
+	if err := validateFileHeader(MagicBytesCheckpointToptrie, VersionV7, inFile); err != nil {
+		return 0, fmt.Errorf("invalid V7 top-trie file header: %w", err)
+	}
+	reader := bufio.NewReaderSize(inFile, defaultBufioReadSize)
+
+	// Read the subtrie node count and carry it over verbatim (unchanged by conversion).
+	subtrieNodeCountBuf := make([]byte, encNodeCountSize)
+	if _, err := io.ReadFull(reader, subtrieNodeCountBuf); err != nil {
+		return 0, fmt.Errorf("could not read subtrie node count: %w", err)
+	}
+
+	closable, err := createWriterForTopTries(outputDir, outputFile, logger)
+	if err != nil {
+		return 0, fmt.Errorf("could not create writer for top tries: %w", err)
+	}
+	defer func() {
+		errToReturn = closeAndMergeError(closable, errToReturn)
+	}()
+
+	writer := NewCRC32Writer(closable)
+	if _, err := writer.Write(encodeVersion(MagicBytesCheckpointToptrie, VersionV6)); err != nil {
+		return 0, fmt.Errorf("cannot write version into top-trie file: %w", err)
+	}
+	if _, err := writer.Write(subtrieNodeCountBuf); err != nil {
+		return 0, fmt.Errorf("cannot write subtrie node count: %w", err)
+	}
+
+	// Convert the top-level nodes (above subtrieLevel).
+	conv := newV7ToV6NodeConverter(topPool)
+	for i := uint64(0); i < topLevelNodesCount; i++ {
+		if err := conv.convertNode(reader, writer); err != nil {
+			return 0, fmt.Errorf("cannot convert top-level node %d: %w", i, err)
+		}
+	}
+
+	if triesCount > 0 {
+		logger.Warn().Msgf("regSize (AllocatedRegSize) is not reconstructed from a V7 checkpoint; "+
+			"writing 0 for all %d trie root record(s). This is a metrics-only field and does not "+
+			"affect trie root hashes.", triesCount)
+	}
+
+	// Re-encode each trie root record from V7 (index + regCount + hash) to V6
+	// (index + regCount + regSize + hash), adding back the register-size field as 0.
+	readScratch := make([]byte, payloadless.EncodedTrieSize)
+	trieBuf := make([]byte, flattener.EncodedTrieSize)
+	for i := uint16(0); i < triesCount; i++ {
+		encTrie, err := payloadless.ReadEncodedTrie(reader, readScratch)
+		if err != nil {
+			return 0, fmt.Errorf("cannot read trie root record %d: %w", i, err)
+		}
+
+		pos := 0
+		binary.BigEndian.PutUint64(trieBuf[pos:], encTrie.RootIndex)
+		pos += encNodeIndexSize
+		binary.BigEndian.PutUint64(trieBuf[pos:], encTrie.RegCount)
+		pos += encNodeIndexSize
+		binary.BigEndian.PutUint64(trieBuf[pos:], 0) // regSize: not reconstructed
+		pos += encNodeIndexSize
+		copy(trieBuf[pos:], encTrie.RootHash[:])
+
+		if _, err := writer.Write(trieBuf); err != nil {
+			return 0, fmt.Errorf("cannot write converted trie root record %d: %w", i, err)
+		}
+	}
+
+	sum, err := storeTopLevelTrieFooter(topLevelNodesCount, triesCount, writer)
+	if err != nil {
+		return 0, fmt.Errorf("could not store top-trie footer: %w", err)
+	}
+	return sum, nil
+}
+
+// v7ToV6NodeConverter streams individual V7-encoded nodes into V6-encoded nodes,
+// reusing internal scratch buffers across calls to avoid per-node allocations.
+// Leaf nodes are reconstructed by looking up their payload in `pool` by the
+// stored leaf hash; the node hash is carried over verbatim.
+//
+// NOT CONCURRENCY SAFE! A single converter must be used by one goroutine at a time.
+type v7ToV6NodeConverter struct {
+	pool       map[hash.Hash]*ledger.Payload
+	prefix     []byte // node type + height + hash (fixedNodePrefixSize)
+	childIndex []byte // interim left + right child indices
+	path       []byte // leaf path
+	leafHash   []byte // leaf hash bytes
+	enc        []byte // scratch for the V6 leaf encoding
+}
+
+// newV7ToV6NodeConverter returns a converter with preallocated scratch buffers.
+func newV7ToV6NodeConverter(pool map[hash.Hash]*ledger.Payload) *v7ToV6NodeConverter {
+	return &v7ToV6NodeConverter{
+		pool:       pool,
+		prefix:     make([]byte, fixedNodePrefixSize),
+		childIndex: make([]byte, 2*encNodeIndexSize),
+		path:       make([]byte, encPathSize),
+		leafHash:   make([]byte, encHashSize),
+		enc:        make([]byte, 1024*4),
+	}
+}
+
+// convertNode reads one V7-encoded node from reader and writes its V6 encoding to
+// writer. Interim nodes are copied verbatim (their on-disk format is identical in
+// V6); leaf nodes have their payload re-sourced from the pool and are re-encoded
+// with the V6 (full-payload) leaf format.
+//
+// Expected error returns during normal operation: none. An unmatched leaf hash
+// is treated as an exception (the source data is incomplete or inconsistent).
+func (c *v7ToV6NodeConverter) convertNode(reader io.Reader, writer io.Writer) error {
+	if _, err := io.ReadFull(reader, c.prefix); err != nil {
+		return fmt.Errorf("cannot read node prefix: %w", err)
+	}
+
+	switch c.prefix[0] {
+	case interimNodeTypeByte:
+		// Interim node: read the two child indices and copy the whole record verbatim.
+		if _, err := io.ReadFull(reader, c.childIndex); err != nil {
+			return fmt.Errorf("cannot read interim node child indices: %w", err)
+		}
+		if _, err := writer.Write(c.prefix); err != nil {
+			return fmt.Errorf("cannot write interim node prefix: %w", err)
+		}
+		if _, err := writer.Write(c.childIndex); err != nil {
+			return fmt.Errorf("cannot write interim node child indices: %w", err)
+		}
+		return nil
+
+	case leafNodeTypeByte:
+		return c.convertLeaf(reader, writer)
+
+	default:
+		return fmt.Errorf("failed to decode node type %d", c.prefix[0])
+	}
+}
+
+// convertLeaf reads the remainder of a V7 leaf node (path + leaf-hash flag +
+// optional leaf hash) from reader, having already consumed the shared prefix into
+// c.prefix, looks up the matching payload, and writes the reconstructed V6 leaf
+// node (with full payload) to writer.
+//
+// The node hash is carried over verbatim from the V7 stream; correctness of the
+// match is guaranteed because the pool is keyed by HashLeaf(path, value), which
+// is exactly the V7 leaf hash. Use checkpoint-verify-hash to independently
+// re-derive and verify node hashes from the reconstructed payloads.
+//
+// Expected error returns during normal operation: none. A missing payload for a
+// present leaf hash is treated as an exception.
+func (c *v7ToV6NodeConverter) convertLeaf(reader io.Reader, writer io.Writer) error {
+	height := binary.BigEndian.Uint16(c.prefix[encNodeTypeSize:])
+	nodeHash, err := hash.ToHash(c.prefix[encNodeTypeSize+encHeightSize:])
+	if err != nil {
+		return fmt.Errorf("failed to decode leaf node hash: %w", err)
+	}
+
+	// Read path (32 bytes).
+	if _, err := io.ReadFull(reader, c.path); err != nil {
+		return fmt.Errorf("cannot read leaf path: %w", err)
+	}
+	path, err := ledger.ToPath(c.path)
+	if err != nil {
+		return fmt.Errorf("failed to decode leaf path: %w", err)
+	}
+
+	// Read the leaf-hash presence flag (1 byte).
+	var flagBuf [encLeafHashFlagSize]byte
+	if _, err := io.ReadFull(reader, flagBuf[:]); err != nil {
+		return fmt.Errorf("cannot read leaf hash flag: %w", err)
+	}
+
+	var payload *ledger.Payload
+	switch flagBuf[0] {
+	case leafHashAbsentFlag:
+		// Unallocated leaf: reconstruct an empty-payload V6 leaf with the
+		// preserved (default-for-height) node hash.
+		//
+		// We must NOT use ledger.EmptyPayload() here: it leaves the encoded key
+		// nil, which makes the V6 leaf encoding self-inconsistent (the length
+		// prefix, derived from the decoded key, overcounts the bytes actually
+		// written) and the resulting part file cannot be read back. An explicit
+		// empty key round-trips correctly. Unallocated leaves are rare — pruned
+		// checkpoints (the production norm) contain none — and the register has
+		// no meaningful key, so an empty key is the faithful representation.
+		payload = ledger.NewPayload(ledger.NewKey(nil), ledger.Value{})
+
+	case leafHashPresentFlag:
+		if _, err := io.ReadFull(reader, c.leafHash); err != nil {
+			return fmt.Errorf("cannot read leaf hash: %w", err)
+		}
+		leafHash, err := hash.ToHash(c.leafHash)
+		if err != nil {
+			return fmt.Errorf("failed to decode leaf hash: %w", err)
+		}
+		payload = c.pool[leafHash]
+		if payload == nil {
+			return fmt.Errorf("no payload found for leaf hash %x at path %x; "+
+				"the previous checkpoint and WAL range do not contain this register's value",
+				leafHash, path)
+		}
+
+	default:
+		return fmt.Errorf("invalid leaf hash flag: %d", flagBuf[0])
+	}
+
+	// Carry the node hash over verbatim (see method doc for the correctness argument).
+	v6leaf := node.NewNode(int(height), nil, nil, path, payload, nodeHash)
+	encoded := flattener.EncodeNode(v6leaf, 0, 0, c.enc)
+	if _, err := writer.Write(encoded); err != nil {
+		return fmt.Errorf("cannot write reconstructed leaf node: %w", err)
+	}
+	return nil
+}
+
+// buildPartitionPayloadPool builds the leaf-hash → payload pool for the given
+// partition (first path nibble) by scanning the previous checkpoint's subtrie
+// part file for that partition and re-scanning the WAL segment range, keeping
+// only pairs whose path falls in the partition.
+//
+// Memory is bounded by this single partition's payloads.
+//
+// TODO(perf): the WAL range (and the previous checkpoint's top-trie) is
+// re-scanned once per partition (16 scans total) to keep peak memory minimal. A
+// future optimization is to make a single WAL pass that splits updates into 16
+// on-disk partition buckets, then read each bucket once. The on-disk persistence
+// format for those buckets is not yet decided.
+//
+// No error returns are expected during normal operation.
+func buildPartitionPayloadPool(
+	partition int,
+	src payloadSource,
+	logger zerolog.Logger,
+) (map[hash.Hash]*ledger.Payload, error) {
+	pool := make(map[hash.Hash]*ledger.Payload)
+
+	// The payloads passed to the source callbacks are only valid for the duration
+	// of the call (they alias a reused read buffer), so the pool deep-copies on
+	// store. Copying here — after the partition and empty filters — also avoids
+	// copying the ~15/16 of WAL payloads that this partition discards.
+	add := func(path ledger.Path, payload *ledger.Payload) {
+		if payload.IsEmpty() {
+			return
+		}
+		leafHash := hash.HashLeaf(hash.Hash(path), payload.Value())
+		pool[leafHash] = payload.DeepCopy()
+	}
+
+	partitionFilteredAdd := func(path ledger.Path, payload *ledger.Payload) {
+		if int(path[0]>>4) != partition {
+			return
+		}
+		add(path, payload)
+	}
+
+	// Source A: the previous checkpoint's subtrie part file for this partition.
+	// All of its leaves belong to this partition by construction.
+	err := streamV6SubtrieLeaves(src.execDir, src.prevFile, partition,
+		src.prevSubtrieChecksums[partition], func(path ledger.Path, payload *ledger.Payload) {
+			add(path, payload)
+		})
+	if err != nil {
+		return nil, fmt.Errorf("could not scan previous checkpoint subtrie %d: %w", partition, err)
+	}
+
+	// Source A': the previous checkpoint's top-trie part file. A register that was
+	// a compactified leaf high in the previous trie (above the subtrie split) lives
+	// in the top-trie file rather than in subtrie `partition`. A later trie may
+	// de-compactify that same register into this partition's subtrie (e.g. once a
+	// sibling register is added), so its payload must be sourced here too. The
+	// top-trie file is small, so scanning it per partition is cheap.
+	err = streamV6TopTrieLeaves(src.execDir, src.prevFile, src.prevTopTrieChecksum, partitionFilteredAdd)
+	if err != nil {
+		return nil, fmt.Errorf("could not scan previous checkpoint top-trie for partition %d: %w", partition, err)
+	}
+
+	// Source B: WAL updates in this partition.
+	if src.walFrom <= src.walTo {
+		err = scanWALUpdates(src.execDir, src.walFrom, src.walTo,
+			logger, fmt.Sprintf("partition %d", partition), partitionFilteredAdd)
+		if err != nil {
+			return nil, fmt.Errorf("could not scan WAL for partition %d: %w", partition, err)
+		}
+	}
+
+	logger.Debug().Int("partition", partition).Int("pool_size", len(pool)).
+		Msg("built partition payload pool")
+	return pool, nil
+}
+
+// buildTopTriePayloadPool collects payloads for any leaf nodes stored in the V7
+// top-trie part file (registers that sit above the subtrie split). Their paths
+// may belong to any partition, so they are sourced by first collecting the set of
+// leaf hashes referenced by the top-trie, then scanning all previous-checkpoint
+// part files and the WAL range for matching payloads.
+//
+// For dense state (e.g. mainnet) the top-trie has no leaf nodes and this returns
+// an empty pool without scanning any source.
+//
+// No error returns are expected during normal operation.
+func buildTopTriePayloadPool(
+	v7Dir string,
+	v7File string,
+	v7TopTrieChecksum uint32,
+	src payloadSource,
+	logger zerolog.Logger,
+) (map[hash.Hash]*ledger.Payload, error) {
+	needed, err := collectTopTrieLeafHashes(v7Dir, v7File, v7TopTrieChecksum)
+	if err != nil {
+		return nil, fmt.Errorf("could not collect top-trie leaf hashes: %w", err)
+	}
+
+	pool := make(map[hash.Hash]*ledger.Payload)
+	if len(needed) == 0 {
+		return pool, nil
+	}
+
+	logger.Info().Int("needed", len(needed)).
+		Msg("V7 top-trie contains leaf nodes; scanning all sources for their payloads")
+
+	// Deep-copy on store: source payloads alias a reused read buffer and are only
+	// valid during the callback.
+	add := func(path ledger.Path, payload *ledger.Payload) {
+		if payload.IsEmpty() {
+			return
+		}
+		leafHash := hash.HashLeaf(hash.Hash(path), payload.Value())
+		if _, ok := needed[leafHash]; !ok {
+			return
+		}
+		pool[leafHash] = payload.DeepCopy()
+	}
+
+	// Scan every previous-checkpoint subtrie part file.
+	for i := 0; i < subtrieCount; i++ {
+		err := streamV6SubtrieLeaves(src.execDir, src.prevFile, i, src.prevSubtrieChecksums[i], add)
+		if err != nil {
+			return nil, fmt.Errorf("could not scan previous checkpoint subtrie %d: %w", i, err)
+		}
+	}
+	// Scan the previous-checkpoint top-trie part file leaves.
+	if err := streamV6TopTrieLeaves(src.execDir, src.prevFile, src.prevTopTrieChecksum, add); err != nil {
+		return nil, fmt.Errorf("could not scan previous checkpoint top-trie: %w", err)
+	}
+	// Scan the WAL range.
+	if src.walFrom <= src.walTo {
+		if err := scanWALUpdates(src.execDir, src.walFrom, src.walTo,
+			logger, "top-trie", add); err != nil {
+			return nil, fmt.Errorf("could not scan WAL for top-trie leaves: %w", err)
+		}
+	}
+
+	if len(pool) < len(needed) {
+		return nil, fmt.Errorf("could not source all top-trie leaf payloads: found %d of %d",
+			len(pool), len(needed))
+	}
+	return pool, nil
+}
+
+// collectTopTrieLeafHashes streams the V7 top-trie part file's top-level nodes
+// and returns the set of leaf hashes for the leaf nodes among them. Leaf nodes
+// with an absent leaf hash (unallocated) need no payload and are skipped.
+//
+// No error returns are expected during normal operation.
+func collectTopTrieLeafHashes(
+	v7Dir string,
+	v7File string,
+	expectedSum uint32,
+) (set map[hash.Hash]struct{}, errToReturn error) {
+	inPath, _ := filePathTopTries(v7Dir, v7File)
+	inFile, err := os.Open(inPath)
+	if err != nil {
+		return nil, fmt.Errorf("could not open V7 top-trie file %v: %w", inPath, err)
+	}
+	defer func() {
+		errToReturn = closeAndMergeError(inFile, errToReturn)
+	}()
+
+	topLevelNodesCount, _, embeddedSum, err := readTopTriesFooter(inFile)
+	if err != nil {
+		return nil, fmt.Errorf("could not read V7 top-trie footer: %w", err)
+	}
+	if embeddedSum != expectedSum {
+		return nil, fmt.Errorf("mismatch V7 top-trie checksum: header has %v, file has %v", expectedSum, embeddedSum)
+	}
+
+	if _, err := inFile.Seek(0, io.SeekStart); err != nil {
+		return nil, fmt.Errorf("could not seek to start of V7 top-trie file: %w", err)
+	}
+	if err := validateFileHeader(MagicBytesCheckpointToptrie, VersionV7, inFile); err != nil {
+		return nil, fmt.Errorf("invalid V7 top-trie file header: %w", err)
+	}
+	reader := bufio.NewReaderSize(inFile, defaultBufioReadSize)
+
+	// Skip the subtrie node count.
+	if _, err := io.CopyN(io.Discard, reader, int64(encNodeCountSize)); err != nil {
+		return nil, fmt.Errorf("could not skip subtrie node count: %w", err)
+	}
+
+	set = make(map[hash.Hash]struct{})
+	prefix := make([]byte, fixedNodePrefixSize)
+	childIndex := make([]byte, 2*encNodeIndexSize)
+	pathBuf := make([]byte, encPathSize)
+	leafHashBuf := make([]byte, encHashSize)
+	var flagBuf [encLeafHashFlagSize]byte
+
+	for i := uint64(0); i < topLevelNodesCount; i++ {
+		if _, err := io.ReadFull(reader, prefix); err != nil {
+			return nil, fmt.Errorf("cannot read node prefix: %w", err)
+		}
+		switch prefix[0] {
+		case interimNodeTypeByte:
+			if _, err := io.ReadFull(reader, childIndex); err != nil {
+				return nil, fmt.Errorf("cannot read interim child indices: %w", err)
+			}
+		case leafNodeTypeByte:
+			if _, err := io.ReadFull(reader, pathBuf); err != nil {
+				return nil, fmt.Errorf("cannot read leaf path: %w", err)
+			}
+			if _, err := io.ReadFull(reader, flagBuf[:]); err != nil {
+				return nil, fmt.Errorf("cannot read leaf hash flag: %w", err)
+			}
+			switch flagBuf[0] {
+			case leafHashAbsentFlag:
+				// unallocated, no payload needed
+			case leafHashPresentFlag:
+				if _, err := io.ReadFull(reader, leafHashBuf); err != nil {
+					return nil, fmt.Errorf("cannot read leaf hash: %w", err)
+				}
+				leafHash, err := hash.ToHash(leafHashBuf)
+				if err != nil {
+					return nil, fmt.Errorf("failed to decode leaf hash: %w", err)
+				}
+				set[leafHash] = struct{}{}
+			default:
+				return nil, fmt.Errorf("invalid leaf hash flag: %d", flagBuf[0])
+			}
+		default:
+			return nil, fmt.Errorf("failed to decode node type %d", prefix[0])
+		}
+	}
+
+	return set, nil
+}
+
+// streamV6SubtrieLeaves opens the V6 subtrie part file at the given index and
+// invokes cb for every leaf node, passing its path and decoded payload. Interim
+// nodes are skipped. The embedded checksum is verified against expectedSum.
+//
+// No error returns are expected during normal operation.
+func streamV6SubtrieLeaves(
+	dir string,
+	file string,
+	index int,
+	expectedSum uint32,
+	cb func(path ledger.Path, payload *ledger.Payload),
+) (errToReturn error) {
+	inPath, _, err := filePathSubTries(dir, file, index)
+	if err != nil {
+		return err
+	}
+	inFile, err := os.Open(inPath)
+	if err != nil {
+		return fmt.Errorf("could not open V6 subtrie file %v: %w", inPath, err)
+	}
+	defer func() {
+		errToReturn = closeAndMergeError(inFile, errToReturn)
+	}()
+
+	nodeCount, embeddedSum, err := readSubTriesFooter(inFile)
+	if err != nil {
+		return fmt.Errorf("could not read V6 subtrie footer: %w", err)
+	}
+	if embeddedSum != expectedSum {
+		return fmt.Errorf("mismatch checksum in V6 subtrie file %v: header has %v, file has %v",
+			index, expectedSum, embeddedSum)
+	}
+
+	if _, err := inFile.Seek(0, io.SeekStart); err != nil {
+		return fmt.Errorf("could not seek to start of V6 subtrie file: %w", err)
+	}
+	if err := validateFileHeader(MagicBytesCheckpointSubtrie, VersionV6, inFile); err != nil {
+		return fmt.Errorf("invalid V6 subtrie file header: %w", err)
+	}
+	reader := bufio.NewReaderSize(inFile, defaultBufioReadSize)
+
+	return streamV6LeafNodes(reader, nodeCount, cb)
+}
+
+// streamV6TopTrieLeaves opens the V6 top-trie part file and invokes cb for every
+// leaf node among its top-level nodes. Interim nodes and trie root records are
+// skipped. The embedded checksum is verified against expectedSum.
+//
+// No error returns are expected during normal operation.
+func streamV6TopTrieLeaves(
+	dir string,
+	file string,
+	expectedSum uint32,
+	cb func(path ledger.Path, payload *ledger.Payload),
+) (errToReturn error) {
+	inPath, _ := filePathTopTries(dir, file)
+	inFile, err := os.Open(inPath)
+	if err != nil {
+		return fmt.Errorf("could not open V6 top-trie file %v: %w", inPath, err)
+	}
+	defer func() {
+		errToReturn = closeAndMergeError(inFile, errToReturn)
+	}()
+
+	topLevelNodesCount, _, embeddedSum, err := readTopTriesFooter(inFile)
+	if err != nil {
+		return fmt.Errorf("could not read V6 top-trie footer: %w", err)
+	}
+	if embeddedSum != expectedSum {
+		return fmt.Errorf("mismatch V6 top-trie checksum: header has %v, file has %v", expectedSum, embeddedSum)
+	}
+
+	if _, err := inFile.Seek(0, io.SeekStart); err != nil {
+		return fmt.Errorf("could not seek to start of V6 top-trie file: %w", err)
+	}
+	if err := validateFileHeader(MagicBytesCheckpointToptrie, VersionV6, inFile); err != nil {
+		return fmt.Errorf("invalid V6 top-trie file header: %w", err)
+	}
+	reader := bufio.NewReaderSize(inFile, defaultBufioReadSize)
+
+	// Skip the subtrie node count.
+	if _, err := io.CopyN(io.Discard, reader, int64(encNodeCountSize)); err != nil {
+		return fmt.Errorf("could not skip subtrie node count: %w", err)
+	}
+
+	return streamV6LeafNodes(reader, topLevelNodesCount, cb)
+}
+
+// streamV6LeafNodes reads nodeCount V6-encoded nodes from reader and invokes cb
+// for each leaf node with its path and decoded payload. Interim nodes are skipped.
+//
+// The payload passed to cb is decoded zero-copy over a reused scratch buffer and
+// is only valid for the duration of the call; a cb that retains it MUST deep-copy.
+//
+// No error returns are expected during normal operation.
+func streamV6LeafNodes(
+	reader io.Reader,
+	nodeCount uint64,
+	cb func(path ledger.Path, payload *ledger.Payload),
+) error {
+	prefix := make([]byte, fixedNodePrefixSize)
+	childIndex := make([]byte, 2*encNodeIndexSize)
+	pathBuf := make([]byte, encPathSize)
+	lenBuf := make([]byte, encPayloadLengthSize)
+	payloadBuf := make([]byte, 1024)
+
+	for i := uint64(0); i < nodeCount; i++ {
+		if _, err := io.ReadFull(reader, prefix); err != nil {
+			return fmt.Errorf("cannot read node prefix: %w", err)
+		}
+		switch prefix[0] {
+		case interimNodeTypeByte:
+			if _, err := io.ReadFull(reader, childIndex); err != nil {
+				return fmt.Errorf("cannot read interim child indices: %w", err)
+			}
+		case leafNodeTypeByte:
+			if _, err := io.ReadFull(reader, pathBuf); err != nil {
+				return fmt.Errorf("cannot read leaf path: %w", err)
+			}
+			path, err := ledger.ToPath(pathBuf)
+			if err != nil {
+				return fmt.Errorf("failed to decode leaf path: %w", err)
+			}
+			if _, err := io.ReadFull(reader, lenBuf); err != nil {
+				return fmt.Errorf("cannot read leaf payload length: %w", err)
+			}
+			size := binary.BigEndian.Uint32(lenBuf)
+			if uint32(cap(payloadBuf)) < size {
+				payloadBuf = make([]byte, size)
+			}
+			buf := payloadBuf[:size]
+			if _, err := io.ReadFull(reader, buf); err != nil {
+				return fmt.Errorf("cannot read leaf payload: %w", err)
+			}
+			// zeroCopy: the payload aliases payloadBuf and is only valid for this
+			// cb call; cb deep-copies if it retains the payload (see doc).
+			payload, err := ledger.DecodePayloadWithoutPrefix(buf, true, payloadEncodingVersion)
+			if err != nil {
+				return fmt.Errorf("failed to decode leaf payload: %w", err)
+			}
+			cb(path, payload)
+		default:
+			return fmt.Errorf("failed to decode node type %d", prefix[0])
+		}
+	}
+	return nil
+}
+
+// scanWALUpdates reads the WAL segment records in the inclusive range [from, to]
+// and invokes cb for every (path, payload) pair in each update record. Delete
+// records are ignored. It logs one line per WAL segment as the scan advances
+// into it.
+//
+// The payload passed to cb is only valid for the duration of the call:
+// [ledger.DecodeTrieUpdate] decodes payloads zero-copy over the WAL record
+// buffer, which the underlying reader reuses on the next record. A cb that
+// retains the payload MUST deep-copy it.
+//
+// No error returns are expected during normal operation.
+func scanWALUpdates(
+	execDir string,
+	from int,
+	to int,
+	logger zerolog.Logger,
+	label string,
+	cb func(path ledger.Path, payload *ledger.Payload),
+) error {
+	sr, err := prometheusWAL.NewSegmentsRangeReader(zerolog.Nop(), prometheusWAL.SegmentRange{
+		Dir:   execDir,
+		First: from,
+		Last:  to,
+	})
+	if err != nil {
+		return fmt.Errorf("cannot create WAL segment reader for [%d, %d]: %w", from, to, err)
+	}
+	defer sr.Close()
+
+	reader := prometheusWAL.NewReader(sr)
+	// Log each WAL segment as the reader advances into it. The combined reader is
+	// kept (rather than reading segments individually) so records spanning a
+	// segment boundary still decode; reader.Segment() reports the segment of the
+	// record just read, so logging on change emits one line per segment.
+	// Note: this scan runs once per partition, so each segment is logged ~once per
+	// partition over the full conversion.
+	totalSegments := to - from + 1
+	currentSegment := -1
+	for reader.Next() {
+		if seg := reader.Segment(); seg != currentSegment {
+			currentSegment = seg
+			logger.Info().
+				Str("scan", label).
+				Int("segment", seg).
+				Int("wal_from", from).
+				Int("wal_to", to).
+				Msgf("[%s] processing WAL segment %d/%d", label, seg-from+1, totalSegments)
+		}
+
+		record := reader.Record()
+		operation, _, update, err := Decode(record)
+		if err != nil {
+			return fmt.Errorf("cannot decode WAL record: %w", err)
+		}
+		if operation != WALUpdate {
+			continue
+		}
+		for i, path := range update.Paths {
+			cb(path, update.Payloads[i])
+		}
+	}
+	if err := reader.Err(); err != nil {
+		return fmt.Errorf("cannot read WAL: %w", err)
+	}
+	return nil
+}
+
+// verifyRootHashesMatch reads the trie root hashes from the reconstructed V6
+// checkpoint and the source V7 checkpoint and verifies they are identical.
+//
+// No error returns are expected during normal operation.
+func verifyRootHashesMatch(
+	v7Dir string,
+	v7File string,
+	v6Dir string,
+	v6File string,
+	logger zerolog.Logger,
+) error {
+	v7Hashes, err := ReadTriesRootHashV7(logger, v7Dir, v7File)
+	if err != nil {
+		return fmt.Errorf("could not read V7 root hashes: %w", err)
+	}
+	v6Hashes, err := ReadTriesRootHash(logger, v6Dir, v6File)
+	if err != nil {
+		return fmt.Errorf("could not read reconstructed V6 root hashes: %w", err)
+	}
+	if len(v7Hashes) != len(v6Hashes) {
+		return fmt.Errorf("trie count mismatch: V7 has %d, V6 has %d", len(v7Hashes), len(v6Hashes))
+	}
+	for i := range v7Hashes {
+		if v7Hashes[i] != v6Hashes[i] {
+			return fmt.Errorf("trie %d root hash mismatch: V7=%s V6=%s", i, v7Hashes[i], v6Hashes[i])
+		}
+	}
+	logger.Info().Int("trie_count", len(v6Hashes)).Msg("verified reconstructed V6 root hashes match V7")
+	return nil
+}
diff --git a/ledger/complete/wal/checkpoint_v6_convert_test.go b/ledger/complete/wal/checkpoint_v6_convert_test.go
new file mode 100644
index 00000000000..ead02a832b2
--- /dev/null
+++ b/ledger/complete/wal/checkpoint_v6_convert_test.go
@@ -0,0 +1,322 @@
+package wal
+
+import (
+	"fmt"
+	"os"
+	"path"
+	"testing"
+
+	prometheusWAL "github.com/onflow/wal/wal"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
+
+	"github.com/onflow/flow-go/ledger"
+	"github.com/onflow/flow-go/ledger/complete/mtrie/trie"
+	"github.com/onflow/flow-go/model/bootstrap"
+	"github.com/onflow/flow-go/module/metrics"
+	"github.com/onflow/flow-go/utils/unittest"
+)
+
+// TestConvertCheckpointV7ToV6_RoundTrip builds a previous full V6 checkpoint plus
+// WAL segments carrying later updates, converts a V7 checkpoint of the resulting
+// forest back into a V6 checkpoint, and verifies the reconstruction is correct.
+//
+// Correctness is checked with VerifyCheckpointHashes, which re-derives each V6
+// leaf hash from its reconstructed payload and compares it to the stored node
+// hash. Because the converter carries node hashes over verbatim, this is the
+// check that actually proves every payload was sourced correctly (a wrong payload
+// with a carried-over hash would fail re-derivation). Trie root hashes and a
+// register spot-check provide additional confidence.
+func TestConvertCheckpointV7ToV6_RoundTrip(t *testing.T) {
+	for _, nWorker := range []uint{1, 4, 16} {
+		t.Run(fmt.Sprintf("nWorker=%d", nWorker), func(t *testing.T) {
+			unittest.RunWithTempDir(t, func(dir string) {
+				logger := zerolog.Nop()
+
+				allTries, lastTrie, lastPaths, lastValues := setupV7ToV6Scenario(t, dir, logger, "checkpoint.00000000")
+
+				outDir := path.Join(dir, "out")
+				require.NoError(t, os.MkdirAll(outDir, 0755))
+
+				v7Name := "checkpoint.00000005.v7"
+				outName := "checkpoint.00000005"
+
+				first, last, err := prometheusWAL.Segments(dir)
+				require.NoError(t, err)
+				require.GreaterOrEqual(t, first, 0, "expected WAL segments to exist")
+
+				// Convert with explicit overrides: previous checkpoint M=0 and the
+				// full available WAL range. Over-scanning the WAL is safe — the pool
+				// is keyed by leaf hash, so extra entries never produce wrong matches.
+				require.NoError(t, ConvertCheckpointV7ToV6(
+					dir, v7Name, dir, 0, first, last, outDir, outName, logger, nWorker))
+
+				// The reconstructed V6 leaf hashes must re-derive from the payloads.
+				require.NoError(t, VerifyCheckpointHashes(logger, outDir, outName, nWorker))
+
+				// Root hashes and trie count must match the source forest.
+				reconstructed, err := OpenAndReadCheckpointV6(outDir, outName, logger)
+				require.NoError(t, err)
+				require.Equal(t, len(allTries), len(reconstructed))
+				for i, expected := range allTries {
+					require.Equal(t, expected.RootHash(), reconstructed[i].RootHash(),
+						"trie %d root hash mismatch", i)
+				}
+
+				// Spot-check that the actual register values were recovered for the
+				// latest trie (the one that mixes previous-checkpoint and WAL sources).
+				reconLast := reconstructed[len(reconstructed)-1]
+				require.Equal(t, lastTrie.RootHash(), reconLast.RootHash())
+				expectedByPath := make(map[ledger.Path]ledger.Value, len(lastPaths))
+				for i, p := range lastPaths {
+					expectedByPath[p] = lastValues[i]
+				}
+				// UnsafeRead permutes its input in place and aligns results to the
+				// permuted order, so compare against the (post-permutation) paths.
+				readPaths := append([]ledger.Path{}, lastPaths...)
+				got := reconLast.UnsafeRead(readPaths)
+				require.Len(t, got, len(readPaths))
+				for i := range readPaths {
+					require.Equal(t, expectedByPath[readPaths[i]], got[i].Value(),
+						"register value mismatch at path %x", readPaths[i])
+				}
+			})
+		})
+	}
+}
+
+// TestConvertCheckpointV7ToV6_RoundTripFromRoot is the round-trip test with the
+// previous full checkpoint stored as the V6 root checkpoint rather than a
+// numbered checkpoint. The previous checkpoint is auto-discovered (prevCheckpointNum
+// = -1), exercising the root-checkpoint fallback and the resulting WAL replay from
+// segment 0.
+func TestConvertCheckpointV7ToV6_RoundTripFromRoot(t *testing.T) {
+	unittest.RunWithTempDir(t, func(dir string) {
+		logger := zerolog.Nop()
+
+		allTries, lastTrie, _, _ := setupV7ToV6Scenario(t, dir, logger, bootstrap.FilenameWALRootCheckpoint)
+
+		outDir := path.Join(dir, "out")
+		require.NoError(t, os.MkdirAll(outDir, 0755))
+
+		first, last, err := prometheusWAL.Segments(dir)
+		require.NoError(t, err)
+		require.Equal(t, 0, first, "root-sourced replay must start at WAL segment 0")
+
+		// Auto-discover the previous checkpoint (no numbered checkpoint exists, so
+		// it falls back to root.checkpoint). The WAL range is given explicitly so
+		// the test does not depend on the V7 number matching the last segment.
+		require.NoError(t, ConvertCheckpointV7ToV6(
+			dir, "checkpoint.00000005.v7", dir, -1, first, last, outDir, "checkpoint.00000005", logger, 4))
+
+		require.NoError(t, VerifyCheckpointHashes(logger, outDir, "checkpoint.00000005", 4))
+
+		reconstructed, err := OpenAndReadCheckpointV6(outDir, "checkpoint.00000005", logger)
+		require.NoError(t, err)
+		require.Equal(t, len(allTries), len(reconstructed))
+		for i, expected := range allTries {
+			require.Equal(t, expected.RootHash(), reconstructed[i].RootHash(), "trie %d root hash mismatch", i)
+		}
+		require.Equal(t, lastTrie.RootHash(), reconstructed[len(reconstructed)-1].RootHash())
+	})
+}
+
+// setupV7ToV6Scenario writes, into dir:
+//   - a previous full V6 checkpoint named prevName (state after update u0),
+//   - WAL segments carrying two later updates u1 (with overwrites and new
+//     registers) and u2 (with overwrites and a deletion), and
+//   - a V7 checkpoint "checkpoint.00000005.v7" of the forest holding all three
+//     trie states.
+//
+// It returns all three trie states, the final trie, and the final trie's paths
+// and expected values for a register spot-check.
+func setupV7ToV6Scenario(t *testing.T, dir string, logger zerolog.Logger, prevName string) (
+	allTries []*trie.MTrie, lastTrie *trie.MTrie, lastPaths []ledger.Path, lastValues []ledger.Value,
+) {
+	// u0: initial state -> trie0. Stored only in the previous full checkpoint.
+	pathsA, payloadsA := randNPathPayloads(50)
+	trie0, _, err := trie.NewTrieWithUpdatedRegisters(trie.NewEmptyMTrie(), pathsA, payloadsA, true)
+	require.NoError(t, err)
+
+	require.NoError(t, StoreCheckpointV6Concurrently([]*trie.MTrie{trie0}, dir, prevName, logger))
+
+	// u1: overwrite the first 10 of A with new values, plus 20 brand-new registers.
+	// Overwrites keep each path's original key and only change the value, honoring
+	// the MTrie invariant that a path's key never changes (in Flow, path = hash(key)).
+	overwrites1 := overwriteValues(t, payloadsA[:10])
+	pathsB, payloadsB := randNPathPayloads(20)
+	u1Paths := append(append([]ledger.Path{}, pathsA[:10]...), pathsB...)
+	u1Payloads := append(append([]ledger.Payload{}, overwrites1...), payloadsB...)
+	trie1, _, err := trie.NewTrieWithUpdatedRegisters(trie0, u1Paths, u1Payloads, true)
+	require.NoError(t, err)
+
+	// u2: overwrite the first 5 of B, and delete one of A (empty payload).
+	overwrites2 := overwriteValues(t, payloadsB[:5])
+	u2Paths := append(append([]ledger.Path{}, pathsB[:5]...), pathsA[10])
+	u2Payloads := append(append([]ledger.Payload{}, overwrites2...), *ledger.EmptyPayload())
+	trie2, _, err := trie.NewTrieWithUpdatedRegisters(trie1, u2Paths, u2Payloads, true)
+	require.NoError(t, err)
+
+	// Record u1 and u2 into the WAL (u0 is intentionally NOT recorded — its values
+	// must come from the previous checkpoint).
+	recordWAL, err := NewDiskWAL(logger, nil, metrics.NewNoopCollector(), dir, 10, pathByteSize, segmentSize)
+	require.NoError(t, err)
+	_, _, err = recordWAL.RecordUpdate(&ledger.TrieUpdate{
+		RootHash: trie0.RootHash(), Paths: u1Paths, Payloads: toPayloadPtrs(u1Payloads)})
+	require.NoError(t, err)
+	_, _, err = recordWAL.RecordUpdate(&ledger.TrieUpdate{
+		RootHash: trie1.RootHash(), Paths: u2Paths, Payloads: toPayloadPtrs(u2Payloads)})
+	require.NoError(t, err)
+	<-recordWAL.Done()
+
+	// V7 checkpoint of the forest holding all three trie states.
+	allTries = []*trie.MTrie{trie0, trie1, trie2}
+	v7Tries, err := FromV6Tries(allTries)
+	require.NoError(t, err)
+	require.NoError(t, StoreCheckpointV7Concurrently(v7Tries, dir, "checkpoint.00000005.v7", logger))
+
+	// For the spot-check: the surviving B registers in trie2 and their values.
+	lastPaths = pathsB
+	lastValues = make([]ledger.Value, len(pathsB))
+	for i := 0; i < 5; i++ {
+		lastValues[i] = overwrites2[i].Value()
+	}
+	for i := 5; i < len(pathsB); i++ {
+		lastValues[i] = payloadsB[i].Value()
+	}
+	return allTries, trie2, lastPaths, lastValues
+}
+
+// overwriteValues returns new payloads that keep each original payload's key but
+// replace its value with a fresh random value, honoring the MTrie invariant that
+// a path's key never changes across updates.
+func overwriteValues(t *testing.T, originals []ledger.Payload) []ledger.Payload {
+	_, randoms := randNPathPayloads(len(originals))
+	out := make([]ledger.Payload, len(originals))
+	for i, orig := range originals {
+		key, err := orig.Key()
+		require.NoError(t, err)
+		out[i] = *ledger.NewPayload(key, randoms[i].Value())
+	}
+	return out
+}
+
+// TestConvertCheckpointV7ToV6_TopTrieLeaf exercises reconstruction of a leaf that
+// lives in the top-trie part file — a register compactified above the subtrie
+// split. A single-register trie's root is exactly such a leaf, so this
+// deterministically pins the buildTopTriePayloadPool path (which is otherwise only
+// hit by chance on larger random tries).
+func TestConvertCheckpointV7ToV6_TopTrieLeaf(t *testing.T) {
+	unittest.RunWithTempDir(t, func(dir string) {
+		logger := zerolog.Nop()
+
+		paths, payloads := randNPathPayloads(1)
+		single, _, err := trie.NewTrieWithUpdatedRegisters(trie.NewEmptyMTrie(), paths, payloads, true)
+		require.NoError(t, err)
+
+		// The register's payload is sourced from the previous checkpoint; no WAL needed.
+		require.NoError(t, StoreCheckpointV6Concurrently([]*trie.MTrie{single}, dir, "checkpoint.00000000", logger))
+		v7Tries, err := FromV6Tries([]*trie.MTrie{single})
+		require.NoError(t, err)
+		require.NoError(t, StoreCheckpointV7Concurrently(v7Tries, dir, "checkpoint.00000005.v7", logger))
+
+		outDir := path.Join(dir, "out")
+		require.NoError(t, os.MkdirAll(outDir, 0755))
+
+		// Empty WAL range (from > to): the payload comes from the previous checkpoint's top-trie.
+		require.NoError(t, ConvertCheckpointV7ToV6(
+			dir, "checkpoint.00000005.v7", dir, 0, 1, 0, outDir, "checkpoint.00000005", logger, 1))
+
+		require.NoError(t, VerifyCheckpointHashes(logger, outDir, "checkpoint.00000005", 1))
+		recon, err := OpenAndReadCheckpointV6(outDir, "checkpoint.00000005", logger)
+		require.NoError(t, err)
+		require.Len(t, recon, 1)
+		require.Equal(t, single.RootHash(), recon[0].RootHash())
+	})
+}
+
+// TestConvertCheckpointV7ToV6_AutoDiscoverPrev verifies that resolvePrevCheckpoint
+// selects the latest V6 checkpoint strictly below the V7 checkpoint number, and
+// errors when none qualifies.
+func TestConvertCheckpointV7ToV6_AutoDiscoverPrev(t *testing.T) {
+	unittest.RunWithTempDir(t, func(dir string) {
+		logger := zerolog.Nop()
+		tries := createSimpleTrie(t)
+		for _, num := range []int{3, 7} {
+			require.NoError(t, StoreCheckpointV6Concurrently(tries, dir, NumberToFilename(num), logger))
+		}
+
+		got, gotFile, err := resolvePrevCheckpoint(dir, 10, -1)
+		require.NoError(t, err)
+		require.Equal(t, 7, got)
+		require.Equal(t, NumberToFilename(7), gotFile)
+
+		got, gotFile, err = resolvePrevCheckpoint(dir, 5, -1)
+		require.NoError(t, err)
+		require.Equal(t, 3, got)
+		require.Equal(t, NumberToFilename(3), gotFile)
+
+		_, _, err = resolvePrevCheckpoint(dir, 2, -1)
+		require.Error(t, err, "no checkpoint below 2 and no root checkpoint exists")
+
+		// Override is honored and must be below N.
+		got, gotFile, err = resolvePrevCheckpoint(dir, 10, 3)
+		require.NoError(t, err)
+		require.Equal(t, 3, got)
+		require.Equal(t, NumberToFilename(3), gotFile)
+		_, _, err = resolvePrevCheckpoint(dir, 5, 5)
+		require.Error(t, err, "override must be < N")
+	})
+}
+
+// TestConvertCheckpointV7ToV6_FallBackToRoot verifies that resolvePrevCheckpoint
+// falls back to the V6 root checkpoint when no numbered V6 checkpoint qualifies,
+// returning -1 as the number and the root checkpoint filename.
+func TestConvertCheckpointV7ToV6_FallBackToRoot(t *testing.T) {
+	unittest.RunWithTempDir(t, func(dir string) {
+		logger := zerolog.Nop()
+		tries := createSimpleTrie(t)
+
+		// With no numbered checkpoint and no root checkpoint, resolution fails.
+		_, _, err := resolvePrevCheckpoint(dir, 10, -1)
+		require.Error(t, err)
+
+		// Write a V6 root checkpoint; resolution now falls back to it.
+		require.NoError(t, StoreCheckpointV6Concurrently(tries, dir, bootstrap.FilenameWALRootCheckpoint, logger))
+
+		got, gotFile, err := resolvePrevCheckpoint(dir, 10, -1)
+		require.NoError(t, err)
+		require.Equal(t, -1, got)
+		require.Equal(t, bootstrap.FilenameWALRootCheckpoint, gotFile)
+
+		// A qualifying numbered checkpoint takes precedence over the root.
+		require.NoError(t, StoreCheckpointV6Concurrently(tries, dir, NumberToFilename(4), logger))
+		got, gotFile, err = resolvePrevCheckpoint(dir, 10, -1)
+		require.NoError(t, err)
+		require.Equal(t, 4, got)
+		require.Equal(t, NumberToFilename(4), gotFile)
+	})
+}
+
+// TestConvertCheckpointV7ToV6_Validation covers argument and filename validation.
+func TestConvertCheckpointV7ToV6_Validation(t *testing.T) {
+	unittest.RunWithTempDir(t, func(dir string) {
+		logger := zerolog.Nop()
+
+		require.Error(t, ConvertCheckpointV7ToV6(dir, "x.v7", dir, -1, -1, -1, dir, "out", logger, 0),
+			"nWorker=0 must be rejected")
+		require.Error(t, ConvertCheckpointV7ToV6(dir, "x.v7", dir, -1, -1, -1, dir, "out", logger, 17),
+			"nWorker > subtrieCount must be rejected")
+		require.Error(t, ConvertCheckpointV7ToV6(dir, "x.v7", dir, -1, -1, -1, dir, "out"+V7FileSuffix, logger, 4),
+			"output filename with V7 suffix must be rejected")
+		require.Error(t, ConvertCheckpointV7ToV6(dir, "missing.v7", dir, -1, -1, -1, dir, "out", logger, 4),
+			"missing V7 input must be reported")
+	})
+}
+
+// TestRequireV6Filename checks the filename guard.
+func TestRequireV6Filename(t *testing.T) {
+	require.Error(t, requireV6Filename(""))
+	require.Error(t, requireV6Filename("checkpoint.00000005"+V7FileSuffix))
+	require.NoError(t, requireV6Filename("checkpoint.00000005"))
+}