diff --git a/docs/encryption_compacted_restart.md b/docs/encryption_compacted_restart.md
new file mode 100644
index 000000000..0fb363c96
--- /dev/null
+++ b/docs/encryption_compacted_restart.md
@@ -0,0 +1,123 @@
+# Encrypted restart after Raft compaction
+
+## Failure
+
+The sidecar's `raft_applied_index` advances only on sidecar-mutating
+encryption entries. Ordinary writes and writer registrations do not advance
+it. A healthy encrypted cluster can therefore retain a sidecar at index 6
+while the FSM has applied index 877 and its Raft prefix has been compacted.
+The startup gap scanner cannot inspect `(6, 877]` and correctly refuses to
+guess whether the missing prefix contains another encryption change.
+
+## Snapshot Witness
+
+After a successful bootstrap or rotation apply, the FSM captures the last
+sidecar-mutating index and a SHA-256 digest of the replicated sidecar state.
+The digest covers wrapped keys, key purposes, active keys, both envelope
+cutovers, and the sidecar index. It excludes replica-local epochs, creation
+times, and the sidecar serialization version. It is a comparison witness,
+not a replacement for authenticated storage values or an epoch authority.
+
+The witness is an `EKVENCP1` extension of the existing length-delimited v2
+FSM header. Older v2 readers can skip the extension. A full physical restore
+preserves it; the cold-start skip path installs it only after full payload
+CRC and Raft-token validation. A newly captured snapshot reuses the witness
+restored from its predecessor, rather than trusting a new disk-sidecar read
+to rewrite the history represented by that snapshot.
+
+The same witness is synchronously persisted as reserved operational FSM
+metadata after each successful encryption apply, before the apply handler
+returns. It does not issue a user timestamp or nonce. When the cold-start
+skip gate retains a newer durable FSM, it can retain that FSM's newer
+witness only through its durable applied index. A checkpoint ahead of that
+index must be reconciled by normal WAL replay; new snapshot capture refuses
+to pair an unreconciled durable record with an older header. Full restore
+requires its body record to match the header. Native physical snapshots
+preserve the record and the offline decoder classifies it as operational
+metadata rather than an MVCC user version.
+
+The startup guard opens the engine's installed snapshot, not an arbitrary
+file in the snapshot directory. Token index, Raft metadata index, token CRC,
+payload CRC, checkpoint index, sidecar state digest, and envelope cutover
+must agree. Only then does that snapshot cover the compacted prefix. The
+snapshot is selected and opened under `snapshotMu`; full checksum verification
+and reads then use that same descriptor without holding up snapshot work.
+Removing the old path after it has been opened cannot invalidate that reader.
+The
+retained suffix remains subject to the original gap scan; encrypted entries
+are authenticated and decrypted before inspecting their opcode. Missing
+keys or invalid authentication tags are startup failures, never "no hit".
+
+A newer snapshot can compact the retained suffix between snapshot selection
+and its scan. Only `ErrCompacted` from that scan triggers a retry against the
+currently installed, fully reverified snapshot. The verified covered prefix
+must advance, and at most three scans are attempted. Absent or inconsistent
+proof, CRC/authentication errors, a nonadvancing snapshot, or continuous
+compaction past that bound still refuse startup. No compacted history is
+treated as harmless merely because a scan was overtaken.
+Each retry rereads the sidecar because the running Raft apply loop may have
+completed a legitimate rotation in the meantime. A regressing sidecar index
+or read failure refuses startup; refreshed metadata does not authorize an
+epoch reset or bypass the snapshot witness comparison.
+
+The membership/collision, pre-bump epoch rollback, post-replay epoch rollback,
+writer-registration, and physical snapshot recovery paths remain enabled.
+The witness deliberately cannot authorize a nonce epoch reset.
+
+## Compatibility and Scope
+
+Older binaries do not recognize the reserved operational key
+`_meta_encryption_snapshot_checkpoint`. The ability to skip the v2 header
+extension does not establish compatibility with this physical body record.
+Downgrading a store written by this version has not been validated as safe.
+Mixed-version native physical snapshot transfer to an older follower is not
+supported by this change: upgrade the receiving follower before allowing it
+to install a snapshot produced by this version. Do not delete the checkpoint
+record or bypass snapshot, sidecar, or epoch checks to enable a downgrade.
+The complete checkpoint may be followed by future v2 header extensions;
+readers consume those extra header bytes without treating them as body data.
+An incomplete known checkpoint still fails parsing.
+
+An old snapshot without a witness does **not** authorize skipping compacted
+history. Such a node can still refuse startup and needs an independently
+validated operational recovery path. This change does not manufacture
+proof for already-compacted legacy data, silently resync a sidecar, remove
+snapshot checks, or claim that the separate DynamoDB acknowledged-append
+consistency anomaly is fixed. No change from PR #1286 is included.
+
+## Verification
+
+- The compacted-gap regression first failed with the original `(6, 877]`
+  scan; the snapshot-witness guard makes the valid case pass.
+- Negative controls retain refusal for stale sidecar/index, changed key
+  material, future checkpoint, absent/truncated proof, retained rotation,
+  snapshot verification error, mismatched token index/CRC, corrupt physical
+  payload, invalid encrypted-tail tag, and a rolled-back local epoch.
+- Snapshot witness tests cover full restore and the verified header-only
+  skip path, including subsequent snapshot production.
+- When the sidecar index equals or exceeds the engine's applied index, the
+  gap guard does not reopen the snapshot. Engine restore still verifies its
+  full CRC; any nonempty gap retains snapshot verification and suffix scans.
+- `go test . ./kv ./internal/encryption ./internal/raftengine/etcd -count=1`
+  passed. The encryption/snapshot/skip-gate selection also passed with `-race`.
+- Scoped `golangci-lint` passed with Go 1.26.6 (0 issues).
+- Local three-process encrypted workload, six verified leader transfers,
+  snapshot interval 20, then stop/reopen of all three processes preserved
+  all 28 items. This guard-specific run used an explicit HTTP leader map;
+  it is not evidence for the separate no-map acceptance requirement.
+
+## Caller Audit
+
+`rg` audit covered all production readers of the FSM header (`kvFSM.Restore`,
+`ParseSnapshotHeader`, `tsoFSM.Restore`, and startup cutover preflight), the
+verified-header skip path, both encryption apply handlers, the startup gap
+guard and scanner factory, and the scanner's sole per-entry predicate call.
+The old three-result header API remains available to callers that do not
+consume the extension. The pure `GuardSidecarBehindRaftLog` API is unchanged.
+The no-gap fast path keeps the original sidecar index rather than advancing
+it. Its sole production caller still invokes the pure gap guard afterward;
+the engine's full-restore and header-only-skip CRC checks are unchanged.
+The snapshot-gap retry helper has one production caller, the startup guard.
+Each retry recomputes the gap from a freshly read sidecar and a freshly verified
+installed snapshot; only the typed suffix-compaction error is retried. The
+pure guard and all its other callers retain their original semantics.
diff --git a/internal/backup/snapshot_reader.go b/internal/backup/snapshot_reader.go
index 653d23bf9..719e0fbee 100644
--- a/internal/backup/snapshot_reader.go
+++ b/internal/backup/snapshot_reader.go
@@ -272,6 +272,7 @@ var pebbleOperationalKeys = [][]byte{
 	[]byte("_meta_min_retained_ts"),
 	[]byte("_meta_pending_min_retained_ts"),
 	[]byte("_meta_applied_index"),
+	[]byte("_meta_encryption_snapshot_checkpoint"),
 	[]byte("_migack"),
 	[]byte("_mighlc"),
 	[]byte("_migpromote"),
diff --git a/internal/encryption/snapshot_checkpoint.go b/internal/encryption/snapshot_checkpoint.go
new file mode 100644
index 000000000..8c235addf
--- /dev/null
+++ b/internal/encryption/snapshot_checkpoint.go
@@ -0,0 +1,46 @@
+package encryption
+
+import (
+	"crypto/sha256"
+	"encoding/json"
+	"maps"
+
+	"github.com/cockroachdb/errors"
+)
+
+// SnapshotCheckpoint witnesses the last sidecar-mutating entry represented
+// by an FSM snapshot. It is meaningful only inside a verified Raft snapshot.
+type SnapshotCheckpoint struct {
+	Index uint64
+	State [sha256.Size]byte
+}
+
+// SidecarCheckpoint excludes node-local nonce epochs and creation times, which
+// differ between replicas. Epoch rollback is checked separately against the
+// writer registry, both before the startup bump and after replay.
+func SidecarCheckpoint(sc *Sidecar) (SnapshotCheckpoint, error) {
+	if sc == nil {
+		return SnapshotCheckpoint{}, errors.New("encryption snapshot checkpoint: missing sidecar")
+	}
+	state := *sc
+	state.Version = 0
+	state.Keys = maps.Clone(sc.Keys)
+	for id, key := range state.Keys {
+		key.LocalEpoch = 0
+		key.Created = ""
+		state.Keys[id] = key
+	}
+	payload, err := json.Marshal(state)
+	if err != nil {
+		return SnapshotCheckpoint{}, errors.Wrap(err, "encode encryption snapshot checkpoint")
+	}
+	return SnapshotCheckpoint{Index: sc.RaftAppliedIndex, State: sha256.Sum256(payload)}, nil
+}
+
+func (a *Applier) SnapshotCheckpoint() (SnapshotCheckpoint, error) {
+	sc, err := ReadSidecar(a.sidecarPath)
+	if err != nil {
+		return SnapshotCheckpoint{}, err
+	}
+	return SidecarCheckpoint(sc)
+}
diff --git a/internal/raftengine/etcd/encryption_checkpoint_test.go b/internal/raftengine/etcd/encryption_checkpoint_test.go
new file mode 100644
index 000000000..15b125f53
--- /dev/null
+++ b/internal/raftengine/etcd/encryption_checkpoint_test.go
@@ -0,0 +1,74 @@
+package etcd
+
+import (
+	"io"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/bootjp/elastickv/internal/encryption/fsmwire"
+	"github.com/stretchr/testify/require"
+	etcdraft "go.etcd.io/raft/v3"
+	raftpb "go.etcd.io/raft/v3/raftpb"
+)
+
+func TestEncryptionScannerEncryptedRetainedTail(t *testing.T) {
+	for _, corrupt := range []bool{false, true} {
+		t.Run(map[bool]string{false: "rotation", true: "invalid-tag"}[corrupt], func(t *testing.T) {
+			cipher, id := raftCipherFixture(t)
+			entry := envelopeEntry(t, cipher, id, 2, []byte{fsmwire.OpRotation})
+			if corrupt {
+				entry.Data[len(entry.Data)-1] ^= 1
+			}
+			engine := newTestEngine(nil, cipher, func() uint64 { return 1 })
+			engine.storage = scannerWithEntries(t, newEntry(1, fsmwire.OpBootstrap), entry).storage
+			hit, err := engine.EncryptionScanner().HasEncryptionRelevantEntryInRange(1, 2)
+			if corrupt {
+				require.Error(t, err, "ciphertext that cannot be inspected must refuse startup")
+			} else {
+				require.NoError(t, err)
+				require.True(t, hit, "encrypted rotation must not be treated as an ordinary write")
+			}
+		})
+	}
+}
+
+func TestEncryptionSnapshotRequiresMatchingPhysicalEnvelope(t *testing.T) {
+	for _, variant := range []string{"valid", "stale-token", "mismatched-index", "corrupt-payload"} {
+		t.Run(variant, func(t *testing.T) {
+			dir := t.TempDir()
+			payload := []byte("verified FSM snapshot bytes")
+			crc, path := writeFSMFileForTest(t, dir, 100, payload)
+			index := uint64(100)
+			if variant == "stale-token" {
+				crc ^= 1
+			}
+			if variant == "mismatched-index" {
+				index = 99
+			}
+			if variant == "corrupt-payload" {
+				file, err := os.OpenFile(path, os.O_RDWR, 0)
+				require.NoError(t, err)
+				_, err = file.WriteAt([]byte("X"), 0)
+				require.NoError(t, err)
+				require.NoError(t, file.Close())
+			}
+			storage := etcdraft.NewMemoryStorage()
+			require.NoError(t, storage.ApplySnapshot(&raftpb.Snapshot{Data: encodeSnapshotToken(index, crc), Metadata: testSnapshotMetadata(100, 1, nil)}))
+			engine := &Engine{storage: storage, fsmSnapDir: filepath.Clean(dir)}
+			reader, gotIndex, err := engine.EncryptionSnapshot()
+			if variant != "valid" {
+				require.Error(t, err)
+				require.Nil(t, reader)
+				return
+			}
+			require.NoError(t, err)
+			defer reader.Close()
+			require.NoError(t, os.Remove(path), "an opened descriptor remains valid after snapshot purge")
+			got, err := io.ReadAll(reader)
+			require.NoError(t, err)
+			require.Equal(t, payload, got)
+			require.Equal(t, uint64(100), gotIndex)
+		})
+	}
+}
diff --git a/internal/raftengine/etcd/encryption_scanner.go b/internal/raftengine/etcd/encryption_scanner.go
index 47c3e0070..4d4f90e3c 100644
--- a/internal/raftengine/etcd/encryption_scanner.go
+++ b/internal/raftengine/etcd/encryption_scanner.go
@@ -42,6 +42,8 @@ const scanMaxBytes uint64 = 64 << 20 // 64 MiB
 // operator's primary problem; the scanner just reports honestly.
 type encryptionScanner struct {
 	storage *etcdraft.MemoryStorage
+	cipher  *encryption.Cipher
+	cutover RaftCutoverIndex
 }
 
 // HasEncryptionRelevantEntryInRange satisfies the
@@ -101,7 +103,11 @@ func (s *encryptionScanner) HasEncryptionRelevantEntryInRange(startExclusive, en
 				cursor, hi))
 		}
 		for _, ent := range batch {
-			if isEncryptionRelevantEntry(ent) {
+			hit, err := s.isEncryptionRelevantEntry(ent)
+			if err != nil {
+				return false, err
+			}
+			if hit {
 				return true, nil
 			}
 		}
@@ -128,15 +134,25 @@ func (s *encryptionScanner) HasEncryptionRelevantEntryInRange(startExclusive, en
 // version 0x01 — NEVER the FSM opcode. A scanner that misreads
 // the layout would return false for every entry in a real Raft
 // log, silently defeating the §9.1 ErrSidecarBehindRaftLog guard.
-func isEncryptionRelevantEntry(ent *raftpb.Entry) bool {
+func (s *encryptionScanner) isEncryptionRelevantEntry(ent *raftpb.Entry) (bool, error) {
 	if ent.GetType() != raftpb.EntryNormal {
-		return false
+		return false, nil
 	}
 	_, payload, ok := decodeProposalEnvelope(ent.GetData())
 	if !ok || len(payload) == 0 {
-		return false
+		return false, nil
+	}
+	if s.cutover != nil && ent.GetIndex() > s.cutover() {
+		if s.cipher == nil {
+			return false, errors.Wrap(ErrRaftUnwrapFailed, "encryption scanner: cipher missing after cutover")
+		}
+		var err error
+		payload, err = unwrapRaftPayload(s.cipher, payload)
+		if err != nil {
+			return false, errors.Wrap(err, "encryption scanner: unwrap retained entry")
+		}
 	}
-	return encryption.IsEncryptionRelevantOpcode(payload[0])
+	return len(payload) > 0 && encryption.IsEncryptionRelevantOpcode(payload[0]), nil
 }
 
 // EncryptionScanner returns the engine's
@@ -161,5 +177,5 @@ func (e *Engine) EncryptionScanner() encryption.EncryptionRelevantScanner {
 	if e == nil {
 		return &encryptionScanner{}
 	}
-	return &encryptionScanner{storage: e.storage}
+	return &encryptionScanner{storage: e.storage, cipher: e.raftCipher, cutover: e.raftCutoverIndex}
 }
diff --git a/internal/raftengine/etcd/encryption_snapshot.go b/internal/raftengine/etcd/encryption_snapshot.go
new file mode 100644
index 000000000..dc4658d35
--- /dev/null
+++ b/internal/raftengine/etcd/encryption_snapshot.go
@@ -0,0 +1,98 @@
+package etcd
+
+import (
+	"bytes"
+	"io"
+	"os"
+
+	"github.com/cockroachdb/errors"
+	etcdraft "go.etcd.io/raft/v3"
+)
+
+// EncryptionSnapshot returns only the snapshot currently installed in this
+// engine, never an unreferenced/newer file found by directory enumeration.
+// The caller may use its metadata only after the token and full payload agree.
+func (e *Engine) EncryptionSnapshot() (io.ReadCloser, uint64, error) {
+	if e == nil || e.storage == nil {
+		return nil, 0, errors.New("encryption snapshot: engine storage unavailable")
+	}
+	source, err := e.openEncryptionSnapshotSource()
+	if err != nil {
+		return nil, 0, err
+	}
+	if source.file == nil {
+		if source.index == 0 {
+			return nil, 0, nil
+		}
+		return io.NopCloser(bytes.NewReader(source.legacy)), source.index, nil
+	}
+	file := source.file
+	if err := verifyEncryptionSnapshotDescriptor(file, file.Name(), source.crc); err != nil {
+		_ = file.Close()
+		return nil, 0, err
+	}
+	info, err := file.Stat()
+	if err != nil {
+		_ = file.Close()
+		return nil, 0, errors.WithStack(err)
+	}
+	return &limitedReadCloser{Reader: io.LimitReader(file, info.Size()-fsmFooterSize), Closer: file}, source.index, nil
+}
+
+type encryptionSnapshotSource struct {
+	file   *os.File
+	legacy []byte
+	index  uint64
+	crc    uint32
+}
+
+func (e *Engine) openEncryptionSnapshotSource() (encryptionSnapshotSource, error) {
+	e.snapshotMu.Lock()
+	defer e.snapshotMu.Unlock()
+	var source encryptionSnapshotSource
+	snapshot, err := e.storage.Snapshot()
+	if err != nil {
+		return source, errors.Wrap(err, "encryption snapshot: read installed snapshot")
+	}
+	if etcdraft.IsEmptySnap(snapshot) || len(snapshot.Data) == 0 {
+		return source, nil
+	}
+	source.index = snapshot.GetMetadata().GetIndex()
+	if !isSnapshotToken(snapshot.Data) {
+		source.legacy = snapshot.Data
+		return source, nil
+	}
+	tok, err := decodeSnapshotToken(snapshot.Data)
+	if err != nil {
+		return source, err
+	}
+	if tok.Index != source.index {
+		return source, errors.Wrap(ErrFSMSnapshotTokenInvalid, "encryption snapshot: token/metadata index mismatch")
+	}
+	source.crc = tok.CRC32C
+	source.file, err = os.Open(fsmSnapPath(e.fsmSnapDir, source.index))
+	if err != nil {
+		return source, statFSMFileError(err)
+	}
+	return source, nil
+}
+
+func verifyEncryptionSnapshotDescriptor(file *os.File, path string, tokenCRC uint32) error {
+	info, err := file.Stat()
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	footer, err := verifyFSMSnapshotPrefix(file, info.Size(), path, tokenCRC)
+	if err != nil {
+		return err
+	}
+	computed, err := computeFSMSnapshotPayloadCRC(file, info.Size())
+	if err != nil {
+		return err
+	}
+	if computed != footer {
+		return errors.Wrap(ErrFSMSnapshotFileCRC, "encryption snapshot payload checksum mismatch")
+	}
+	_, err = file.Seek(0, io.SeekStart)
+	return errors.WithStack(err)
+}
diff --git a/internal/raftengine/etcd/wal_store.go b/internal/raftengine/etcd/wal_store.go
index 32c7cc88f..876cc6a1e 100644
--- a/internal/raftengine/etcd/wal_store.go
+++ b/internal/raftengine/etcd/wal_store.go
@@ -542,7 +542,16 @@ func applyHeaderStateOnSkip(fsm StateMachine, snapPath string, tokenCRC uint32)
 		return errors.Wrapf(ErrFSMSnapshotFileCRC,
 			"path=%s footer=%08x computed=%08x", snapPath, footer, computed)
 	}
+	return applyVerifiedSnapshotHeader(fsm, file, info.Size()-fsmFooterSize)
+}
 
+func applyVerifiedSnapshotHeader(fsm StateMachine, file *os.File, payloadSize int64) error {
+	if setter, ok := fsm.(interface{ RestoreSnapshotHeader(io.Reader) error }); ok {
+		if _, err := file.Seek(0, io.SeekStart); err != nil {
+			return errors.WithStack(err)
+		}
+		return errors.Wrap(setter.RestoreSnapshotHeader(io.LimitReader(file, payloadSize)), "restore verified snapshot header")
+	}
 	setter, hasSetter := fsm.(raftengine.SnapshotHeaderApplier)
 	if !hasSetter {
 		return nil
@@ -551,7 +560,6 @@ func applyHeaderStateOnSkip(fsm StateMachine, snapPath string, tokenCRC uint32)
 	if _, err := file.Seek(0, io.SeekStart); err != nil {
 		return errors.WithStack(err)
 	}
-	payloadSize := info.Size() - fsmFooterSize
 	ceiling, cutover, err := setter.ParseSnapshotHeader(io.LimitReader(file, payloadSize))
 	if err != nil {
 		return errors.WithStack(err)
diff --git a/keyviz/flusher_test.go b/keyviz/flusher_test.go
index df82a1de2..24abcf590 100644
--- a/keyviz/flusher_test.go
+++ b/keyviz/flusher_test.go
@@ -62,25 +62,88 @@ func TestRunFlusherAttachesAlignedHotKeysWindow(t *testing.T) {
 		RunFlusher(ctx, s, s.Step())
 		done <- struct{}{}
 	}()
+	t.Cleanup(func() {
+		cancel()
+		for range 2 {
+			select {
+			case <-done:
+			case <-time.After(time.Second):
+				t.Fatal("hot-key workers did not return after cancel")
+			}
+		}
+	})
 	for i := 0; i < 8; i++ {
 		s.Observe(1, []byte("hot"), OpWrite, 0, LabelLegacy)
 	}
 
 	require.Eventually(t, func() bool {
-		for _, col := range s.Snapshot(time.Time{}, time.Time{}) {
-			if len(col.HotKeys) == 0 {
+		// Keep producing while polling: the 16-column history retains only
+		// 80ms, so a scheduler pause can evict every initial observation.
+		s.Observe(1, []byte("hot"), OpWrite, 0, LabelLegacy)
+		return alignedHotKeysWindowObserved(s.Snapshot(time.Time{}, time.Time{}))
+	}, time.Second, 5*time.Millisecond)
+}
+
+func alignedHotKeysWindowObserved(columns []MatrixColumn) bool {
+	observed := false
+	for _, col := range columns {
+		for _, snapshot := range col.HotKeys {
+			if len(snapshot.Entries) == 0 {
 				continue
 			}
-			snapshot := col.HotKeys[0]
-			return snapshot.WindowStart.Equal(col.WindowStart) &&
-				snapshot.WindowEnd.Equal(col.At) &&
-				len(snapshot.Entries) > 0
+			if !snapshot.WindowStart.Equal(col.WindowStart) || !snapshot.WindowEnd.Equal(col.At) {
+				return false
+			}
+			observed = true
 		}
-		return false
-	}, time.Second, 5*time.Millisecond)
-	cancel()
-	<-done
-	<-done
+	}
+	return observed
+}
+
+func TestAlignedHotKeysWindowObservedSkipsEmptyWindows(t *testing.T) {
+	start := time.Unix(1_700_000_000, 0)
+	end := start.Add(5 * time.Millisecond)
+	columns := []MatrixColumn{
+		{WindowStart: start, At: end, HotKeys: []KeyvizHotKeysSnapshot{{WindowStart: start, WindowEnd: end}}},
+		{WindowStart: end, At: end.Add(5 * time.Millisecond), HotKeys: []KeyvizHotKeysSnapshot{{
+			WindowStart: end, WindowEnd: end.Add(5 * time.Millisecond), Entries: []KeyvizHotKeyEntry{{Key: []byte("hot")}},
+		}}},
+	}
+	require.True(t, alignedHotKeysWindowObserved(columns))
+	columns[1].HotKeys[0].WindowStart = start
+	require.False(t, alignedHotKeysWindowObserved(columns), "a populated misaligned window must still fail")
+	require.False(t, alignedHotKeysWindowObserved(columns[:1]), "empty windows cannot satisfy the assertion")
+}
+
+func TestFlusherAlignmentProbeAfterHistoryEviction(t *testing.T) {
+	start := time.Unix(1_700_000_000, 0)
+	step := 5 * time.Millisecond
+	s := NewMemSampler(MemSamplerOptions{
+		Step: step, HistoryColumns: 2, HotKeysEnabled: true,
+		HotKeysPerRoute: 8, HotKeysSampleRate: 1, HotKeysQueueSize: 32,
+		HotKeysMaxKeyLen: 64, Now: func() time.Time { return start },
+	})
+	require.True(t, s.RegisterRoute(1, []byte("a"), []byte("z"), 1))
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		RunHotKeysAggregator(ctx, s)
+	}()
+	t.Cleanup(func() {
+		cancel()
+		<-done
+	})
+	s.Observe(1, []byte("hot"), OpWrite, 0, LabelLegacy)
+	require.True(t, s.flushWindow(ctx, start.Add(step)))
+	require.True(t, alignedHotKeysWindowObserved(s.Snapshot(time.Time{}, time.Time{})))
+	for i := 2; i <= 3; i++ {
+		require.True(t, s.flushWindow(ctx, start.Add(time.Duration(i)*step)))
+	}
+	require.False(t, alignedHotKeysWindowObserved(s.Snapshot(time.Time{}, time.Time{})), "the initial sample has been evicted")
+	s.Observe(1, []byte("hot"), OpWrite, 0, LabelLegacy)
+	require.True(t, s.flushWindow(ctx, start.Add(4*step)))
+	require.True(t, alignedHotKeysWindowObserved(s.Snapshot(time.Time{}, time.Time{})), "an earlier empty window must not hide the new sample")
 }
 
 func TestFlushWindowKeepsHotKeyAndCounterInSameColumn(t *testing.T) {
diff --git a/keyviz/sampler_test.go b/keyviz/sampler_test.go
index 127d46635..2f4a804f4 100644
--- a/keyviz/sampler_test.go
+++ b/keyviz/sampler_test.go
@@ -478,28 +478,30 @@ func TestConcurrentRegisterAndObserveRace(t *testing.T) {
 	t.Parallel()
 	s, _ := newTestSampler(t, MemSamplerOptions{Step: time.Second, HistoryColumns: 8})
 	const routes = 64
+	const observers = 4
+	const rounds = 64
 	var observed atomic.Uint64
 
-	stop := make(chan struct{})
+	start := make(chan struct{})
+	var ready sync.WaitGroup
+	ready.Add(1 + observers)
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
+		ready.Done()
+		<-start
 		for i := uint64(1); i <= routes; i++ {
 			s.RegisterRoute(i, []byte{byte('a' + i%26)}, []byte{byte('a' + (i+1)%26)}, 0)
-			time.Sleep(time.Microsecond)
 		}
 	}()
-	wg.Add(4)
-	for w := 0; w < 4; w++ {
+	wg.Add(observers)
+	for w := 0; w < observers; w++ {
 		go func() {
 			defer wg.Done()
-			for {
-				select {
-				case <-stop:
-					return
-				default:
-				}
+			ready.Done()
+			<-start
+			for range rounds {
 				for i := uint64(1); i <= routes; i++ {
 					s.Observe(i, make([]byte, 0), OpRead, 0, LabelLegacy)
 					observed.Add(1)
@@ -508,12 +510,12 @@ func TestConcurrentRegisterAndObserveRace(t *testing.T) {
 		}()
 	}
 
-	time.Sleep(20 * time.Millisecond)
-	close(stop)
+	ready.Wait()
+	close(start)
 	wg.Wait()
 
-	if observed.Load() == 0 {
-		t.Fatal("observers ran zero iterations — schedule starvation?")
+	if got, want := observed.Load(), uint64(observers*rounds*routes); got != want {
+		t.Fatalf("observer iterations = %d, want %d", got, want)
 	}
 }
 
diff --git a/kv/fsm.go b/kv/fsm.go
index 3c196c1a9..8c706c543 100644
--- a/kv/fsm.go
+++ b/kv/fsm.go
@@ -11,6 +11,7 @@ import (
 	"sync/atomic"
 
 	"github.com/bootjp/elastickv/distribution"
+	"github.com/bootjp/elastickv/internal/encryption"
 	"github.com/bootjp/elastickv/internal/encryption/fsmwire"
 	"github.com/bootjp/elastickv/internal/fskeys"
 	"github.com/bootjp/elastickv/internal/raftengine"
@@ -70,7 +71,8 @@ type kvFSM struct {
 	// cutoverSource provides the writer-side view of the Phase-2
 	// envelope cutover index for snapshot v1/v2 selection (Stage
 	// 8a §3.3). nil = always v1 output.
-	cutoverSource CutoverSource
+	cutoverSource        CutoverSource
+	encryptionCheckpoint encryption.SnapshotCheckpoint
 	// snapLatch enforces the §3.3 no-downgrade invariant: once a
 	// non-zero cutover is observed, all subsequent snapshots from
 	// this FSM emit v2 even if the sidecar later reads 0.
@@ -1015,6 +1017,9 @@ func (f *kvFSM) Snapshot() (raftengine.Snapshot, error) {
 	if err := f.rejectSnapshotWithActiveBackupPin(); err != nil {
 		return nil, err
 	}
+	if err := f.checkSnapshotCheckpointConsistency(); err != nil {
+		return nil, err
+	}
 	snapshot, err := f.store.Snapshot()
 	if err != nil {
 		return nil, errors.WithStack(err)
@@ -1025,13 +1030,14 @@ func (f *kvFSM) Snapshot() (raftengine.Snapshot, error) {
 		cutover = f.cutoverSource.RaftEnvelopeCutoverIndex()
 	}
 	emitted := f.snapLatch.observe(cutover)
-	useV2 := emitted != 0 || f.snapLatch.latched()
+	useV2 := emitted != 0 || f.snapLatch.latched() || f.encryptionCheckpoint.Index != 0
 
 	return &kvFSMSnapshot{
-		snapshot:  snapshot,
-		ceilingMs: hlcCeilingFromHLC(f.hlc),
-		cutover:   emitted,
-		useV2:     useV2,
+		snapshot:   snapshot,
+		ceilingMs:  hlcCeilingFromHLC(f.hlc),
+		cutover:    emitted,
+		useV2:      useV2,
+		checkpoint: f.encryptionCheckpoint,
 	}, nil
 }
 
@@ -1047,7 +1053,7 @@ func (f *kvFSM) Restore(r io.Reader) error {
 	if !ok {
 		br = bufio.NewReader(r)
 	}
-	ceilingU, cutover, err := ReadSnapshotHeader(br)
+	ceilingU, cutover, checkpoint, err := ReadSnapshotHeaderWithCheckpoint(br)
 	if err != nil {
 		return errors.WithStack(err)
 	}
@@ -1061,6 +1067,11 @@ func (f *kvFSM) Restore(r io.Reader) error {
 	if err := f.store.Restore(io.NopCloser(br)); err != nil {
 		return errors.WithStack(err)
 	}
+	checkpoint, err = f.checkpointForDurableFSM(checkpoint, false)
+	if err != nil {
+		return err
+	}
+	f.encryptionCheckpoint = checkpoint
 	// Volatile pins do not travel in a snapshot, and the entries that would
 	// have released them are compacted away by the time this replica installs
 	// one. A pin this node applied before falling behind would otherwise keep
diff --git a/kv/fsm_encryption.go b/kv/fsm_encryption.go
index 7a3c4f5a5..b498cb9df 100644
--- a/kv/fsm_encryption.go
+++ b/kv/fsm_encryption.go
@@ -122,6 +122,9 @@ func (f *kvFSM) applyBootstrap(raftIdx uint64, data []byte) any {
 		return haltErr(errors.Wrap(errors.Mark(err, encryption.ErrEncryptionApply),
 			"kv/fsm: apply bootstrap"))
 	}
+	if err := f.captureEncryptionCheckpoint(raftIdx); err != nil {
+		return haltErr(errors.Mark(err, encryption.ErrEncryptionApply))
+	}
 	return nil
 }
 
@@ -137,6 +140,9 @@ func (f *kvFSM) applyRotation(raftIdx uint64, data []byte) any {
 		return haltErr(errors.Wrap(errors.Mark(err, encryption.ErrEncryptionApply),
 			"kv/fsm: apply rotation"))
 	}
+	if err := f.captureEncryptionCheckpoint(raftIdx); err != nil {
+		return haltErr(errors.Mark(err, encryption.ErrEncryptionApply))
+	}
 	return nil
 }
 
diff --git a/kv/snapshot.go b/kv/snapshot.go
index b88c6eb6a..2a539b1fc 100644
--- a/kv/snapshot.go
+++ b/kv/snapshot.go
@@ -8,6 +8,7 @@ import (
 	"sync"
 	"sync/atomic"
 
+	"github.com/bootjp/elastickv/internal/encryption"
 	"github.com/bootjp/elastickv/internal/raftengine"
 	"github.com/bootjp/elastickv/store"
 	"github.com/cockroachdb/errors"
@@ -109,9 +110,10 @@ type kvFSMSnapshot struct {
 	// useV2 records the v1/v2 selection made at construction time so a
 	// concurrent sidecar mutation after Snapshot() returns cannot flip
 	// the format mid-write.
-	useV2 bool
-	once  sync.Once
-	err   error
+	useV2      bool
+	checkpoint encryption.SnapshotCheckpoint
+	once       sync.Once
+	err        error
 }
 
 func (f *kvFSMSnapshot) WriteTo(w io.Writer) (int64, error) {
@@ -133,8 +135,8 @@ func (f *kvFSMSnapshot) WriteTo(w io.Writer) (int64, error) {
 // and returns the populated prefix length. v2 emits the minimum payload
 // (ceiling+cutover) — forward-compat trailing bytes are a reader-side
 // concern; the writer never emits unused trailing slots.
-func (f *kvFSMSnapshot) encodeHeader() ([hlcSnapshotV2HeaderLen]byte, int) {
-	var hdr [hlcSnapshotV2HeaderLen]byte
+func (f *kvFSMSnapshot) encodeHeader() ([snapshotCheckpointHeaderLen]byte, int) {
+	var hdr [snapshotCheckpointHeaderLen]byte
 	if !f.useV2 {
 		copy(hdr[:8], hlcSnapshotMagic[:])
 		binary.BigEndian.PutUint64(hdr[8:16], uint64(f.ceilingMs)) //nolint:gosec // ceiling is a Unix ms timestamp, always positive
@@ -144,6 +146,13 @@ func (f *kvFSMSnapshot) encodeHeader() ([hlcSnapshotV2HeaderLen]byte, int) {
 	binary.BigEndian.PutUint16(hdr[8:10], uint16(hlcSnapshotV2MinPayload))
 	binary.BigEndian.PutUint64(hdr[10:18], uint64(f.ceilingMs)) //nolint:gosec // same as v1
 	binary.BigEndian.PutUint64(hdr[18:26], f.cutover)
+	if f.checkpoint.Index != 0 {
+		binary.BigEndian.PutUint16(hdr[8:10], uint16(hlcSnapshotV2MinPayload+encryptionCheckpointBytes))
+		copy(hdr[26:34], encryptionCheckpointMagic)
+		binary.BigEndian.PutUint64(hdr[34:42], f.checkpoint.Index)
+		copy(hdr[42:74], f.checkpoint.State[:])
+		return hdr, snapshotCheckpointHeaderLen
+	}
 	return hdr, hlcSnapshotV2HeaderLen
 }
 
@@ -185,6 +194,11 @@ func (f *kvFSMSnapshot) closeSnapshot() error {
 //   - Anything else (including streams shorter than 8 bytes): (0, 0, nil)
 //     and ALL bytes left in the *bufio.Reader for the inner-store path.
 func ReadSnapshotHeader(r *bufio.Reader) (ceiling, cutover uint64, err error) {
+	ceiling, cutover, _, err = ReadSnapshotHeaderWithCheckpoint(r)
+	return ceiling, cutover, err
+}
+
+func ReadSnapshotHeaderWithCheckpoint(r *bufio.Reader) (ceiling, cutover uint64, checkpoint encryption.SnapshotCheckpoint, err error) {
 	// Step 0: peek up to 8 bytes. Short streams (< 8 bytes) fall through
 	// to the headerless-legacy branch with the partial bytes left in
 	// place — this preserves TestFSMSnapshotRestoreSmallLegacy and is
@@ -197,25 +211,26 @@ func ReadSnapshotHeader(r *bufio.Reader) (ceiling, cutover uint64, err error) {
 	peeked, perr := r.Peek(8) //nolint:mnd
 	if len(peeked) < 8 {      //nolint:mnd
 		if perr != nil && !errors.Is(perr, io.EOF) && !errors.Is(perr, io.ErrUnexpectedEOF) {
-			return 0, 0, errors.Wrap(perr, "snapshot header: peek")
+			return 0, 0, checkpoint, errors.Wrap(perr, "snapshot header: peek")
 		}
 		// io.EOF / io.ErrUnexpectedEOF on a short stream: treat as
 		// headerless legacy. Bytes that were peeked stay in r per
 		// bufio.Reader semantics.
-		return 0, 0, nil
+		return 0, 0, checkpoint, nil
 	}
 
 	switch {
 	case isV2Magic(peeked):
 		return readV2(r)
 	case isV1Magic(peeked):
-		return readV1(r)
+		ceiling, err = readV1(r)
+		return ceiling, 0, checkpoint, err
 	case isUnknownEKVTHLC(peeked):
-		return 0, 0, ErrSnapshotHeaderUnknownMagic
+		return 0, 0, checkpoint, ErrSnapshotHeaderUnknownMagic
 	default:
 		// Headerless legacy: leave the peeked bytes in r so the
 		// inner-store path sees the payload from byte 0.
-		return 0, 0, nil
+		return 0, 0, checkpoint, nil
 	}
 }
 
@@ -245,28 +260,29 @@ func isUnknownEKVTHLC(p []byte) bool {
 		p[6] == hlcSnapshotMagicPrefix[6]
 }
 
-func readV1(r *bufio.Reader) (uint64, uint64, error) {
+func readV1(r *bufio.Reader) (uint64, error) {
 	var rest [16]byte //nolint:mnd
 	if _, err := io.ReadFull(r, rest[:]); err != nil {
-		return 0, 0, errors.Wrap(err, "v1 header: read magic+ceiling")
+		return 0, errors.Wrap(err, "v1 header: read magic+ceiling")
 	}
 	ceiling := binary.BigEndian.Uint64(rest[8:16])
-	return ceiling, 0, nil
+	return ceiling, nil
 }
 
-func readV2(r *bufio.Reader) (uint64, uint64, error) {
+func readV2(r *bufio.Reader) (uint64, uint64, encryption.SnapshotCheckpoint, error) {
+	var checkpoint encryption.SnapshotCheckpoint
 	// Consume the 8 magic bytes.
 	if _, err := r.Discard(8); err != nil { //nolint:mnd
-		return 0, 0, errors.Wrap(err, "v2 header: discard magic")
+		return 0, 0, checkpoint, errors.Wrap(err, "v2 header: discard magic")
 	}
 	// Read the 2-byte length prefix.
 	var lenBuf [hlcSnapshotV2LenBytes]byte
 	if _, err := io.ReadFull(r, lenBuf[:]); err != nil {
-		return 0, 0, errors.Wrap(err, "v2 header: read length prefix")
+		return 0, 0, checkpoint, errors.Wrap(err, "v2 header: read length prefix")
 	}
 	plen := int(binary.BigEndian.Uint16(lenBuf[:]))
 	if plen < hlcSnapshotV2MinPayload || plen > maxSnapshotHeaderPayload {
-		return 0, 0, ErrSnapshotHeaderInvalidLength
+		return 0, 0, checkpoint, ErrSnapshotHeaderInvalidLength
 	}
 	// Read exactly plen payload bytes. Parse ceiling/cutover from the
 	// first 16; ignore the remainder (forward-compat extension area).
@@ -276,9 +292,10 @@ func readV2(r *bufio.Reader) (uint64, uint64, error) {
 	var buf [maxSnapshotHeaderPayload]byte
 	payload := buf[:plen]
 	if _, err := io.ReadFull(r, payload); err != nil {
-		return 0, 0, errors.Wrap(err, "v2 header: read payload")
+		return 0, 0, checkpoint, errors.Wrap(err, "v2 header: read payload")
 	}
 	ceiling := binary.BigEndian.Uint64(payload[0:8])
 	cutover := binary.BigEndian.Uint64(payload[8:16])
-	return ceiling, cutover, nil
+	checkpoint, err := decodeSnapshotCheckpoint(payload)
+	return ceiling, cutover, checkpoint, err
 }
diff --git a/kv/snapshot_encryption_checkpoint.go b/kv/snapshot_encryption_checkpoint.go
new file mode 100644
index 000000000..c2e5fb96f
--- /dev/null
+++ b/kv/snapshot_encryption_checkpoint.go
@@ -0,0 +1,124 @@
+package kv
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/binary"
+	"io"
+
+	"github.com/bootjp/elastickv/internal/encryption"
+	"github.com/cockroachdb/errors"
+)
+
+var encryptionCheckpointMagic = []byte("EKVENCP1")
+
+const encryptionCheckpointBytes = 8 + 8 + 32
+const snapshotCheckpointHeaderLen = hlcSnapshotV2HeaderLen + encryptionCheckpointBytes
+
+func decodeSnapshotCheckpoint(payload []byte) (encryption.SnapshotCheckpoint, error) {
+	var checkpoint encryption.SnapshotCheckpoint
+	if len(payload) < hlcSnapshotV2MinPayload+len(encryptionCheckpointMagic) ||
+		!bytes.Equal(payload[hlcSnapshotV2MinPayload:hlcSnapshotV2MinPayload+len(encryptionCheckpointMagic)], encryptionCheckpointMagic) {
+		return checkpoint, nil
+	}
+	if len(payload) < hlcSnapshotV2MinPayload+encryptionCheckpointBytes {
+		return checkpoint, errors.New("snapshot encryption checkpoint: invalid length")
+	}
+	checkpoint.Index = binary.BigEndian.Uint64(payload[24:32])
+	copy(checkpoint.State[:], payload[32:64])
+	if checkpoint.Index == 0 {
+		return checkpoint, errors.New("snapshot encryption checkpoint: zero index")
+	}
+	return checkpoint, nil
+}
+
+func (f *kvFSM) captureEncryptionCheckpoint(index uint64) error {
+	source, ok := f.encryption.(interface {
+		SnapshotCheckpoint() (encryption.SnapshotCheckpoint, error)
+	})
+	if !ok || index == 0 {
+		return nil
+	}
+	checkpoint, err := source.SnapshotCheckpoint()
+	if err != nil {
+		return errors.Wrap(err, "capture encryption snapshot checkpoint")
+	}
+	// During WAL replay the sidecar can already represent a later entry.
+	// Do not claim that future state in a snapshot of an earlier prefix.
+	if checkpoint.Index == index {
+		if sink, ok := f.store.(interface {
+			SaveEncryptionSnapshotCheckpoint(encryption.SnapshotCheckpoint) error
+		}); ok {
+			if err := sink.SaveEncryptionSnapshotCheckpoint(checkpoint); err != nil {
+				return errors.Wrap(err, "persist applied encryption checkpoint")
+			}
+		}
+		f.encryptionCheckpoint = checkpoint
+	}
+	return nil
+}
+
+// RestoreSnapshotHeader is used only after the engine has verified the whole
+// physical snapshot against its Raft token. Parse errors leave state untouched.
+func (f *kvFSM) RestoreSnapshotHeader(r io.Reader) error {
+	ceiling, cutover, checkpoint, err := ReadSnapshotHeaderWithCheckpoint(bufio.NewReader(r))
+	if err != nil {
+		return err
+	}
+	checkpoint, err = f.checkpointForDurableFSM(checkpoint, true)
+	if err != nil {
+		return err
+	}
+	f.ApplySnapshotHeader(ceiling, cutover)
+	f.encryptionCheckpoint = checkpoint
+	return nil
+}
+
+func (f *kvFSM) checkpointForDurableFSM(snapshot encryption.SnapshotCheckpoint, skip bool) (encryption.SnapshotCheckpoint, error) {
+	source, ok := f.store.(interface {
+		LoadEncryptionSnapshotCheckpoint() (encryption.SnapshotCheckpoint, bool, error)
+	})
+	if !ok {
+		return snapshot, nil
+	}
+	stored, present, err := source.LoadEncryptionSnapshotCheckpoint()
+	if err != nil {
+		return snapshot, errors.Wrap(err, "load durable FSM encryption checkpoint")
+	}
+	if !present {
+		return snapshot, nil
+	}
+	if stored == snapshot {
+		return stored, nil
+	}
+	if !skip || stored.Index <= snapshot.Index {
+		return snapshot, errors.New("encryption checkpoint: physical snapshot header and FSM record disagree")
+	}
+	applied, present, err := f.LastAppliedIndex()
+	if err != nil {
+		return snapshot, err
+	}
+	if present && stored.Index <= applied {
+		return stored, nil
+	}
+	// The durable checkpoint may precede a crash before the applied-index
+	// marker. That tail is not skipped and must replay through the handler.
+	return snapshot, nil
+}
+
+func (f *kvFSM) checkSnapshotCheckpointConsistency() error {
+	source, ok := f.store.(interface {
+		LoadEncryptionSnapshotCheckpoint() (encryption.SnapshotCheckpoint, bool, error)
+	})
+	if !ok {
+		return nil
+	}
+	stored, present, err := source.LoadEncryptionSnapshotCheckpoint()
+	if err != nil {
+		return errors.Wrap(err, "verify snapshot checkpoint before capture")
+	}
+	if present && stored != f.encryptionCheckpoint {
+		return errors.New("encryption checkpoint: durable tail must reconcile before snapshot capture")
+	}
+	return nil
+}
diff --git a/kv/snapshot_encryption_checkpoint_test.go b/kv/snapshot_encryption_checkpoint_test.go
new file mode 100644
index 000000000..c0fad08b3
--- /dev/null
+++ b/kv/snapshot_encryption_checkpoint_test.go
@@ -0,0 +1,156 @@
+package kv
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/binary"
+	"io"
+	"testing"
+
+	"github.com/bootjp/elastickv/internal/encryption"
+	"github.com/bootjp/elastickv/store"
+	"github.com/stretchr/testify/require"
+)
+
+type checkpointApplier struct {
+	fakeApplier
+	checkpoint encryption.SnapshotCheckpoint
+}
+
+func (a *checkpointApplier) SnapshotCheckpoint() (encryption.SnapshotCheckpoint, error) {
+	return a.checkpoint, nil
+}
+
+func TestEncryptionSnapshotCheckpointRoundTrip(t *testing.T) {
+	cp := encryption.SnapshotCheckpoint{Index: 6, State: [32]byte{1, 2, 3}}
+	for _, headerOnly := range []bool{false, true} {
+		t.Run(map[bool]string{false: "full-physical-restore", true: "skip-gate-header-restore"}[headerOnly], func(t *testing.T) {
+			source, ok := NewKvFSMWithHLC(store.NewMVCCStore(), NewHLC()).(*kvFSM)
+			require.True(t, ok)
+			source.encryption = &checkpointApplier{checkpoint: cp}
+			require.NoError(t, source.captureEncryptionCheckpoint(6))
+			snapshot, err := source.Snapshot()
+			require.NoError(t, err)
+			defer snapshot.Close()
+			var payload bytes.Buffer
+			_, err = snapshot.WriteTo(&payload)
+			require.NoError(t, err)
+			_, _, parsed, err := ReadSnapshotHeaderWithCheckpoint(bufio.NewReader(bytes.NewReader(payload.Bytes())))
+			require.NoError(t, err)
+			require.Equal(t, cp, parsed)
+			target, ok := NewKvFSMWithHLC(store.NewMVCCStore(), NewHLC()).(*kvFSM)
+			require.True(t, ok)
+			if headerOnly {
+				err = target.RestoreSnapshotHeader(bytes.NewReader(payload.Bytes()))
+			} else {
+				err = target.Restore(bytes.NewReader(payload.Bytes()))
+			}
+			require.NoError(t, err)
+			require.Equal(t, cp, target.encryptionCheckpoint)
+			// A rollback on disk must not overwrite the restored witness merely
+			// because this process starts making newer snapshots.
+			target.encryption = &checkpointApplier{checkpoint: encryption.SnapshotCheckpoint{Index: 5}}
+			require.NoError(t, target.captureEncryptionCheckpoint(4))
+			require.Equal(t, cp, target.encryptionCheckpoint)
+			next, err := target.Snapshot()
+			require.NoError(t, err)
+			defer next.Close()
+			var nextPayload bytes.Buffer
+			_, err = next.WriteTo(&nextPayload)
+			require.NoError(t, err)
+			_, _, carried, err := ReadSnapshotHeaderWithCheckpoint(bufio.NewReader(&nextPayload))
+			require.NoError(t, err)
+			require.Equal(t, cp, carried)
+		})
+	}
+}
+
+func TestEncryptionSnapshotCheckpointRejectsTruncationWithoutMutation(t *testing.T) {
+	fsm, ok := NewKvFSMWithHLC(store.NewMVCCStore(), NewHLC()).(*kvFSM)
+	require.True(t, ok)
+	fsm.encryptionCheckpoint = encryption.SnapshotCheckpoint{Index: 6, State: [32]byte{1}}
+	snapshot, err := fsm.Snapshot()
+	require.NoError(t, err)
+	defer snapshot.Close()
+	var payload bytes.Buffer
+	_, err = snapshot.WriteTo(&payload)
+	require.NoError(t, err)
+	before := fsm.encryptionCheckpoint
+	require.Error(t, fsm.RestoreSnapshotHeader(bytes.NewReader(payload.Bytes()[:snapshotCheckpointHeaderLen-1])))
+	require.Equal(t, before, fsm.encryptionCheckpoint)
+}
+
+func TestEncryptionCheckpointAllowsTrailingV2Extensions(t *testing.T) {
+	cp := encryption.SnapshotCheckpoint{Index: 6, State: [32]byte{1, 2, 3}}
+	header := &kvFSMSnapshot{useV2: true, checkpoint: cp}
+	encoded, size := header.encodeHeader()
+	const extension = "FUTURE01"
+	payload := append(bytes.Clone(encoded[:size]), extension...)
+	binary.BigEndian.PutUint16(payload[8:10], uint16(hlcSnapshotV2MinPayload+encryptionCheckpointBytes+len(extension)))
+	body := []byte("unchanged physical snapshot body")
+	reader := bufio.NewReader(bytes.NewReader(append(payload, body...)))
+	_, _, got, err := ReadSnapshotHeaderWithCheckpoint(reader)
+	require.NoError(t, err)
+	require.Equal(t, cp, got)
+	remaining, err := io.ReadAll(reader)
+	require.NoError(t, err)
+	require.Equal(t, body, remaining)
+}
+
+type durableCheckpointFixture struct {
+	store.MVCCStore
+	checkpoint encryption.SnapshotCheckpoint
+	applied    uint64
+}
+
+func (s *durableCheckpointFixture) LoadEncryptionSnapshotCheckpoint() (encryption.SnapshotCheckpoint, bool, error) {
+	return s.checkpoint, true, nil
+}
+
+func (s *durableCheckpointFixture) LastAppliedIndex() (uint64, bool, error) {
+	return s.applied, true, nil
+}
+
+func TestEncryptionCheckpointSkipPreservesPostSnapshotRotation(t *testing.T) {
+	old := encryption.SnapshotCheckpoint{Index: 6, State: [32]byte{6}}
+	newer := encryption.SnapshotCheckpoint{Index: 890, State: [32]byte{8}}
+	header := &kvFSMSnapshot{useV2: true, checkpoint: old}
+	encoded, size := header.encodeHeader()
+	for _, applied := range []uint64{900, 889} {
+		t.Run(map[bool]string{true: "durable-tail-skipped", false: "unapplied-tail-must-replay"}[applied == 900], func(t *testing.T) {
+			st := &durableCheckpointFixture{MVCCStore: store.NewMVCCStore(), checkpoint: newer, applied: applied}
+			fsm, ok := NewKvFSMWithHLC(st, NewHLC()).(*kvFSM)
+			require.True(t, ok)
+			require.NoError(t, fsm.RestoreSnapshotHeader(bytes.NewReader(encoded[:size])))
+			want := old
+			if applied == 900 {
+				want = newer
+			}
+			require.Equal(t, want, fsm.encryptionCheckpoint)
+			next, err := fsm.Snapshot()
+			if applied == 889 {
+				require.Error(t, err, "the future durable record must replay before any new snapshot")
+				return
+			}
+			require.NoError(t, err)
+			defer next.Close()
+			var out bytes.Buffer
+			_, err = next.WriteTo(&out)
+			require.NoError(t, err)
+			_, _, cp, err := ReadSnapshotHeaderWithCheckpoint(bufio.NewReader(&out))
+			require.NoError(t, err)
+			require.Equal(t, want, cp)
+		})
+	}
+}
+
+func TestEncryptionCheckpointSkipRejectsChangedDurableRecord(t *testing.T) {
+	old := encryption.SnapshotCheckpoint{Index: 6, State: [32]byte{6}}
+	header := &kvFSMSnapshot{useV2: true, checkpoint: old}
+	encoded, size := header.encodeHeader()
+	st := &durableCheckpointFixture{MVCCStore: store.NewMVCCStore(), checkpoint: encryption.SnapshotCheckpoint{Index: 6, State: [32]byte{7}}, applied: 900}
+	fsm, ok := NewKvFSMWithHLC(st, NewHLC()).(*kvFSM)
+	require.True(t, ok)
+	require.Error(t, fsm.RestoreSnapshotHeader(bytes.NewReader(encoded[:size])))
+	require.Zero(t, fsm.encryptionCheckpoint)
+}
diff --git a/main_encryption_compacted_restart_test.go b/main_encryption_compacted_restart_test.go
new file mode 100644
index 000000000..01d06d4b6
--- /dev/null
+++ b/main_encryption_compacted_restart_test.go
@@ -0,0 +1,241 @@
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/bootjp/elastickv/internal/encryption"
+	"github.com/bootjp/elastickv/kv"
+	"github.com/bootjp/elastickv/store"
+	"github.com/stretchr/testify/require"
+	etcdraft "go.etcd.io/raft/v3"
+)
+
+type checkpointGapEngine struct {
+	stubGapEngine
+	payload       []byte
+	snapshotIndex uint64
+	snapshotErr   error
+	snapshotCalls int
+}
+
+func (e *checkpointGapEngine) EncryptionSnapshot() (io.ReadCloser, uint64, error) {
+	e.snapshotCalls++
+	return io.NopCloser(bytes.NewReader(e.payload)), e.snapshotIndex, e.snapshotErr
+}
+
+func TestEncryptionCheckpointReadsSnapshotOnlyForGap(t *testing.T) {
+	for _, applied := range []uint64{5, 6, 7} {
+		t.Run(fmt.Sprintf("applied-%d", applied), func(t *testing.T) {
+			engine := &checkpointGapEngine{
+				stubGapEngine: stubGapEngine{appliedIndex: applied},
+				snapshotErr:   io.ErrUnexpectedEOF,
+			}
+			sc := &encryption.Sidecar{RaftAppliedIndex: 6}
+			start, err := sidecarGapStartFromSnapshot(engine, sc)
+			if applied > sc.RaftAppliedIndex {
+				require.ErrorIs(t, err, io.ErrUnexpectedEOF)
+				require.Equal(t, 1, engine.snapshotCalls)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, sc.RaftAppliedIndex, start)
+			require.Zero(t, engine.snapshotCalls)
+		})
+	}
+}
+
+type checkpointScanner struct {
+	t          *testing.T
+	start, end uint64
+	hit        bool
+}
+
+type advancingCheckpointScanner struct {
+	t       *testing.T
+	engine  *checkpointGapEngine
+	variant string
+	calls   int
+	path    string
+}
+
+func (s *advancingCheckpointScanner) HasEncryptionRelevantEntryInRange(start, end uint64) (bool, error) {
+	s.calls++
+	require.Equal(s.t, uint64(877), end)
+	if s.variant == "keeps-compacting" {
+		require.Equal(s.t, s.engine.snapshotIndex, start)
+		s.engine.snapshotIndex++
+		return false, etcdraft.ErrCompacted
+	}
+	if s.calls > 1 {
+		require.Equal(s.t, uint64(870), start)
+		return s.variant == "retained-rotation", nil
+	}
+	require.Equal(s.t, uint64(860), start)
+	if s.variant == "unavailable" {
+		return false, etcdraft.ErrUnavailable
+	}
+	if s.variant != "unchanged" {
+		s.engine.snapshotIndex = 870
+	}
+	s.prepareNextSnapshot()
+	return false, etcdraft.ErrCompacted
+}
+
+func (s *advancingCheckpointScanner) prepareNextSnapshot() {
+	switch s.variant {
+	case "missing-proof":
+		s.engine.payload = []byte("EKVTHLC1\x00\x00\x00\x00\x00\x00\x00\x01")
+	case "changed-proof":
+		s.engine.payload[len(s.engine.payload)-1] ^= 1
+	case "verification-error":
+		s.engine.snapshotErr = io.ErrUnexpectedEOF
+	case "sidecar-advanced", "sidecar-rollback":
+		sc := &encryption.Sidecar{Version: encryption.SidecarVersion, RaftAppliedIndex: 7}
+		if s.variant == "sidecar-rollback" {
+			sc.RaftAppliedIndex = 5
+		}
+		require.NoError(s.t, encryption.WriteSidecar(s.path, sc))
+		s.engine.payload = checkpointHeader(s.t, sc)
+	case "sidecar-read-error":
+		require.NoError(s.t, os.Remove(s.path))
+	}
+}
+
+func TestEncryptionCheckpointRetriesCompactionOnlyWithNewProof(t *testing.T) {
+	for _, variant := range []string{"advance", "unchanged", "missing-proof", "changed-proof", "verification-error", "retained-rotation", "unavailable", "keeps-compacting", "sidecar-advanced", "sidecar-rollback", "sidecar-read-error"} {
+		t.Run(variant, func(t *testing.T) {
+			sc := &encryption.Sidecar{Version: encryption.SidecarVersion, RaftAppliedIndex: 6}
+			engine := &checkpointGapEngine{
+				stubGapEngine: stubGapEngine{appliedIndex: 877},
+				payload:       checkpointHeader(t, sc), snapshotIndex: 860,
+			}
+			scanner := &advancingCheckpointScanner{t: t, engine: engine, variant: variant}
+			engine.scanner = scanner
+			path := filepath.Join(t.TempDir(), "keys.json")
+			scanner.path = path
+			require.NoError(t, encryption.WriteSidecar(path, sc))
+			err := runSidecarBehindRaftLogGuard(engine, path, 1)
+			if variant == "advance" || variant == "sidecar-advanced" {
+				require.NoError(t, err)
+				require.Equal(t, 2, scanner.calls)
+			} else {
+				require.Error(t, err)
+			}
+			requireCheckpointRetryError(t, variant, err)
+			if variant == "unavailable" {
+				require.Equal(t, 1, scanner.calls)
+			}
+			if variant == "keeps-compacting" {
+				require.Equal(t, 3, scanner.calls, "continuous compaction must stop at the retry bound")
+			}
+		})
+	}
+}
+
+func requireCheckpointRetryError(t *testing.T, variant string, err error) {
+	t.Helper()
+	switch variant {
+	case "unchanged", "missing-proof", "keeps-compacting":
+		require.ErrorIs(t, err, etcdraft.ErrCompacted)
+	case "changed-proof", "retained-rotation", "sidecar-rollback":
+		require.ErrorIs(t, err, encryption.ErrSidecarBehindRaftLog)
+	case "verification-error":
+		require.ErrorIs(t, err, io.ErrUnexpectedEOF)
+	case "sidecar-read-error":
+		require.ErrorIs(t, err, os.ErrNotExist)
+	case "unavailable":
+		require.ErrorIs(t, err, etcdraft.ErrUnavailable)
+	}
+}
+
+func (s *checkpointScanner) HasEncryptionRelevantEntryInRange(start, end uint64) (bool, error) {
+	if start < s.start {
+		return false, etcdraft.ErrCompacted
+	}
+	require.Equal(s.t, s.start, start)
+	require.Equal(s.t, s.end, end)
+	return s.hit, nil
+}
+
+func checkpointHeader(t *testing.T, sc *encryption.Sidecar) []byte {
+	t.Helper()
+	cp, err := encryption.SidecarCheckpoint(sc)
+	require.NoError(t, err)
+	var out bytes.Buffer
+	out.WriteString("EKVTHLC2")
+	require.NoError(t, binary.Write(&out, binary.BigEndian, uint16(64)))
+	require.NoError(t, binary.Write(&out, binary.BigEndian, uint64(1234)))
+	require.NoError(t, binary.Write(&out, binary.BigEndian, sc.RaftEnvelopeCutoverIndex))
+	out.WriteString("EKVENCP1")
+	require.NoError(t, binary.Write(&out, binary.BigEndian, cp.Index))
+	out.Write(cp.State[:])
+	return out.Bytes()
+}
+
+func TestEncryptionCompactedRestartCheckpoint(t *testing.T) {
+	for _, variant := range []string{"valid", "stale-sidecar", "mismatched-keys", "future-checkpoint", "missing-proof", "retained-rotation", "snapshot-verification-error", "truncated-proof"} {
+		t.Run(variant, func(t *testing.T) {
+			sc := &encryption.Sidecar{Version: encryption.SidecarVersion, RaftAppliedIndex: 6,
+				StorageEnvelopeActive: true, StorageEnvelopeCutoverIndex: 5, RaftEnvelopeCutoverIndex: 6,
+				Active: encryption.ActiveKeys{Storage: 1, Raft: 2},
+				Keys:   map[string]encryption.SidecarKey{"1": {Purpose: "storage", Wrapped: []byte("wrapped-storage"), LocalEpoch: 8}, "2": {Purpose: "raft", Wrapped: []byte("wrapped-raft")}}}
+			payload := checkpointHeader(t, sc)
+			engine := &checkpointGapEngine{stubGapEngine: stubGapEngine{appliedIndex: 877,
+				scanner: &checkpointScanner{t: t, start: 860, end: 877}}, payload: payload, snapshotIndex: 860}
+			switch variant {
+			case "stale-sidecar":
+				sc.RaftAppliedIndex = 5
+			case "mismatched-keys":
+				key := sc.Keys["1"]
+				key.Wrapped = []byte("different-wrapped-key")
+				sc.Keys["1"] = key
+			case "future-checkpoint":
+				binary.BigEndian.PutUint64(engine.payload[34:42], 861)
+			case "missing-proof":
+				engine.payload = []byte("EKVTHLC1\x00\x00\x00\x00\x00\x00\x00\x01")
+			case "retained-rotation":
+				scanner, ok := engine.scanner.(*checkpointScanner)
+				require.True(t, ok)
+				scanner.hit = true
+			case "snapshot-verification-error":
+				engine.snapshotErr = io.ErrUnexpectedEOF
+			case "truncated-proof":
+				engine.payload = payload[:len(payload)-1]
+			}
+			path := filepath.Join(t.TempDir(), "keys.json")
+			require.NoError(t, encryption.WriteSidecar(path, sc))
+			err := runSidecarBehindRaftLogGuard(engine, path, 1)
+			if variant == "valid" {
+				require.NoError(t, err, "verified snapshot witnesses the compacted prefix; retained tail still scanned")
+			} else {
+				require.Error(t, err, "uncertain or inconsistent metadata must refuse startup")
+			}
+		})
+	}
+}
+
+func TestEncryptionCheckpointDoesNotAuthorizeEpochRollback(t *testing.T) {
+	sc := &encryption.Sidecar{RaftAppliedIndex: 6, Keys: map[string]encryption.SidecarKey{"1": {LocalEpoch: 8}}}
+	payload := checkpointHeader(t, sc)
+	_, _, before, err := kv.ReadSnapshotHeaderWithCheckpoint(bufio.NewReader(bytes.NewReader(payload)))
+	require.NoError(t, err)
+	key := sc.Keys["1"]
+	key.LocalEpoch = 1
+	sc.Keys["1"] = key
+	after, err := encryption.SidecarCheckpoint(sc)
+	require.NoError(t, err)
+	require.Equal(t, before, after, "snapshot state is replica-independent, not an epoch authorization")
+	st := newRegistrationTestStore(t)
+	preSeedRegistryRow(t, st, 1, 1, 8)
+	reg, err := store.WriterRegistryFor(st)
+	require.NoError(t, err)
+	err = checkLocalEpochRollbackBeforeBump(reg, 1, 1, key.LocalEpoch, true)
+	require.ErrorIs(t, err, encryption.ErrLocalEpochRollback)
+}
diff --git a/main_encryption_snapshot_checkpoint.go b/main_encryption_snapshot_checkpoint.go
new file mode 100644
index 000000000..f804c8c44
--- /dev/null
+++ b/main_encryption_snapshot_checkpoint.go
@@ -0,0 +1,95 @@
+package main
+
+import (
+	"bufio"
+	"io"
+
+	"github.com/bootjp/elastickv/internal/encryption"
+	"github.com/bootjp/elastickv/kv"
+	"github.com/cockroachdb/errors"
+	etcdraft "go.etcd.io/raft/v3"
+)
+
+const maxEncryptionSnapshotGapAttempts = 3
+
+func verifySidecarGapWithSnapshot(engine encryptionGapEngine, sidecarPath string, sc *encryption.Sidecar) error {
+	var previousStart uint64
+	var scanErr error
+	for attempt := range maxEncryptionSnapshotGapAttempts {
+		if attempt > 0 {
+			fresh, err := encryption.ReadSidecar(sidecarPath)
+			if err != nil {
+				return errors.Wrap(err, "encryption startup guard: reload sidecar after compaction")
+			}
+			if fresh.RaftAppliedIndex < sc.RaftAppliedIndex {
+				return errors.Wrap(encryption.ErrSidecarBehindRaftLog, "sidecar index regressed during gap verification")
+			}
+			sc = fresh
+		}
+		start, err := sidecarGapStartFromSnapshot(engine, sc)
+		if err != nil {
+			return err
+		}
+		if scanErr != nil && start <= previousStart {
+			return errors.Wrap(scanErr, "encryption startup guard: snapshot proof did not advance")
+		}
+		scanErr = encryption.GuardSidecarBehindRaftLog(start, engine.AppliedIndex(), engine.EncryptionScanner())
+		if !errors.Is(scanErr, etcdraft.ErrCompacted) {
+			return errors.Wrap(scanErr, "encryption startup guard: scan retained suffix")
+		}
+		// Snapshot work can compact the selected generation during the scan.
+		// Retry only if a freshly verified witness advances its covered prefix.
+		previousStart = start
+	}
+	return errors.Wrap(scanErr, "encryption startup guard: snapshot compaction retry limit")
+}
+
+func sidecarGapStartFromSnapshot(engine encryptionGapEngine, sc *encryption.Sidecar) (uint64, error) {
+	start := sc.RaftAppliedIndex
+	if start >= engine.AppliedIndex() {
+		// Engine restore already validated the installed snapshot. No
+		// compacted gap needs a witness, so do not repeat its full CRC scan.
+		return start, nil
+	}
+	source, ok := engine.(interface {
+		EncryptionSnapshot() (io.ReadCloser, uint64, error)
+	})
+	if !ok {
+		return start, nil
+	}
+	payload, index, err := source.EncryptionSnapshot()
+	if err != nil {
+		return 0, errors.Wrap(err, "encryption startup guard: verify snapshot checkpoint")
+	}
+	if payload == nil {
+		return start, nil
+	}
+	defer payload.Close()
+	if index <= start {
+		return start, nil
+	}
+	if index > engine.AppliedIndex() {
+		return 0, errors.Wrap(encryption.ErrSidecarBehindRaftLog, "snapshot checkpoint exceeds applied index")
+	}
+	return verifySidecarSnapshotCheckpoint(payload, index, sc)
+}
+
+func verifySidecarSnapshotCheckpoint(payload io.Reader, index uint64, sc *encryption.Sidecar) (uint64, error) {
+	_, cutover, checkpoint, err := kv.ReadSnapshotHeaderWithCheckpoint(bufio.NewReader(payload))
+	if err != nil {
+		return 0, errors.Wrap(err, "encryption startup guard: read snapshot checkpoint")
+	}
+	if checkpoint.Index == 0 {
+		// Old snapshots carry no proof about encryption changes in the
+		// compacted prefix. Keep the original complete-scan requirement.
+		return sc.RaftAppliedIndex, nil
+	}
+	want, err := encryption.SidecarCheckpoint(sc)
+	if err != nil {
+		return 0, errors.WithStack(err)
+	}
+	if checkpoint.Index > index || checkpoint != want || cutover != sc.RaftEnvelopeCutoverIndex {
+		return 0, errors.Wrap(encryption.ErrSidecarBehindRaftLog, "snapshot checkpoint disagrees with sidecar")
+	}
+	return index, nil
+}
diff --git a/main_encryption_startup_guard.go b/main_encryption_startup_guard.go
index 13c62aeb3..a8171a25e 100644
--- a/main_encryption_startup_guard.go
+++ b/main_encryption_startup_guard.go
@@ -184,11 +184,7 @@ func runSidecarBehindRaftLogGuard(
 			slog.Uint64("engine_applied_index", gapEngine.AppliedIndex()))
 		return nil
 	}
-	if err := encryption.GuardSidecarBehindRaftLog(
-		sidecar.RaftAppliedIndex,
-		gapEngine.AppliedIndex(),
-		gapEngine.EncryptionScanner(),
-	); err != nil {
+	if err := verifySidecarGapWithSnapshot(gapEngine, sidecarPath, sidecar); err != nil {
 		return pkgerrors.Wrapf(err,
 			"encryption startup guard: sidecar=%q default_group=%d",
 			sidecarPath, defaultGroup)
diff --git a/store/encryption_checkpoint.go b/store/encryption_checkpoint.go
new file mode 100644
index 000000000..242ae8b20
--- /dev/null
+++ b/store/encryption_checkpoint.go
@@ -0,0 +1,54 @@
+package store
+
+import (
+	"encoding/binary"
+
+	"github.com/bootjp/elastickv/internal/encryption"
+	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/pebble/v2"
+)
+
+var encryptionCheckpointKey = []byte("_meta_encryption_snapshot_checkpoint")
+
+const encryptionCheckpointSize = 8 + 32
+
+// SaveEncryptionSnapshotCheckpoint is an operational metadata write, like the
+// writer registry: no user timestamp, nonce, or applied watermark is issued.
+// It is durable before the encryption handler returns and later data entries
+// can advance the durable applied index past it safely.
+func (s *pebbleStore) SaveEncryptionSnapshotCheckpoint(cp encryption.SnapshotCheckpoint) error {
+	s.applyMu.Lock()
+	defer s.applyMu.Unlock()
+	s.dbMu.RLock()
+	defer s.dbMu.RUnlock()
+	if cp.Index == 0 {
+		return errors.New("encryption snapshot checkpoint: zero durable index")
+	}
+	var raw [encryptionCheckpointSize]byte
+	binary.BigEndian.PutUint64(raw[:8], cp.Index)
+	copy(raw[8:], cp.State[:])
+	return errors.Wrap(s.db.Set(encryptionCheckpointKey, raw[:], pebble.Sync), "persist encryption snapshot checkpoint")
+}
+
+func (s *pebbleStore) LoadEncryptionSnapshotCheckpoint() (encryption.SnapshotCheckpoint, bool, error) {
+	s.dbMu.RLock()
+	defer s.dbMu.RUnlock()
+	var cp encryption.SnapshotCheckpoint
+	raw, closer, err := s.db.Get(encryptionCheckpointKey)
+	if errors.Is(err, pebble.ErrNotFound) {
+		return cp, false, nil
+	}
+	if err != nil {
+		return cp, false, errors.Wrap(err, "read durable encryption snapshot checkpoint")
+	}
+	defer closer.Close()
+	if len(raw) != encryptionCheckpointSize {
+		return cp, false, errors.New("encryption snapshot checkpoint: corrupt durable record")
+	}
+	cp.Index = binary.BigEndian.Uint64(raw[:8])
+	copy(cp.State[:], raw[8:])
+	if cp.Index == 0 {
+		return cp, false, errors.New("encryption snapshot checkpoint: zero durable index")
+	}
+	return cp, true, nil
+}
diff --git a/store/encryption_checkpoint_test.go b/store/encryption_checkpoint_test.go
new file mode 100644
index 000000000..a481bbf58
--- /dev/null
+++ b/store/encryption_checkpoint_test.go
@@ -0,0 +1,55 @@
+package store
+
+import (
+	"bytes"
+	"io"
+	"path/filepath"
+	"testing"
+
+	"github.com/bootjp/elastickv/internal/encryption"
+	"github.com/stretchr/testify/require"
+)
+
+func TestEncryptionCheckpointDurableMetadataSurvivesReopenAndSnapshot(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "source")
+	st, err := NewPebbleStore(path)
+	require.NoError(t, err)
+	checkpointStore, ok := st.(interface {
+		SaveEncryptionSnapshotCheckpoint(encryption.SnapshotCheckpoint) error
+	})
+	require.True(t, ok)
+	cp := encryption.SnapshotCheckpoint{Index: 890, State: [32]byte{8, 9}}
+	require.NoError(t, checkpointStore.SaveEncryptionSnapshotCheckpoint(cp))
+	require.Equal(t, uint64(0), st.LastCommitTS(), "operational metadata cannot issue a user timestamp")
+	require.NoError(t, st.Close())
+	st, err = NewPebbleStore(path)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, st.Close()) })
+	loader, ok := st.(interface {
+		LoadEncryptionSnapshotCheckpoint() (encryption.SnapshotCheckpoint, bool, error)
+	})
+	require.True(t, ok)
+	got, present, err := loader.LoadEncryptionSnapshotCheckpoint()
+	require.NoError(t, err)
+	require.True(t, present)
+	require.Equal(t, cp, got)
+	require.True(t, IsPebbleOperationalKey(encryptionCheckpointKey))
+	snap, err := st.Snapshot()
+	require.NoError(t, err)
+	defer snap.Close()
+	var payload bytes.Buffer
+	_, err = snap.WriteTo(&payload)
+	require.NoError(t, err)
+	dst, err := NewPebbleStore(filepath.Join(t.TempDir(), "restored"))
+	require.NoError(t, err)
+	defer dst.Close()
+	require.NoError(t, dst.Restore(io.NopCloser(&payload)))
+	reader, ok := dst.(interface {
+		LoadEncryptionSnapshotCheckpoint() (encryption.SnapshotCheckpoint, bool, error)
+	})
+	require.True(t, ok)
+	got, present, err = reader.LoadEncryptionSnapshotCheckpoint()
+	require.NoError(t, err)
+	require.True(t, present)
+	require.Equal(t, cp, got)
+}
diff --git a/store/lsm_store.go b/store/lsm_store.go
index baed5010b..18cfb151d 100644
--- a/store/lsm_store.go
+++ b/store/lsm_store.go
@@ -727,6 +727,7 @@ func isPebbleMetaKey(rawKey []byte) bool {
 		bytes.Equal(rawKey, metaMinRetainedTSBytes) ||
 		bytes.Equal(rawKey, metaPendingMinRetainedTSBytes) ||
 		bytes.Equal(rawKey, metaAppliedIndexBytes) ||
+		bytes.Equal(rawKey, encryptionCheckpointKey) ||
 		isMigrationMetadataKey(rawKey) ||
 		isPebbleWriterRegistryKey(rawKey)
 }