diff --git a/api.go b/api.go
index eef03e5a..c71b7097 100644
--- a/api.go
+++ b/api.go
@@ -56,6 +56,18 @@ type Storage interface {
 	Index(ctx context.Context, block VerifiedBlock, certificate Finalization) error
 }
 
+type FullStorage interface {
+	FullBlockRetriever
+	NumBlocks() uint64
+	Index(ctx context.Context, block FullBlock, certificate Finalization) error
+}
+
+type FullBlockRetriever interface {
+	// Retrieve returns the block and finalization at [seq].
+	// If [seq] the block cannot be found, returns ErrBlockNotFound.
+	Retrieve(seq uint64) (FullBlock, Finalization, error)
+}
+
 type Communication interface {
 	// Nodes returns all nodes that participate in the epoch.
 	Nodes() Nodes
@@ -92,6 +104,12 @@ type Block interface {
 	Verify(ctx context.Context) (VerifiedBlock, error)
 }
 
+// Created temporarily to avoid the massive circular
+type SealingBlockInfo struct {
+	Epoch        uint64
+	PChainHeight uint64
+}
+
 type VerifiedBlock interface {
 	// BlockHeader encodes a succinct and collision-free representation of a block.
 	BlockHeader() BlockHeader
@@ -100,6 +118,14 @@ type VerifiedBlock interface {
 
 	// Bytes returns a byte encoding of the block
 	Bytes() ([]byte, error)
+
+	SealingBlockInfo() *SealingBlockInfo
+}
+
+// Contains all functions on the block
+type FullBlock interface {
+	VerifiedBlock
+	Block
 }
 
 // BlockDeserializer deserializes blocks according to formatting
diff --git a/nonvalidator/non_validator.go b/nonvalidator/non_validator.go
new file mode 100644
index 00000000..fc568ac7
--- /dev/null
+++ b/nonvalidator/non_validator.go
@@ -0,0 +1,367 @@
+// Copyright (C) 2019-2025, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package nonvalidator
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/ava-labs/simplex"
+	"go.uber.org/zap"
+)
+
+var (
+	errNoGenesis = errors.New("No Genesis Found")
+)
+
+type epochMetadata struct {
+	nodes               simplex.Nodes
+	nodeLookup          map[string]struct{}
+	epoch               uint64
+	signatureAggregator simplex.SignatureAggregator
+	sealingMetadata     *simplex.SealingBlockInfo
+}
+
+func newEpochMetadata(epoch uint64, nodes []simplex.Node, sigCreator simplex.SignatureAggregatorCreator, sealingMetadata *simplex.SealingBlockInfo) *epochMetadata {
+	lookup := make(map[string]struct{}, len(nodes))
+	for _, node := range nodes {
+		lookup[string(node.Node)] = struct{}{}
+	}
+
+	return &epochMetadata{
+		nodes:               nodes,
+		nodeLookup:          lookup,
+		epoch:               epoch,
+		signatureAggregator: sigCreator(nodes),
+		sealingMetadata:     sealingMetadata,
+	}
+}
+
+// ValidatorSetRetriever returns the validator set for the given epoch.
+// We use epochs not pChainHeight, because NonValidators receive simplex.Blocks
+// which are interfaces that do not expose the pChainHeight.
+type ValidatorSetRetriever func(pChainReference uint64) ([]simplex.Node, error)
+type Config struct {
+	Logger                     simplex.Logger
+	Storage                    simplex.Storage
+	ValidatorSetRetriever      ValidatorSetRetriever
+	SignatureAggregatorCreator simplex.SignatureAggregatorCreator
+	// how many rounds we allow to look past our current
+	MaxRoundWindow uint64
+	// amount of dependencies we are willing to load into the block verifier
+	MaxDependencies uint64
+}
+
+type NonValidator struct {
+	Config
+
+	ctx         context.Context
+	cancelCtx   context.CancelFunc
+	haltedError error
+
+	// incompleteSequences stores sequences that we have not collected
+	// both a block and finalization for. Once both have been received, they are verified & indexed.
+	// TODO: garbage collect old sequences
+	incompleteSequences map[uint64]*finalizedSeq
+
+	// lastAcceptedEpoch contains the metadata of the highest epoch that our NonValidating node has accepted.
+	// An epoch that is accepted means we have verified and indexed the sealing block which created this epoch.
+	// i.e. Seq 100 seals Epoch 50. Epoch 100 is the last accepted epoch.
+	// All blocks before this epoch have been accepted and indexed.
+	// lastAcceptedEpoch *epochMetadata
+
+	// epochs contain a map of all epochs that have their validator set verified.
+	epochs map[uint64]epochMetadata
+
+	verifier *simplex.BlockDependencyManager
+}
+
+// NewNonValidator creates a NonValidator with the given `config`.
+func NewNonValidator(config Config) (*NonValidator, error) {
+	ctx, cancelFunc := context.WithCancel(context.Background())
+
+	// Retrieve the last block so we know where to start bootstrapping from
+	lastBlockHeight := config.Storage.NumBlocks()
+
+	if lastBlockHeight == 0 {
+		return nil, errNoGenesis
+	}
+
+	// TODO: test when the last block retrieved is a sealing block. Currently we would set the lastAcceptedEpoch to sealingblock.epoch, but technically we could set it to sealingblock.seq. Would save a step.
+	lastBlock, _, err := config.Storage.Retrieve(lastBlockHeight - 1)
+	if err != nil {
+		return nil, err
+	}
+	lastBlockEpoch := lastBlock.BlockHeader().Epoch
+	pChainReference, _, err := config.Storage.Retrieve(lastBlockEpoch)
+	if err != nil {
+		return nil, err
+	}
+
+	nodes, err := config.ValidatorSetRetriever(pChainReference.SealingBlockInfo().PChainHeight)
+	if err != nil {
+		return nil, err
+	}
+
+	lastAccepted := newEpochMetadata(lastBlockEpoch, nodes, config.SignatureAggregatorCreator, lastBlock.SealingBlockInfo())
+	epochs := make(map[uint64]epochMetadata)
+	epochs[lastBlockEpoch] = *lastAccepted
+
+	scheduler := simplex.NewScheduler(config.Logger, simplex.DefaultProcessingBlocks)
+
+	return &NonValidator{
+		Config:              config,
+		incompleteSequences: make(map[uint64]*finalizedSeq),
+		ctx:                 ctx,
+		cancelCtx:           cancelFunc,
+		epochs:              epochs,
+		verifier:            simplex.NewBlockVerificationScheduler(config.Logger, config.MaxDependencies, scheduler),
+	}, nil
+}
+
+func (n *NonValidator) Start() {
+	n.broadcastLatestEpoch()
+}
+
+func (n *NonValidator) Stop() {
+	n.cancelCtx()
+}
+
+// TODO: Broadcast the last known epoch to bootstrap the node. Collect responses marking the latest sealing block.
+// Keep rebroadcasting requests for that sealing block until we have enough responses.
+func (n *NonValidator) broadcastLatestEpoch() {
+
+}
+
+// this function should be ran under a lock?
+func (n *NonValidator) HandleMessage(msg *simplex.Message, from simplex.NodeID) error {
+	// A closed context means we have shut down.
+	select {
+	case <-n.ctx.Done():
+		return nil
+	default:
+	}
+
+	if n.haltedError != nil {
+		return n.haltedError
+	}
+
+	switch {
+	case msg.BlockDigestRequest != nil:
+		// TODO: it seems reasonable for our non-validator to be able to process these messages and send out responses.
+		return nil
+	// TODO: create a test for sending a block message but a nil block
+	case msg.BlockMessage != nil && msg.BlockMessage.Block != nil:
+		return n.handleBlock(msg.BlockMessage.Block, from)
+	case msg.Finalization != nil && msg.Finalization != nil:
+		return n.handleFinalization(msg.Finalization, from)
+	default:
+		n.Logger.Debug("Received unexpected message", zap.Any("Message", msg), zap.Stringer("from", from))
+		return nil
+	}
+}
+
+// handleBlock handles a block message. BlockMessages are sent when the leader proposes a block for its round.
+// We only process blocks if they are from the leader and for the current epoch.
+// Otherwise, we wait to process blocks until we receive a finalization.
+func (n *NonValidator) handleBlock(block simplex.Block, from simplex.NodeID) error {
+	bh := block.BlockHeader()
+
+	if bh.Seq > n.MaxRoundWindow+n.Storage.NumBlocks() {
+		n.Logger.Debug("Received a block from a sequence too far ahead", zap.Uint64("Num Blocks", n.Storage.NumBlocks()), zap.Uint64("Block Sequence", bh.Seq), zap.Stringer("From", from))
+		return nil
+	}
+
+	epoch, ok := n.epochs[bh.Epoch]
+	if !ok {
+		n.Logger.Debug("Received a block from an epoch we do not have", zap.Uint64("Epoch", bh.Epoch), zap.Stringer("From", from))
+		return nil
+	}
+
+	if !bytes.Equal(simplex.LeaderForRound(epoch.nodes.NodeIDs(), bh.Round), from) {
+		n.Logger.Debug("Received a block not from the leader of that round", zap.Uint64("Epoch", bh.Epoch), zap.Stringer("From", from))
+		return nil
+	}
+
+	// If we have already verified the block discard it
+	if n.isAccepted(bh.Seq) {
+		n.Logger.Debug("Already accepted a block from this round")
+		return nil
+	}
+
+	incomplete, ok := n.incompleteSequences[bh.Seq]
+	// we have not received any blocks or finalizations for this sequence
+	if !ok {
+		incompleteSeq := &finalizedSeq{
+			block: block,
+		}
+		n.incompleteSequences[bh.Seq] = incompleteSeq
+		n.Logger.Debug("Stored incomplete sequence", zap.Stringer("Sequence", incompleteSeq))
+		return nil
+	}
+
+	// Duplicate block, or finalization not yet received.
+	if incomplete.block != nil || incomplete.finalization == nil {
+		return nil
+	}
+
+	if !bytes.Equal(incomplete.finalization.Finalization.Digest[:], bh.Digest[:]) {
+		n.Logger.Info(
+			"Received a block from the leader of a round whose digest mismatches the finalization",
+			zap.Stringer("Finalization Digest", incomplete.finalization.Finalization.Digest),
+			zap.Stringer("Block digest", bh.Digest),
+			zap.Stringer("From", from),
+		)
+		return nil
+	}
+
+	// add test that ensure this is here. otherwise i think an adversarial node can have us schedule many tasks
+	incomplete.block = block
+
+	finalizedBlockTask := n.createFinalizedBlockVerificationTask(block, incomplete.finalization)
+
+	if bh.Seq == 0 || n.isAccepted(bh.Seq-1) {
+		return n.verifier.ScheduleTaskWithDependencies(finalizedBlockTask, bh.Seq, nil, []uint64{})
+	}
+
+	return n.verifier.ScheduleTaskWithDependencies(finalizedBlockTask, bh.Seq, &(bh.Prev), []uint64{})
+}
+
+func (n *NonValidator) isAccepted(seq uint64) bool {
+	return n.Storage.NumBlocks() > seq
+}
+
+func (n *NonValidator) createFinalizedBlockVerificationTask(block simplex.Block, finalization *simplex.Finalization) func() simplex.Digest {
+	return func() simplex.Digest {
+		md := block.BlockHeader()
+		n.Logger.Debug("Block verification started", zap.Uint64("round", md.Round))
+		start := time.Now()
+		defer func() {
+			elapsed := time.Since(start)
+			n.Logger.Debug("Block verification ended", zap.Uint64("round", md.Round), zap.Duration("elapsed", elapsed))
+		}()
+
+		verifiedBlock, err := block.Verify(n.ctx)
+
+		// is this block a sealing block? set epochs
+		if sealingInfo := verifiedBlock.SealingBlockInfo(); sealingInfo != nil {
+			nodes, err := n.ValidatorSetRetriever(sealingInfo.PChainHeight)
+			if err != nil {
+				n.haltedError = err
+				return md.Digest
+			}
+
+			n.epochs[sealingInfo.Epoch] = *newEpochMetadata(sealingInfo.Epoch, nodes, n.SignatureAggregatorCreator, sealingInfo)
+		}
+
+		// We have failed verifying a finalized block
+		if err != nil {
+			n.Logger.Info("Failed verifying a block that has a finalization", zap.Uint64("Block Seq", md.Seq), zap.Stringer("Block Digest", md.Digest), zap.Error(err))
+			return md.Digest
+		}
+
+		if err := n.Storage.Index(n.ctx, verifiedBlock, *finalization); err != nil {
+			n.haltedError = err
+			n.Logger.Info("Failed indexing a block and finalization", zap.Uint64("Block Seq", md.Seq), zap.Stringer("Block Digest", md.Digest), zap.Error(err))
+			return md.Digest
+		}
+
+		n.removeIncompleteSeqs(md.Seq)
+
+		return md.Digest
+	}
+}
+
+func (n *NonValidator) removeIncompleteSeqs(startSeq uint64) {
+	for seq, _ := range n.incompleteSequences {
+		if seq <= startSeq {
+			delete(n.incompleteSequences, seq)
+		}
+	}
+}
+
+// handleFinalization process a finalization message. If its for a future epoch, it will forward the finalization
+// to the replication handler.
+func (n *NonValidator) handleFinalization(finalization *simplex.Finalization, from simplex.NodeID) error {
+	bh := finalization.Finalization.BlockHeader
+
+	if n.isAccepted(bh.Seq) {
+		n.Logger.Debug("Received a stale finalization", zap.Uint64("Seq", bh.Seq), zap.Stringer("From", from))
+		return nil
+	}
+
+	epoch, ok := n.epochs[bh.Epoch]
+	if !ok {
+		// This finalization is after our lastAcceptedEpoch and is for an unknown Epoch, begin replication
+		n.Logger.Debug("Received a finalization from an unknown epoch", zap.Uint64("Unknown Epoch", bh.Epoch), zap.Stringer("From", from))
+		return nil
+	}
+
+	if err := simplex.VerifyQC(finalization.QC, n.Logger, "Finalization", epoch.signatureAggregator.IsQuorum, epoch.nodeLookup, finalization, from); err != nil {
+		n.Logger.Debug("Received an invalid finalization",
+			zap.Int("round", int(bh.Round)),
+			zap.Stringer("NodeID", from))
+		return nil
+	}
+
+	incomplete, ok := n.incompleteSequences[bh.Seq]
+	if !ok {
+		// we have not received anything for this sequence
+		incompleteSeq := &finalizedSeq{
+			finalization: finalization,
+		}
+		n.incompleteSequences[bh.Seq] = incompleteSeq
+		n.Logger.Debug("Stored incomplete sequence", zap.Stringer("Sequence", incompleteSeq))
+		return nil
+	}
+
+	// Duplicate finalization received.
+	if incomplete.finalization != nil {
+		// sanity check: should never happen.
+		if !bytes.Equal(incomplete.finalization.Finalization.Bytes(), finalization.Finalization.Bytes()) {
+			n.Logger.Warn(
+				"Mismatching finalizations",
+				zap.Uint64("Incoming Sequence", finalization.Finalization.Seq),
+				zap.Uint64("Stored sequence", incomplete.finalization.Finalization.Seq),
+			)
+			n.haltedError = fmt.Errorf("Conflicting finalizations")
+			return fmt.Errorf("Conflicting finalizations")
+		}
+		return nil
+	}
+
+	incomplete.finalization = finalization
+
+	// No block received yet for this sequence.
+	if incomplete.block == nil {
+		// TODO: notify replication
+		return nil
+	}
+
+	digest := incomplete.block.BlockHeader().Digest
+	if !bytes.Equal(bh.Digest[:], digest[:]) {
+		// TODO: this means the leader has equivocated and sent us a wrong block while another has been finalized.
+		// We should probably handle replication for this block?
+		n.Logger.Info(
+			"Received a block from the leader of a round whose digest mismatches the finalization",
+			zap.Stringer("Finalization Digest", bh.Digest),
+			zap.Stringer("Block digest", digest),
+			zap.Stringer("From", from),
+		)
+
+		// TODO: replication here as well
+		return nil
+	}
+
+	finalizedBlockTask := n.createFinalizedBlockVerificationTask(incomplete.block, incomplete.finalization)
+
+	if bh.Seq == 0 || n.isAccepted(bh.Seq-1) {
+		return n.verifier.ScheduleTaskWithDependencies(finalizedBlockTask, bh.Seq, nil, []uint64{})
+	}
+
+	return n.verifier.ScheduleTaskWithDependencies(finalizedBlockTask, bh.Seq, &(bh.Prev), []uint64{})
+}
diff --git a/nonvalidator/non_validator_test.go b/nonvalidator/non_validator_test.go
new file mode 100644
index 00000000..2827531c
--- /dev/null
+++ b/nonvalidator/non_validator_test.go
@@ -0,0 +1,300 @@
+// Copyright (C) 2019-2025, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package nonvalidator
+
+import (
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/ava-labs/simplex"
+	"github.com/ava-labs/simplex/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+type errQC struct{}
+
+func (errQC) Signers() []simplex.NodeID { return nil }
+func (errQC) Verify([]byte) error       { return errors.New("verification failed") }
+func (errQC) Bytes() []byte             { return nil }
+
+func newTestNonValidator(t *testing.T, nodes []simplex.NodeID, lastVerifiedBlock simplex.Block) *NonValidator {
+	config := Config{
+		Storage:           testutil.NewNonValidatorInMemoryStorage(),
+		Logger:            testutil.MakeLogger(t, 1),
+		GenesisValidators: nodes,
+	}
+
+	return NewNonValidator(config, lastVerifiedBlock)
+}
+
+func blockMessage(t *testing.T, block simplex.Block, from simplex.NodeID) *simplex.Message {
+	vote, err := testutil.NewTestVote(block, from)
+	require.NoError(t, err)
+	return &simplex.Message{
+		BlockMessage: &simplex.BlockMessage{
+			Block: block,
+			Vote:  *vote,
+		},
+	}
+}
+
+func TestHandleFinalizationMessage(t *testing.T) {
+	nodes := []simplex.NodeID{{1}, {2}, {3}, {4}}
+
+	tests := []struct {
+		name              string
+		lastVerifiedBlock *testutil.TestBlock
+		// sendBlock controls whether a block message is sent before the finalization.
+		sendBlock         bool
+		blockSender       simplex.NodeID
+		expectVerified    bool
+		expectedNumBlocks uint64
+	}{
+		{
+			name: "Finalization Only No Block",
+		},
+		{
+			name:        "Block From Non-Leader Then Finalization",
+			sendBlock:   true,
+			blockSender: nodes[1],
+		},
+		{
+			name:              "Block From Leader Then Finalization",
+			sendBlock:         true,
+			blockSender:       nodes[0],
+			expectVerified:    true,
+			expectedNumBlocks: 1,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var lastVerified simplex.Block
+			if tt.lastVerifiedBlock != nil {
+				lastVerified = tt.lastVerifiedBlock
+			}
+			v := newTestNonValidator(t, nodes, lastVerified)
+
+			var verified atomic.Bool
+			blockToSend := testutil.NewTestBlock(simplex.ProtocolMetadata{
+				Round: 0,
+				Seq:   0,
+				Epoch: 0,
+			}, simplex.Blacklist{})
+			blockToSend.OnVerify = func() {
+				verified.Store(true)
+			}
+
+			if tt.sendBlock {
+				err := v.HandleMessage(blockMessage(t, blockToSend, nodes[0]), tt.blockSender)
+				require.NoError(t, err)
+			}
+
+			finalization, _ := testutil.NewFinalizationRecord(t, v.Logger, &testutil.TestSignatureAggregator{}, blockToSend, nodes)
+			err := v.HandleMessage(&simplex.Message{Finalization: &finalization}, nodes[0])
+			require.NoError(t, err)
+
+			if tt.expectVerified {
+				require.Eventually(t, verified.Load, 2*time.Second, 20*time.Millisecond)
+			} else {
+				require.Never(t, verified.Load, 2*time.Second, 20*time.Millisecond)
+			}
+			require.Equal(t, tt.expectedNumBlocks, v.Storage.NumBlocks())
+		})
+	}
+}
+
+func TestHandleBlockDigestMismatch(t *testing.T) {
+	nodes := []simplex.NodeID{{1}, {2}, {3}, {4}}
+	v := newTestNonValidator(t, nodes, nil)
+
+	metadata := simplex.ProtocolMetadata{Seq: 0, Epoch: 0, Round: 0}
+	blockA := testutil.NewTestBlock(metadata, simplex.Blacklist{})
+	blockB := testutil.NewTestBlock(metadata, simplex.Blacklist{})
+	blockB.Data = []byte("different")
+	blockB.ComputeDigest()
+
+	// send finalization for blockB
+	finalization, _ := testutil.NewFinalizationRecord(t, v.Logger, &testutil.TestSignatureAggregator{}, blockB, nodes)
+	err := v.HandleMessage(&simplex.Message{Finalization: &finalization}, nodes[0])
+	require.NoError(t, err)
+
+	// send block message for blockA — digest differs from stored finalization
+	err = v.HandleMessage(blockMessage(t, blockA, nodes[0]), nodes[0])
+	require.NoError(t, err)
+
+	require.Equal(t, uint64(0), v.Storage.NumBlocks())
+}
+
+func TestHandleFinalizationDigestMismatch(t *testing.T) {
+	nodes := []simplex.NodeID{{1}, {2}, {3}, {4}}
+	v := newTestNonValidator(t, nodes, nil)
+
+	metadata := simplex.ProtocolMetadata{Seq: 0, Epoch: 0, Round: 0}
+	blockA := testutil.NewTestBlock(metadata, simplex.Blacklist{})
+	blockB := testutil.NewTestBlock(metadata, simplex.Blacklist{})
+	blockB.Data = []byte("different")
+	blockB.ComputeDigest()
+
+	// send block message for blockA from leader
+	err := v.HandleMessage(blockMessage(t, blockA, nodes[0]), nodes[0])
+	require.NoError(t, err)
+
+	// send finalization for blockB — digest differs from stored block
+	finalization, _ := testutil.NewFinalizationRecord(t, v.Logger, &testutil.TestSignatureAggregator{}, blockB, nodes)
+	err = v.HandleMessage(&simplex.Message{Finalization: &finalization}, nodes[0])
+	require.NoError(t, err)
+
+	require.Equal(t, uint64(0), v.Storage.NumBlocks())
+}
+
+func TestHandleFinalizationFailsVerification(t *testing.T) {
+	nodes := []simplex.NodeID{{1}, {2}, {3}, {4}}
+	v := newTestNonValidator(t, nodes, nil)
+
+	var verified atomic.Bool
+	block := testutil.NewTestBlock(simplex.ProtocolMetadata{
+		Round: 0,
+		Seq:   0,
+		Epoch: 0,
+	}, simplex.Blacklist{})
+	block.OnVerify = func() {
+		verified.Store(true)
+	}
+
+	// send block from leader
+	err := v.HandleMessage(blockMessage(t, block, nodes[0]), nodes[0])
+	require.NoError(t, err)
+
+	// send a finalization that fails verification
+	finalization := &simplex.Finalization{
+		QC: errQC{},
+	}
+	err = v.HandleMessage(&simplex.Message{Finalization: finalization}, nodes[0])
+	require.NoError(t, err)
+
+	require.Never(t, verified.Load, 2*time.Second, 20*time.Millisecond)
+	require.Equal(t, uint64(0), v.Storage.NumBlocks())
+}
+
+func TestBlockVerifyCalledOnce(t *testing.T) {
+	nodes := []simplex.NodeID{{1}, {2}, {3}, {4}}
+	v := newTestNonValidator(t, nodes, nil)
+
+	verificationDelay := make(chan struct{})
+	var verifyCount atomic.Int32
+
+	block := testutil.NewTestBlock(simplex.ProtocolMetadata{Seq: 0, Round: 0, Epoch: 0}, simplex.Blacklist{})
+	block.VerificationDelay = verificationDelay
+	block.OnVerify = func() {
+		verifyCount.Add(1)
+		// Schedule a second verification while task 1 is still inside Verify.
+		// The OneTimeVerifier should return the cached result without calling Verify again.
+		_ = v.verifier.triggerVerify(block)
+	}
+
+	finalization, _ := testutil.NewFinalizationRecord(t, v.Logger, &testutil.TestSignatureAggregator{}, block, nodes)
+	err := v.HandleMessage(&simplex.Message{Finalization: &finalization}, nodes[0])
+	require.NoError(t, err)
+
+	err = v.HandleMessage(blockMessage(t, block, nodes[0]), nodes[0])
+	require.NoError(t, err)
+
+	// Unblock the in-progress verification.
+	close(verificationDelay)
+
+	require.Eventually(t, func() bool { return verifyCount.Load() == 1 }, 2*time.Second, 20*time.Millisecond)
+	require.Never(t, func() bool { return verifyCount.Load() > 1 }, 200*time.Millisecond, 20*time.Millisecond)
+}
+
+func TestHandleBlockMessage(t *testing.T) {
+	nodes := []simplex.NodeID{{1}, {2}, {3}, {4}}
+
+	tests := []struct {
+		name              string
+		lastVerifiedBlock *testutil.TestBlock
+		// finalizationBlock returns the block to finalize; nil means no finalization is sent.
+		// lastVerified may be nil when lastVerifiedBlock is not set for the test case.
+		finalizationBlock func(lastVerified, blockToSend *testutil.TestBlock) *testutil.TestBlock
+		blockSender       simplex.NodeID
+		blockSeq          uint64
+		expectVerified    bool
+		expectedNumBlocks uint64
+	}{
+		{
+			name:        "Next to Verify But No Finalization",
+			blockSender: nodes[0],
+		},
+		{
+			name:              "BlockMessage not sent from leader",
+			finalizationBlock: func(_, blockToSend *testutil.TestBlock) *testutil.TestBlock { return blockToSend },
+			blockSender:       nodes[1],
+		},
+		{
+			name: "Already Verified",
+			lastVerifiedBlock: testutil.NewTestBlock(simplex.ProtocolMetadata{
+				Round: 0,
+				Seq:   0,
+				Epoch: 0,
+			}, simplex.Blacklist{}),
+			finalizationBlock: func(lastVerified, _ *testutil.TestBlock) *testutil.TestBlock { return lastVerified },
+			blockSender:       nodes[1],
+		},
+		{
+			name:              "Finalization Received",
+			finalizationBlock: func(_, blockToSend *testutil.TestBlock) *testutil.TestBlock { return blockToSend },
+			blockSender:       nodes[0],
+			expectVerified:    true,
+			expectedNumBlocks: 1,
+		},
+		{
+			// seq 1 arrives with a finalization but seq 0 has not been verified yet,
+			// so the block is indexed but verification is deferred.
+			name:              "Finalization Received But Not Next To Verify",
+			finalizationBlock: func(_, blockToSend *testutil.TestBlock) *testutil.TestBlock { return blockToSend },
+			blockSender:       nodes[0],
+			blockSeq:          1,
+			expectVerified:    false,
+			expectedNumBlocks: 1,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var lastVerified simplex.Block
+			if tt.lastVerifiedBlock != nil {
+				lastVerified = tt.lastVerifiedBlock
+			}
+			v := newTestNonValidator(t, nodes, lastVerified)
+
+			var verified atomic.Bool
+			blockToSend := testutil.NewTestBlock(simplex.ProtocolMetadata{
+				Round: 0,
+				Seq:   tt.blockSeq,
+				Epoch: 0,
+			}, simplex.Blacklist{})
+			blockToSend.OnVerify = func() {
+				verified.Store(true)
+			}
+
+			if tt.finalizationBlock != nil {
+				finalizeBlock := tt.finalizationBlock(tt.lastVerifiedBlock, blockToSend)
+				finalization, _ := testutil.NewFinalizationRecord(t, v.Logger, &testutil.TestSignatureAggregator{}, finalizeBlock, nodes)
+				v.HandleMessage(&simplex.Message{Finalization: &finalization}, nodes[0])
+			}
+
+			err := v.HandleMessage(blockMessage(t, blockToSend, nodes[0]), tt.blockSender)
+			require.NoError(t, err)
+
+			if tt.expectVerified {
+				require.Eventually(t, verified.Load, 2*time.Second, 20*time.Millisecond)
+			} else {
+				require.Never(t, verified.Load, 2*time.Second, 20*time.Millisecond)
+			}
+			require.Equal(t, tt.expectedNumBlocks, v.Storage.NumBlocks())
+		})
+	}
+}
diff --git a/nonvalidator/util.go b/nonvalidator/util.go
new file mode 100644
index 00000000..7027977e
--- /dev/null
+++ b/nonvalidator/util.go
@@ -0,0 +1,41 @@
+// Copyright (C) 2019-2025, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package nonvalidator
+
+import (
+	"fmt"
+
+	"github.com/ava-labs/simplex"
+)
+
+type finalizedSeq struct {
+	block        simplex.Block
+	finalization *simplex.Finalization
+}
+
+func (f *finalizedSeq) String() string {
+	seq := uint64(0)
+	digest := simplex.Digest{}
+	if f.block != nil {
+		seq = f.block.BlockHeader().Seq
+		digest = f.block.BlockHeader().Digest
+	}
+	if f.finalization != nil {
+		seq = f.finalization.Finalization.Seq
+		digest = f.finalization.Finalization.Digest
+	}
+
+	return fmt.Sprintf("FinalizedSeq {BlockDigest: %s, Seq: %d, BlockExists %t, FinalizationExists %t}", digest, seq, f.block != nil, f.finalization != nil)
+}
+
+func (f *finalizedSeq) GetSeq() uint64 {
+	if f.block != nil {
+		return f.block.BlockHeader().Seq
+	}
+	if f.finalization != nil {
+		return f.finalization.Finalization.Seq
+	}
+
+	return 0
+}
diff --git a/testutil/block.go b/testutil/block.go
index 176e39ee..bc832e9f 100644
--- a/testutil/block.go
+++ b/testutil/block.go
@@ -58,6 +58,10 @@ func (tb *TestBlock) Verify(context.Context) (simplex.VerifiedBlock, error) {
 	return tb, nil
 }
 
+func (tb *TestBlock) SealingBlockInfo() *simplex.SealingBlockInfo {
+	return nil
+}
+
 func (tb *TestBlock) ComputeDigest() {
 	var bb bytes.Buffer
 	tbBytes, err := tb.Bytes()
diff --git a/testutil/random_network/block.go b/testutil/random_network/block.go
index 59e3f3f9..535b8ff0 100644
--- a/testutil/random_network/block.go
+++ b/testutil/random_network/block.go
@@ -55,6 +55,10 @@ func (b *Block) BlockHeader() simplex.BlockHeader {
 	}
 }
 
+func (b *Block) SealingBlockInfo() *simplex.SealingBlockInfo {
+	return nil
+}
+
 type encodedBlock struct {
 	ProtocolMetadata []byte
 	TXs              []asn1TX