Repository: ObolNetwork/charon

Issue:
Attestation source and target vote consensus
# 🎯 Problem to be solved
In the Holesky Pectra fork. Two clients (who made up a supermajority) chose the wrong fork. All nodes using these BNs voted incorrectly. This crossed the 66% threshold, resulting in them justifying the wrong block and ultimately getting slashed (or being offline forever more). This is the worst case scenario for eth staking. Unfortunately, charon was not able to save any of its DVs either, because we don’t have a halting feature and we were unanimously on the bad versions too, so halting wouldn’t have worked. 

# 🛠️ Proposed solution
We have 2 options on Charon's side:
1. No charon node signs an attestation if one charon node has different source and target roots.
2. A charon node does not sign an attestation if it has different source and target roots.

While the second option will be ideally what will help us prevent the case from above, it will be:
- harder to implement, as it will require overriding the cluster's thresholds;
- an easy way of DoS-ing the whole cluster, which is a big no.

Hence, for first iteration, we are considering only the first option. While it may not prevent such extreme scenarios as the one experience during Pectra, it might be really beneficial for beacon nodes on wrong minority forks.

In the future, a separate feature for "tighter attestation control" might be considered that will override option 1 to option 2 (potentially enabled only for specific time-frame, i.e.: during ethereum hardforks).

# 🧪 Tests
- [ ] Tested by new automated unit/integration/smoke tests - force a beaconmock to supply different attestation source and target roots
- [ ] Manually tested on core team/canary/test clusters
- [ ] Manually tested on local compose simnet


Hints from issue comments:

Solution patch:
diff --git a/app/app.go b/app/app.go
index ac79a686..2647f828 100644
--- a/app/app.go
+++ b/app/app.go
@@ -607,7 +607,7 @@ func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
 	// Consensus
 	consensusController, err := consensus.NewConsensusController(
 		ctx, p2pNode, sender, peers, p2pKey,
-		deadlineFunc, gaterFunc, consensusDebugger)
+		deadlineFunc, gaterFunc, consensusDebugger, featureset.Enabled(featureset.ChainSplitHalt))
 	if err != nil {
 		return err
 	}
diff --git a/app/featureset/featureset.go b/app/featureset/featureset.go
index c1244a8a..b76ec74c 100644
--- a/app/featureset/featureset.go
+++ b/app/featureset/featureset.go
@@ -66,6 +66,10 @@ const (
 
 	// FetchOnlyCommIdx0 enables querying the beacon node for attestation data only for committee index 0.
 	FetchOnlyCommIdx0 = "fetch_only_commidx_0"
+
+	// ChainSplitHalt compares locally fetched attestation's target and source to leader's proposed target and source attestation.
+	// In case they differ, Charon does not sign the attestation.
+	ChainSplitHalt = "chain_split_halt"
 )
 
 var (
@@ -83,7 +87,8 @@ var (
 		ProposalTimeout:      statusAlpha,
 		QUIC:                 statusAlpha,
 		FetchOnlyCommIdx0:    statusAlpha,
-		// Add all features and there status here.
+		ChainSplitHalt:       statusAlpha,
+		// Add all features and their status here.
 	}
 
 	// minStatus defines the minimum enabled status.
diff --git a/core/consensus/controller.go b/core/consensus/controller.go
index a6fa3e8b..a79a061f 100644
--- a/core/consensus/controller.go
+++ b/core/consensus/controller.go
@@ -39,11 +39,11 @@ type consensusController struct {
 // NewConsensusController creates a new consensus controller with the default consensus protocol.
 func NewConsensusController(ctx context.Context, p2pNode host.Host, sender *p2p.Sender,
 	peers []p2p.Peer, p2pKey *k1.PrivateKey, deadlineFunc core.DeadlineFunc,
-	gaterFunc core.DutyGaterFunc, debugger Debugger,
+	gaterFunc core.DutyGaterFunc, debugger Debugger, compareAttestations bool,
 ) (core.ConsensusController, error) {
 	qbftDeadliner := core.NewDeadliner(ctx, "consensus.qbft", deadlineFunc)
 
-	defaultConsensus, err := qbft.NewConsensus(p2pNode, sender, peers, p2pKey, qbftDeadliner, gaterFunc, debugger.AddInstance)
+	defaultConsensus, err := qbft.NewConsensus(p2pNode, sender, peers, p2pKey, qbftDeadliner, gaterFunc, debugger.AddInstance, compareAttestations)
 	if err != nil {
 		return nil, err
 	}
diff --git a/core/consensus/instance/instance_io.go b/core/consensus/instance/instance_io.go
index 84dfaf43..86170abf 100644
--- a/core/consensus/instance/instance_io.go
+++ b/core/consensus/instance/instance_io.go
@@ -21,6 +21,7 @@ func NewIO[T any]() *IO[T] {
 		RecvBuffer:  make(chan T, RecvBufferSize),
 		HashCh:      make(chan [32]byte, 1),
 		ValueCh:     make(chan proto.Message, 1),
+		VerifyCh:    make(chan proto.Message, 1),
 		ErrCh:       make(chan error, 1),
 		DecidedAtCh: make(chan time.Time, 1),
 	}
@@ -35,6 +36,7 @@ type IO[T any] struct {
 	RecvBuffer   chan T             // Outer receive buffers.
 	HashCh       chan [32]byte      // Async input hash channel.
 	ValueCh      chan proto.Message // Async input value channel.
+	VerifyCh     chan proto.Message // Async input value channel used for comparing local value to leader value during consensus.
 	ErrCh        chan error         // Async output error channel.
 	DecidedAtCh  chan time.Time     // Async output decided timestamp channel.
 }
diff --git a/core/consensus/qbft/msg.go b/core/consensus/qbft/msg.go
index 0a887577..12d17614 100644
--- a/core/consensus/qbft/msg.go
+++ b/core/consensus/qbft/msg.go
@@ -41,7 +41,7 @@ func newMsg(pbMsg *pbv1.QBFTMsg, justification []*pbv1.QBFTMsg, values map[[32]b
 		}
 	}
 
-	var justImpls []qbft.Msg[core.Duty, [32]byte]
+	var justImpls []qbft.Msg[core.Duty, [32]byte, proto.Message]
 
 	for _, j := range justification {
 		impl, err := newMsg(j, nil, values)
@@ -62,7 +62,7 @@ func newMsg(pbMsg *pbv1.QBFTMsg, justification []*pbv1.QBFTMsg, values map[[32]b
 	}, nil
 }
 
-// Msg wraps *pbv1.QBFTMsg and justifications and implements qbft.Msg[core.Duty, [32]byte].
+// Msg wraps *pbv1.QBFTMsg and justifications and implements qbft.Msg[core.Duty, [32]byte, proto.Message].
 type Msg struct {
 	msg               *pbv1.QBFTMsg
 	valueHash         [32]byte
@@ -70,7 +70,7 @@ type Msg struct {
 	values            map[[32]byte]*anypb.Any
 
 	justificationProtos []*pbv1.QBFTMsg
-	justification       []qbft.Msg[core.Duty, [32]byte]
+	justification       []qbft.Msg[core.Duty, [32]byte, proto.Message]
 }
 
 func (m Msg) Type() qbft.MsgType {
@@ -93,6 +93,15 @@ func (m Msg) Value() [32]byte {
 	return m.valueHash
 }
 
+func (m Msg) ValueSource() (proto.Message, error) {
+	v, ok := m.values[m.valueHash]
+	if !ok {
+		return nil, errors.New("value not found")
+	}
+
+	return v, nil
+}
+
 func (m Msg) Values() map[[32]byte]*anypb.Any {
 	return m.values
 }
@@ -109,7 +118,7 @@ func (m Msg) PreparedValue() [32]byte {
 	return m.preparedValueHash
 }
 
-func (m Msg) Justification() []qbft.Msg[core.Duty, [32]byte] {
+func (m Msg) Justification() []qbft.Msg[core.Duty, [32]byte, proto.Message] {
 	return m.justification
 }
 
@@ -218,4 +227,4 @@ func toHash32(val []byte) ([32]byte, bool) {
 	return resp, true
 }
 
-var _ qbft.Msg[core.Duty, [32]byte] = Msg{} // Interface assertion
+var _ qbft.Msg[core.Duty, [32]byte, proto.Message] = Msg{} // Interface assertion
diff --git a/core/consensus/qbft/qbft.go b/core/consensus/qbft/qbft.go
index dd07915f..c4b2f978 100644
--- a/core/consensus/qbft/qbft.go
+++ b/core/consensus/qbft/qbft.go
@@ -5,6 +5,7 @@ package qbft
 import (
 	"context"
 	"fmt"
+	"slices"
 	"strings"
 	"sync"
 	"time"
@@ -36,20 +37,22 @@ import (
 
 type subscriber func(ctx context.Context, duty core.Duty, value proto.Message) error
 
+var supportedCompareDuties = []core.DutyType{core.DutyAttester}
+
 // newDefinition returns a qbft definition (this is constant across all consensus instances).
 func newDefinition(nodes int, subs func() []subscriber, roundTimer timer.RoundTimer,
-	decideCallback func(qcommit []qbft.Msg[core.Duty, [32]byte]),
-) qbft.Definition[core.Duty, [32]byte] {
-	quorum := qbft.Definition[int, int]{Nodes: nodes}.Quorum()
+	decideCallback func(qcommit []qbft.Msg[core.Duty, [32]byte, proto.Message]), compareAttestations bool,
+) qbft.Definition[core.Duty, [32]byte, proto.Message] {
+	quorum := qbft.Definition[core.Duty, [32]byte, proto.Message]{Nodes: nodes}.Quorum()
 
-	return qbft.Definition[core.Duty, [32]byte]{
+	return qbft.Definition[core.Duty, [32]byte, proto.Message]{
 		// IsLeader is a deterministic leader election function.
 		IsLeader: func(duty core.Duty, round, process int64) bool {
 			return leader(duty, round, nodes) == process
 		},
 
 		// Decide sends consensus output to subscribers.
-		Decide: func(ctx context.Context, duty core.Duty, _ [32]byte, qcommit []qbft.Msg[core.Duty, [32]byte]) {
+		Decide: func(ctx context.Context, duty core.Duty, _ [32]byte, qcommit []qbft.Msg[core.Duty, [32]byte, proto.Message]) {
 			msg, ok := qcommit[0].(Msg)
 			if !ok {
 				log.Error(ctx, "Invalid message type", nil)
@@ -77,18 +80,84 @@ func newDefinition(nodes int, subs func() []subscriber, roundTimer timer.RoundTi
 			}
 		},
 
+		Compare: func(ctx context.Context, msg qbft.Msg[core.Duty, [32]byte, proto.Message], inputValueSourceCh <-chan proto.Message, inputValueSource proto.Message, returnErrCh chan error, returnProtoCh chan proto.Message) {
+			if !compareAttestations {
+				returnErrCh <- nil
+				return
+			}
+
+			if !slices.Contains(supportedCompareDuties, msg.Instance().Type) {
+				returnErrCh <- nil
+				return
+			}
+
+			attLeaderAnyPbProto, err := msg.ValueSource()
+			if err != nil {
+				returnErrCh <- errors.Wrap(err, "msg has no value source", z.Any("msg", msg))
+				return
+			}
+
+			attLeaderAnyPb, ok := attLeaderAnyPbProto.(*anypb.Any)
+			if !ok {
+				returnErrCh <- errors.New("protoMessage interface to *anypb.Any struct", z.Any("attLeaderAnyPbProto", attLeaderAnyPbProto))
+				return
+			}
+
+			attLeaderSetProto, err := attLeaderAnyPb.UnmarshalNew()
+			if err != nil {
+				returnErrCh <- errors.Wrap(err, "unmarshal *anypb.Any", z.Any("attLeaderAnyPb", attLeaderAnyPb))
+				return
+			}
+
+			attLeaderSet, ok := attLeaderSetProto.(*pbv1.UnsignedDataSet)
+			if !ok {
+				returnErrCh <- errors.New("protoMessage interface to *pbv1.UnsignedDataSet struct", z.Any("attLeaderSetProto", attLeaderSetProto))
+				return
+			}
+
+			switch msg.Instance().Type {
+			case core.DutyAttester:
+				if inputValueSource == nil {
+					select {
+					case <-ctx.Done():
+						returnErrCh <- errors.New("timeout on waiting for local value")
+						return
+					case inputValueSource = <-inputValueSourceCh:
+						returnProtoCh <- inputValueSource
+					}
+				}
+
+				attLocalSet, ok := inputValueSource.(*pbv1.UnsignedDataSet)
+				if !ok {
+					returnErrCh <- errors.New("inputValueSource to pbv1.UnsignedDataSet")
+					return
+				}
+
+				err = attestationChecker(ctx, attLeaderSet, attLocalSet)
+				if err != nil {
+					returnErrCh <- errors.Wrap(err, "attestation checker failed")
+					return
+				}
+			default:
+				returnErrCh <- errors.New("bug: checking not supported duty", z.Any("duty", msg.Instance().Type))
+				return
+			}
+
+			returnErrCh <- nil
+		},
+
 		NewTimer: roundTimer.Timer,
 
 		// LogUponRule logs upon rules at debug level.
 		LogUponRule: func(ctx context.Context, _ core.Duty, _, round int64,
-			_ qbft.Msg[core.Duty, [32]byte], uponRule qbft.UponRule,
+			_ qbft.Msg[core.Duty, [32]byte, proto.Message], uponRule qbft.UponRule,
 		) {
 			log.Debug(ctx, "QBFT upon rule triggered", z.Any("rule", uponRule), z.I64("round", round))
 		},
 
 		// LogRoundChange logs round changes at debug level.
 		LogRoundChange: func(ctx context.Context, duty core.Duty, process, round, newRound int64, //nolint:revive // keep process variable name for clarity
-			uponRule qbft.UponRule, msgs []qbft.Msg[core.Duty, [32]byte],
+			uponRule qbft.UponRule, msgs []qbft.Msg[core.Duty, [32]byte, proto.Message],
 		) {
 			fields := []z.Field{
 				z.Any("rule", uponRule),
@@ -108,7 +177,7 @@ func newDefinition(nodes int, subs func() []subscriber, roundTimer timer.RoundTi
 			log.Debug(ctx, "QBFT round changed", fields...)
 		},
 
-		LogUnjust: func(ctx context.Context, _ core.Duty, _ int64, msg qbft.Msg[core.Duty, [32]byte]) {
+		LogUnjust: func(ctx context.Context, _ core.Duty, _ int64, msg qbft.Msg[core.Duty, [32]byte, proto.Message]) {
 			log.Warn(ctx, "Unjustified consensus message from peer", nil,
 				z.Any("type", msg.Type()),
 				z.I64("peer", msg.Source()),
@@ -123,9 +192,62 @@ func newDefinition(nodes int, subs func() []subscriber, roundTimer timer.RoundTi
 	}
 }
 
+func attestationChecker(ctx context.Context, attLeaderSet *pbv1.UnsignedDataSet, attLocalSet *pbv1.UnsignedDataSet) error {
+	attLocalSetCore, err := core.UnsignedDataSetFromProto(core.DutyAttester, attLocalSet)
+	if err != nil {
+		return errors.Wrap(err, "attLocal to unsigned data set duty attester")
+	}
+
+	attLeaderSetCore, err := core.UnsignedDataSetFromProto(core.DutyAttester, attLeaderSet)
+	if err != nil {
+		return errors.Wrap(err, "attLeader to unsigned data set duty attester")
+	}
+
+	for attLeaderKey, attLeaderData := range attLeaderSetCore {
+		attLocalData, ok := attLocalSetCore[attLeaderKey]
+		if !ok {
+			log.Warn(ctx, "", errors.New("no local attestation found, skipping"), z.Any("pk", attLeaderKey))
+			continue
+		}
+
+		attLeaderAttestationData, ok := attLeaderData.(core.AttestationData)
+		if !ok {
+			return errors.New("unable to parse leader unsigned data to core attestation data", z.Any("data", attLeaderData))
+		}
+
+		attLocalAttestationData, ok := attLocalData.(core.AttestationData)
+		if !ok {
+			return errors.New("unable to parse local unsigned data to core attestation data", z.Any("data", attLocalData))
+		}
+
+		mismatch := ""
+		if attLeaderAttestationData.Data.Source.Epoch != attLocalAttestationData.Data.Source.Epoch {
+			mismatch += "leader attestation source epoch differs from local source epoch;"
+		}
+
+		if attLeaderAttestationData.Data.Source.Root != attLocalAttestationData.Data.Source.Root {
+			mismatch += "leader attestation source root differs from local source root;"
+		}
+
+		if attLeaderAttestationData.Data.Target.Epoch != attLocalAttestationData.Data.Target.Epoch {
+			mismatch += "leader attestation target epoch differs from local target epoch;"
+		}
+
+		if attLeaderAttestationData.Data.Target.Root != attLocalAttestationData.Data.Target.Root {
+			mismatch += "leader attestation target root differs from local target root;"
+		}
+
+		if mismatch != "" {
+			return errors.New(mismatch, z.Any("public_key", attLeaderKey), z.Any("leader", attLeaderAttestationData.Data), z.Any("local", attLocalAttestationData.Data))
+		}
+	}
+
+	return nil
+}
+
 // NewConsensus returns a new consensus QBFT component.
 func NewConsensus(p2pNode host.Host, sender *p2p.Sender, peers []p2p.Peer, p2pKey *k1.PrivateKey,
-	deadliner core.Deadliner, gaterFunc core.DutyGaterFunc, snifferFunc func(*pbv1.SniffedConsensusInstance),
+	deadliner core.Deadliner, gaterFunc core.DutyGaterFunc, snifferFunc func(*pbv1.SniffedConsensusInstance), compareAttestations bool,
 ) (*Consensus, error) {
 	// Extract peer pubkeys.
 	keys := make(map[int64]*k1.PublicKey)
@@ -143,18 +265,19 @@ func NewConsensus(p2pNode host.Host, sender *p2p.Sender, peers []p2p.Peer, p2pKe
 	}
 
 	c := &Consensus{
-		p2pNode:     p2pNode,
-		sender:      sender,
-		peers:       peers,
-		peerLabels:  labels,
-		privkey:     p2pKey,
-		pubkeys:     keys,
-		deadliner:   deadliner,
-		snifferFunc: snifferFunc,
-		gaterFunc:   gaterFunc,
-		dropFilter:  log.Filter(),
-		timerFunc:   timer.GetRoundTimerFunc(),
-		metrics:     metrics.NewConsensusMetrics(protocols.QBFTv2ProtocolID),
+		p2pNode:             p2pNode,
+		sender:              sender,
+		peers:               peers,
+		peerLabels:          labels,
+		privkey:             p2pKey,
+		pubkeys:             keys,
+		deadliner:           deadliner,
+		snifferFunc:         snifferFunc,
+		gaterFunc:           gaterFunc,
+		dropFilter:          log.Filter(),
+		timerFunc:           timer.GetRoundTimerFunc(),
+		metrics:             metrics.NewConsensusMetrics(protocols.QBFTv2ProtocolID),
+		compareAttestations: compareAttestations,
 	}
 	c.mutable.instances = make(map[core.Duty]*instance.IO[Msg])
 
@@ -164,19 +287,20 @@ func NewConsensus(p2pNode host.Host, sender *p2p.Sender, peers []p2p.Peer, p2pKe
 // Consensus implements core.Consensus & priority.coreConsensus.
 type Consensus struct {
 	// Immutable state
-	p2pNode     host.Host
-	sender      *p2p.Sender
-	peerLabels  []string
-	peers       []p2p.Peer
-	pubkeys     map[int64]*k1.PublicKey
-	privkey     *k1.PrivateKey
-	subs        []subscriber
-	deadliner   core.Deadliner
-	snifferFunc func(*pbv1.SniffedConsensusInstance)
-	gaterFunc   core.DutyGaterFunc
-	dropFilter  z.Field // Filter buffer overflow errors (possible DDoS)
-	timerFunc   timer.RoundTimerFunc
-	metrics     metrics.ConsensusMetrics
+	p2pNode             host.Host
+	sender              *p2p.Sender
+	peerLabels          []string
+	peers               []p2p.Peer
+	pubkeys             map[int64]*k1.PublicKey
+	privkey             *k1.PrivateKey
+	subs                []subscriber
+	deadliner           core.Deadliner
+	snifferFunc         func(*pbv1.SniffedConsensusInstance)
+	gaterFunc           core.DutyGaterFunc
+	dropFilter          z.Field // Filter buffer overflow errors (possible DDoS)
+	timerFunc           timer.RoundTimerFunc
+	metrics             metrics.ConsensusMetrics
+	compareAttestations bool
 
 	// Mutable state
 	mutable struct {
@@ -297,6 +421,14 @@ func (c *Consensus) propose(ctx context.Context, duty core.Duty, value proto.Mes
 		return errors.New("input channel full")
 	}
 
+	if c.compareAttestations {
+		select {
+		case inst.VerifyCh <- value:
+		default:
+			return errors.New("input channel full")
+		}
+	}
+
 	// Instrument consensus duration using decidedAt output.
 	proposedAt := time.Now()
 
@@ -412,7 +544,7 @@ func (c *Consensus) runInstance(parent context.Context, duty core.Duty) (err err
 		span.End()
 	}()
 
-	decideCallback := func(qcommit []qbft.Msg[core.Duty, [32]byte]) {
+	decideCallback := func(qcommit []qbft.Msg[core.Duty, [32]byte, proto.Message]) {
 		round := qcommit[0].Round()
 		decided = true
 
@@ -440,10 +572,10 @@ func (c *Consensus) runInstance(parent context.Context, duty core.Duty) (err err
 	}
 
 	// Create a new qbft definition for this instance.
-	def := newDefinition(len(c.peers), c.subscribers, roundTimer, decideCallback)
+	def := newDefinition(len(c.peers), c.subscribers, roundTimer, decideCallback, c.compareAttestations)
 
 	// Create a new transport that handles sending and receiving for this instance.
-	t := newTransport(c, c.privkey, inst.ValueCh, make(chan qbft.Msg[core.Duty, [32]byte]), newSniffer(int64(def.Nodes), peerIdx))
+	t := newTransport(c, c.privkey, inst.ValueCh, make(chan qbft.Msg[core.Duty, [32]byte, proto.Message]), newSniffer(int64(def.Nodes), peerIdx))
 
 	// Provide sniffed buffer to snifferFunc at the end.
 	defer func() {
@@ -454,13 +586,13 @@ func (c *Consensus) runInstance(parent context.Context, duty core.Duty) (err err
 	go t.ProcessReceives(ctx, c.getRecvBuffer(duty))
 
 	// Create a qbft transport from the transport
-	qt := qbft.Transport[core.Duty, [32]byte]{
+	qt := qbft.Transport[core.Duty, [32]byte, proto.Message]{
 		Broadcast: t.Broadcast,
 		Receive:   t.RecvBuffer(),
 	}
 
 	// Run the algo, blocking until the context is cancelled.
-	err = qbft.Run(ctx, def, qt, duty, peerIdx, inst.HashCh)
+	err = qbft.Run(ctx, def, qt, duty, peerIdx, inst.HashCh, inst.VerifyCh)
 	if err != nil && !isContextErr(err) {
 		span.AddEvent("qbft.Error")
 		c.metrics.IncConsensusError()
@@ -644,7 +776,7 @@ type roundStep struct {
 }
 
 // groupRoundMessages groups messages by type and returns which peers were present and missing for each type.
-func groupRoundMessages(msgs []qbft.Msg[core.Duty, [32]byte], peers int, round int64, leader int) []roundStep {
+func groupRoundMessages(msgs []qbft.Msg[core.Duty, [32]byte, proto.Message], peers int, round int64, leader int) []roundStep {
 	// checkPeers returns two slices of peer indexes, one with peers
 	// present with the message type and one with messing peers.
 	checkPeers := func(typ qbft.MsgType) (present []int, missing []int) {
diff --git a/core/consensus/qbft/transport.go b/core/consensus/qbft/transport.go
index a85fd3ca..ee47981d 100644
--- a/core/consensus/qbft/transport.go
+++ b/core/consensus/qbft/transport.go
@@ -27,7 +27,7 @@ type transport struct {
 	// Immutable state
 	broadcaster broadcaster
 	privkey     *k1.PrivateKey
-	recvBuffer  chan qbft.Msg[core.Duty, [32]byte] // Instance inner receive buffer.
+	recvBuffer  chan qbft.Msg[core.Duty, [32]byte, proto.Message] // Instance inner receive buffer.
 	sniffer     *sniffer
 
 	// Mutable state
@@ -38,7 +38,7 @@ type transport struct {
 
 // newTransport creates a new qbftTransport.
 func newTransport(broadcaster broadcaster, privkey *k1.PrivateKey, valueCh <-chan proto.Message,
-	recvBuffer chan qbft.Msg[core.Duty, [32]byte], sniffer *sniffer,
+	recvBuffer chan qbft.Msg[core.Duty, [32]byte, proto.Message], sniffer *sniffer,
 ) *transport {
 	return &transport{
 		broadcaster: broadcaster,
@@ -92,7 +92,7 @@ func (t *transport) getValue(hash [32]byte) (*anypb.Any, error) {
 // Broadcast creates a msg and sends it to all peers (including self).
 func (t *transport) Broadcast(ctx context.Context, typ qbft.MsgType, duty core.Duty,
 	peerIdx int64, round int64, valueHash [32]byte, pr int64, pvHash [32]byte,
-	justification []qbft.Msg[core.Duty, [32]byte],
+	justification []qbft.Msg[core.Duty, [32]byte, proto.Message],
 ) error {
 	// Get all hashes
 	var hashes [][32]byte
@@ -169,7 +169,7 @@ func (t *transport) SnifferInstance() *pbv1.SniffedConsensusInstance {
 }
 
 // RecvBuffer returns the inner receive buffer.
-func (t *transport) RecvBuffer() chan qbft.Msg[core.Duty, [32]byte] {
+func (t *transport) RecvBuffer() chan qbft.Msg[core.Duty, [32]byte, proto.Message] {
 	return t.recvBuffer
 }
 
@@ -177,7 +177,7 @@ func (t *transport) RecvBuffer() chan qbft.Msg[core.Duty, [32]byte] {
 // and wrapping that in a msg type.
 func createMsg(typ qbft.MsgType, duty core.Duty,
 	peerIdx int64, round int64, vHash [32]byte, pr int64, pvHash [32]byte,
-	values map[[32]byte]*anypb.Any, justification []qbft.Msg[core.Duty, [32]byte],
+	values map[[32]byte]*anypb.Any, justification []qbft.Msg[core.Duty, [32]byte, proto.Message],
 	privkey *k1.PrivateKey,
 ) (Msg, error) {
 	pbMsg := &pbv1.QBFTMsg{
diff --git a/core/qbft/qbft.go b/core/qbft/qbft.go
index 188041cc..ddf28c00 100644
--- a/core/qbft/qbft.go
+++ b/core/qbft/qbft.go
@@ -12,37 +12,41 @@ import (
 	"time"
 
 	"github.com/obolnetwork/charon/app/errors"
+	"github.com/obolnetwork/charon/app/log"
 )
 
 // Transport abstracts the transport layer between processes in the consensus system.
-type Transport[I any, V comparable] struct {
+type Transport[I any, V comparable, C any] struct {
 	// Broadcast sends a message with the provided fields to all other
 	// processes in the system (including this process).
 	//
 	// Note that a non-nil error exits the algorithm.
-	Broadcast func(ctx context.Context, typ MsgType, instance I, source int64, round int64, value V, pr int64, pv V, justification []Msg[I, V]) error
+	Broadcast func(ctx context.Context, typ MsgType, instance I, source int64, round int64, value V, pr int64, pv V, justification []Msg[I, V, C]) error
 
 	// Receive returns a stream of messages received
 	// from other processes in the system (including this process).
-	Receive <-chan Msg[I, V]
+	Receive <-chan Msg[I, V, C]
 }
 
 // Definition defines the consensus system parameters that are external to the qbft algorithm.
 // This remains constant across multiple instances of consensus (calls to Run).
-type Definition[I any, V comparable] struct {
+type Definition[I any, V comparable, C any] struct {
 	// IsLeader is a deterministic leader election function.
 	IsLeader func(instance I, round, process int64) bool
 	// NewTimer returns a new timer channel and stop function for the round.
 	NewTimer func(round int64) (<-chan time.Time, func())
+	// Compare is called when leader proposes value and we compare it with our local value.
+	// Compare is an opt-in feature that should instantly return nil on returnErr channel if it is not turned on.
+	Compare func(ctx context.Context, qcommit Msg[I, V, C], inputValueSourceCh <-chan C, inputValueSource C, returnErr chan error, returnValue chan C)
 	// Decide is called when consensus has been reached on a value.
-	Decide func(ctx context.Context, instance I, value V, qcommit []Msg[I, V])
+	Decide func(ctx context.Context, instance I, value V, qcommit []Msg[I, V, C])
 	// LogUponRule allows debug logging of triggered upon rules on message receipt.
-	LogUponRule func(ctx context.Context, instance I, process, round int64, msg Msg[I, V], uponRule UponRule)
+	LogUponRule func(ctx context.Context, instance I, process, round int64, msg Msg[I, V, C], uponRule UponRule)
 	// LogRoundChange allows debug logging of round changes.
 	// It includes the rule that triggered it and all received round messages.
-	LogRoundChange func(ctx context.Context, instance I, process, round, newRound int64, uponRule UponRule, msgs []Msg[I, V])
+	LogRoundChange func(ctx context.Context, instance I, process, round, newRound int64, uponRule UponRule, msgs []Msg[I, V, C])
 	// LogUnjust allows debug logging of unjust messages.
-	LogUnjust func(ctx context.Context, instance I, process int64, msg Msg[I, V])
+	LogUnjust func(ctx context.Context, instance I, process int64, msg Msg[I, V, C])
 
 	// Nodes is the total number of nodes/processes participating in consensus.
 	Nodes int
@@ -52,13 +56,13 @@ type Definition[I any, V comparable] struct {
 
 // Quorum returns the quorum count for the system.
 // See IBFT 2.0 paper for correct formula: https://arxiv.org/pdf/1909.10194.pdf
-func (d Definition[I, V]) Quorum() int {
+func (d Definition[I, V, C]) Quorum() int {
 	return int(math.Ceil(float64(d.Nodes*2) / 3))
 }
 
 // Faulty returns the maximum number of faulty/byzantium nodes supported in the system.
 // See IBFT 2.0 paper for correct formula: https://arxiv.org/pdf/1909.10194.pdf
-func (d Definition[I, V]) Faulty() int {
+func (d Definition[I, V, C]) Faulty() int {
 	return int(math.Floor(float64(d.Nodes-1) / 3))
 }
 
@@ -96,7 +100,7 @@ var typeLabels = map[MsgType]string{
 }
 
 // Msg defines the inter process messages.
-type Msg[I any, V comparable] interface {
+type Msg[I any, V comparable, C any] interface {
 	// Type of the message.
 	Type() MsgType
 	// Instance identifies the consensus instance.
@@ -105,14 +109,16 @@ type Msg[I any, V comparable] interface {
 	Source() int64
 	// Round the message pertains to.
 	Round() int64
-	// Value being proposed.
+	// Value being proposed, usually a hash.
 	Value() V
+	// ValueSource being proposed, usually the value that was hashed and is returned in Value().
+	ValueSource() (C, error)
 	// PreparedRound is the justified prepared round.
 	PreparedRound() int64
 	// PreparedValue is the justified prepared value.
 	PreparedValue() V
 	// Justification is the set of messages that explicitly justifies this message.
-	Justification() []Msg[I, V]
+	Justification() []Msg[I, V, C]
 }
 
 // UponRule defines the event based rules that are triggered when messages are received.
@@ -152,6 +158,12 @@ type dedupKey struct {
 	Round    int64
 }
 
+// errors
+var (
+	errCompare = errors.New("compare leader value with local value failed")
+	errTimeout = errors.New("timeout")
+)
+
 // InputValue is a convenience function to create a populated input value channel.
 func InputValue[V comparable](inputValue V) <-chan V {
 	ch := make(chan V, 1)
@@ -160,10 +172,19 @@ func InputValue[V comparable](inputValue V) <-chan V {
 	return ch
 }
 
+// InputValueSource is a convenience function to create a populated input value source channel.
+func InputValueSource[C any](inputValueSource C) <-chan C {
+	ch := make(chan C, 1)
+	ch <- inputValueSource
+
+	return ch
+}
+
 // Run executes the consensus algorithm until the context closed.
 // The generic type I is the instance of consensus and can be anything.
 // The generic type V is the arbitrary data value being proposed; it only requires an Equal method.
-func Run[I any, V comparable](ctx context.Context, d Definition[I, V], t Transport[I, V], instance I, process int64, inputValueCh <-chan V) (err error) {
+// The generic type C is the compare value, used to compare leader's proposed value with local value and can be anything.
+func Run[I any, V comparable, C any](ctx context.Context, d Definition[I, V, C], t Transport[I, V, C], instance I, process int64, inputValueCh <-chan V, inputValueSourceCh <-chan C) (err error) {
 	defer func() {
 		// Panics are used for assertions and sanity checks to reduce lines of code
 		// and to improve readability. Catch them here.
@@ -181,12 +202,14 @@ func Run[I any, V comparable](ctx context.Context, d Definition[I, V], t Transpo
 	var (
 		round                 int64 = 1
 		inputValue            V
-		ppjCache              []Msg[I, V] // Cached pre-prepare justification for the current round (nil value is unset).
+		inputValueSource      C
+		ppjCache              []Msg[I, V, C] // Cached pre-prepare justification for the current round (nil value is unset).
 		preparedRound         int64
 		preparedValue         V
-		preparedJustification []Msg[I, V]
-		qCommit               []Msg[I, V]
-		buffer                = make(map[int64][]Msg[I, V])
+		compareFailureRound   int64
+		preparedJustification []Msg[I, V, C]
+		qCommit               []Msg[I, V, C]
+		buffer                = make(map[int64][]Msg[I, V, C])
 		dedupRules            = make(map[dedupKey]bool)
 		timerChan             <-chan time.Time
 		stopTimer             func()
@@ -195,7 +218,7 @@ func Run[I any, V comparable](ctx context.Context, d Definition[I, V], t Transpo
 	// === Helpers ==
 
 	// broadcastMsg broadcasts a non-ROUND-CHANGE message for current round.
-	broadcastMsg := func(typ MsgType, value V, justification []Msg[I, V]) error {
+	broadcastMsg := func(typ MsgType, value V, justification []Msg[I, V, C]) error {
 		return t.Broadcast(ctx, typ, instance, process, round,
 			value, 0, zeroVal[V](), justification)
 	}
@@ -209,7 +232,7 @@ func Run[I any, V comparable](ctx context.Context, d Definition[I, V], t Transpo
 	// broadcastOwnPrePrepare broadcasts a PRE-PREPARE message with current state
 	// and our own input value if present, otherwise it caches the justification
 	// to be used when the input value becomes available.
-	broadcastOwnPrePrepare := func(justification []Msg[I, V]) error {
+	broadcastOwnPrePrepare := func(justification []Msg[I, V, C]) error {
 		if justification == nil {
 			panic("bug: justification must not be nil")
 		} else if ppjCache != nil {
@@ -226,7 +249,7 @@ func Run[I any, V comparable](ctx context.Context, d Definition[I, V], t Transpo
 	}
 
 	// bufferMsg adds the message to each process' FIFO queue.
-	bufferMsg := func(msg Msg[I, V]) {
+	bufferMsg := func(msg Msg[I, V, C]) {
 		fifo := buffer[msg.Source()]
 
 		fifo = append(fifo, msg)
@@ -266,7 +289,7 @@ func Run[I any, V comparable](ctx context.Context, d Definition[I, V], t Transpo
 
 	{ // Algorithm 1:11
 		if d.IsLeader(instance, round, process) { // Note round==1 at this point.
-			err := broadcastOwnPrePrepare([]Msg[I, V]{}) // Empty justification since round==1
+			err := broadcastOwnPrePrepare([]Msg[I, V, C]{}) // Empty justification since round==1
 			if err != nil {
 				return err
 			}
@@ -302,7 +325,7 @@ func Run[I any, V comparable](ctx context.Context, d Definition[I, V], t Transpo
 				break
 			}
 
-			if !isJustified(d, instance, msg) { // Drop unjust messages
+			if !isJustified(d, instance, msg, compareFailureRound) { // Drop unjust messages
 				d.LogUnjust(ctx, instance, process, msg)
 				break
 			}
@@ -328,12 +351,34 @@ func Run[I any, V comparable](ctx context.Context, d Definition[I, V], t Transpo
 				err = broadcastMsg(MsgPrepare, msg.Value(), nil)
 
 			case UponQuorumPrepares: // Algorithm 2:4
-				// Only applicable to current round
-				preparedRound = round /* == msg.Round*/
-				preparedValue = msg.Value()
-				preparedJustification = justification
+				var errC error
+
+				inputValueSource, errC = compare(ctx, d, msg, inputValueSourceCh, inputValueSource, timerChan)
+				if errC != nil {
+					switch {
+					case errors.Is(errC, errCompare):
+						compareFailureRound = msg.Round()
+					case errors.Is(errC, errTimeout):
+						// As compare function is blocking on waiting local data, round might timeout in the meantime.
+						// If this happens, we trigger round change.
+						// Algorithm 3:1
+						changeRound(round+1, UponRoundTimeout)
+
+						stopTimer()
+						timerChan, stopTimer = d.NewTimer(round)
+
+						err = broadcastRoundChange()
+					default:
+						err = errors.New("bug: expected only comparison or timeout error")
+					}
+				} else {
+					// Only applicable to current round
+					preparedRound = round /* == msg.Round*/
+					preparedValue = msg.Value()
+					preparedJustification = justification
 
-				err = broadcastMsg(MsgCommit, preparedValue, nil)
+					err = broadcastMsg(MsgCommit, preparedValue, nil)
+				}
 
 			case UponQuorumCommits, UponJustifiedDecided: // Algorithm 2:8
 				// Applicable to any round (since can be justified)
@@ -358,7 +403,8 @@ func Run[I any, V comparable](ctx context.Context, d Definition[I, V], t Transpo
 
 			case UponQuorumRoundChanges: // Algorithm 3:11
 				// Only applicable to current round (round > 1)
-				if _, pv, ok := getSingleJustifiedPrPv(d, justification); ok {
+				pr, pv, ok := getSingleJustifiedPrPv(d, justification)
+				if ok && compareFailureRound != pr {
 					// Send pre-prepare using prepared value (not our own input value)
 					err = broadcastMsg(MsgPrePrepare, pv, justification)
 				} else {
@@ -390,9 +436,39 @@ func Run[I any, V comparable](ctx context.Context, d Definition[I, V], t Transpo
 	}
 }
 
+func compare[I any, V comparable, C any](ctx context.Context, d Definition[I, V, C], msg Msg[I, V, C], inputValueSourceCh <-chan C, inputValueSource C, timerChan <-chan time.Time) (C, error) {
+	compareErr := make(chan error, 1)
+	compareValue := make(chan C, 1)
+
+	ctxCompare, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	// d.Compare has 2 roles:
+	// 1. Read from the inputValueSourceCh (if inputValueSource is empty). If it read from the channel, it returns the value on compareValue channel.
+	// 2. Compare the value read from inputValueSourceCh (or inputValueSource if it is not empty) to the value proposed by the leader.
+	// If comparison or any other unexpected error occurs, the error is returned on compareErr channel.
+	go d.Compare(ctxCompare, msg, inputValueSourceCh, inputValueSource, compareErr, compareValue)
+
+	for {
+		select {
+		case err := <-compareErr:
+			if err != nil {
+				log.Warn(ctx, errCompare.Error(), err)
+				return inputValueSource, errCompare
+			}
+
+			return inputValueSource, nil
+		case inputValueSource = <-compareValue:
+		case <-timerChan:
+			log.Warn(ctx, "", errors.New("timeout on waiting for data used for comparing local and leader proposed data"))
+			return inputValueSource, errTimeout
+		}
+	}
+}
+
 // extractRoundMsgs returns all messages from the provided round.
-func extractRoundMsgs[I any, V comparable](buffer map[int64][]Msg[I, V], round int64) []Msg[I, V] {
-	var resp []Msg[I, V]
+func extractRoundMsgs[I any, V comparable, C any](buffer map[int64][]Msg[I, V, C], round int64) []Msg[I, V, C] {
+	var resp []Msg[I, V, C]
 
 	for _, msgs := range buffer {
 		for _, msg := range msgs {
@@ -406,7 +482,7 @@ func extractRoundMsgs[I any, V comparable](buffer map[int64][]Msg[I, V], round i
 }
 
 // classify returns the rule triggered upon receipt of the last message and its justifications.
-func classify[I any, V comparable](d Definition[I, V], instance I, round, process int64, buffer map[int64][]Msg[I, V], msg Msg[I, V]) (UponRule, []Msg[I, V]) {
+func classify[I any, V comparable, C any](d Definition[I, V, C], instance I, round, process int64, buffer map[int64][]Msg[I, V, C], msg Msg[I, V, C]) (UponRule, []Msg[I, V, C]) {
 	switch msg.Type() {
 	case MsgDecided:
 		return UponJustifiedDecided, msg.Justification()
@@ -484,7 +560,7 @@ func classify[I any, V comparable](d Definition[I, V], instance I, round, proces
 
 // nextMinRound implements algorithm 3:6 and returns the next minimum round
 // from received round change messages.
-func nextMinRound[I any, V comparable](d Definition[I, V], frc []Msg[I, V], round int64) int64 {
+func nextMinRound[I any, V comparable, C any](d Definition[I, V, C], frc []Msg[I, V, C], round int64) int64 {
 	// Get all RoundChange messages with round (rj) higher than current round (ri)
 	if len(frc) < d.Faulty()+1 {
 		panic("bug: Frc too short")
@@ -509,11 +585,11 @@ func nextMinRound[I any, V comparable](d Definition[I, V], frc []Msg[I, V], roun
 }
 
 // isJustified returns true if message is justified or if it does not need justification.
-func isJustified[I any, V comparable](d Definition[I, V], instance I, msg Msg[I, V]) bool {
+func isJustified[I any, V comparable, C any](d Definition[I, V, C], instance I, msg Msg[I, V, C], compareFailureRound int64) bool {
 	//nolint:revive // `case MsgPrepare` and `case MsgCommit` having same result is not an issue, it improves readability.
 	switch msg.Type() {
 	case MsgPrePrepare:
-		return isJustifiedPrePrepare(d, instance, msg)
+		return isJustifiedPrePrepare(d, instance, msg, compareFailureRound)
 	case MsgPrepare:
 		return true
 	case MsgCommit:
@@ -529,7 +605,7 @@ func isJustified[I any, V comparable](d Definition[I, V], instance I, msg Msg[I,
 
 // isJustifiedRoundChange returns true if the ROUND_CHANGE message's
 // prepared round and value is justified.
-func isJustifiedRoundChange[I any, V comparable](d Definition[I, V], msg Msg[I, V]) bool {
+func isJustifiedRoundChange[I any, V comparable, C any](d Definition[I, V, C], msg Msg[I, V, C]) bool {
 	if msg.Type() != MsgRoundChange {
 		panic("bug: not a round change message")
 	}
@@ -540,8 +616,7 @@ func isJustifiedRoundChange[I any, V comparable](d Definition[I, V], msg Msg[I,
 	pv := msg.PreparedValue()
 
 	if len(prepares) == 0 {
-		// If no justification, ensure null prepared round and value.
-		return pr == 0 && isZeroVal(pv)
+		return true
 	}
 
 	// No need to check for all possible combinations, since justified should only contain a one.
@@ -550,7 +625,7 @@ func isJustifiedRoundChange[I any, V comparable](d Definition[I, V], msg Msg[I,
 		return false
 	}
 
-	uniq := uniqSource[I, V]()
+	uniq := uniqSource[I, V, C]()
 	for _, prepare := range prepares {
 		if !uniq(prepare) {
 			return false
@@ -574,7 +649,7 @@ func isJustifiedRoundChange[I any, V comparable](d Definition[I, V], msg Msg[I,
 
 // isJustifiedDecided returns true if the decided message is justified by quorum COMMIT messages
 // of identical round and value.
-func isJustifiedDecided[I any, V comparable](d Definition[I, V], msg Msg[I, V]) bool {
+func isJustifiedDecided[I any, V comparable, C any](d Definition[I, V, C], msg Msg[I, V, C]) bool {
 	if msg.Type() != MsgDecided {
 		panic("bug: not a decided message")
 	}
@@ -586,7 +661,7 @@ func isJustifiedDecided[I any, V comparable](d Definition[I, V], msg Msg[I, V])
 }
 
 // isJustifiedPrePrepare returns true if the PRE-PREPARE message is justified.
-func isJustifiedPrePrepare[I any, V comparable](d Definition[I, V], instance I, msg Msg[I, V]) bool {
+func isJustifiedPrePrepare[I any, V comparable, C any](d Definition[I, V, C], instance I, msg Msg[I, V, C], compareFailureRound int64) bool {
 	if msg.Type() != MsgPrePrepare {
 		panic("bug: not a preprepare message")
 	}
@@ -595,7 +670,8 @@ func isJustifiedPrePrepare[I any, V comparable](d Definition[I, V], instance I,
 		return false
 	}
 
-	if msg.Round() == 1 {
+	// Justified if PrePrepare is the first round OR if comparison failed previous round.
+	if msg.Round() == 1 || (msg.Round() == compareFailureRound+1) {
 		return true
 	}
 
@@ -613,7 +689,7 @@ func isJustifiedPrePrepare[I any, V comparable](d Definition[I, V], instance I,
 
 // containsJustifiedQrc implements algorithm 4:1 and returns true and pv if
 // the messages contains a justified quorum ROUND_CHANGEs (Qrc).
-func containsJustifiedQrc[I any, V comparable](d Definition[I, V], justification []Msg[I, V], round int64) (V, bool) {
+func containsJustifiedQrc[I any, V comparable, C any](d Definition[I, V, C], justification []Msg[I, V, C], round int64) (V, bool) {
 	qrc := filterRoundChange(justification, round)
 	if len(qrc) < d.Quorum() {
 		return zeroVal[V](), false
@@ -663,12 +739,12 @@ func containsJustifiedQrc[I any, V comparable](d Definition[I, V], justification
 
 // getSingleJustifiedPrPv extracts the single justified Pr and Pv from quorum
 // PREPARES in list of messages. It expects only one possible combination.
-func getSingleJustifiedPrPv[I any, V comparable](d Definition[I, V], msgs []Msg[I, V]) (int64, V, bool) {
+func getSingleJustifiedPrPv[I any, V comparable, C any](d Definition[I, V, C], msgs []Msg[I, V, C]) (int64, V, bool) {
 	var (
 		pr    int64
 		pv    V
 		count int
-		uniq  = uniqSource[I, V]()
+		uniq  = uniqSource[I, V, C]()
 	)
 
 	for _, msg := range msgs {
@@ -694,7 +770,7 @@ func getSingleJustifiedPrPv[I any, V comparable](d Definition[I, V], msgs []Msg[
 }
 
 // getJustifiedQrc implements algorithm 4:1 and returns a justified quorum ROUND_CHANGEs (Qrc).
-func getJustifiedQrc[I any, V comparable](d Definition[I, V], all []Msg[I, V], round int64) ([]Msg[I, V], bool) {
+func getJustifiedQrc[I any, V comparable, C any](d Definition[I, V, C], all []Msg[I, V, C], round int64) ([]Msg[I, V, C], bool) {
 	if qrc, ok := quorumNullPrepared(d, all, round); ok {
 		// Return any quorum null pv ROUND_CHANGE messages as Qrc.
 		return qrc, true
@@ -705,11 +781,11 @@ func getJustifiedQrc[I any, V comparable](d Definition[I, V], all []Msg[I, V], r
 	for _, prepares := range getPrepareQuorums(d, all) {
 		// See if we have quorum ROUND-CHANGE with HIGHEST_PREPARED(qrc) == prepares.Round.
 		var (
-			qrc                []Msg[I, V]
+			qrc                []Msg[I, V, C]
 			hasHighestPrepared bool
 			pr                 = prepares[0].Round()
 			pv                 = prepares[0].Value()
-			uniq               = uniqSource[I, V]()
+			uniq               = uniqSource[I, V, C]()
 		)
 		for _, rc := range roundChanges {
 			if rc.PreparedRound() > pr {
@@ -738,8 +814,8 @@ func getJustifiedQrc[I any, V comparable](d Definition[I, V], all []Msg[I, V], r
 // getFPlus1RoundChanges returns true and Faulty+1 ROUND-CHANGE messages (Frc) with
 // the rounds higher than the provided round. It returns the highest round
 // per process in order to jump furthest.
-func getFPlus1RoundChanges[I any, V comparable](d Definition[I, V], all []Msg[I, V], round int64) ([]Msg[I, V], bool) {
-	highestBySource := make(map[int64]Msg[I, V])
+func getFPlus1RoundChanges[I any, V comparable, C any](d Definition[I, V, C], all []Msg[I, V, C], round int64) ([]Msg[I, V, C], bool) {
+	highestBySource := make(map[int64]Msg[I, V, C])
 
 	for _, msg := range all {
 		if msg.Type() != MsgRoundChange {
@@ -765,7 +841,7 @@ func getFPlus1RoundChanges[I any, V comparable](d Definition[I, V], all []Msg[I,
 		return nil, false
 	}
 
-	var resp []Msg[I, V]
+	var resp []Msg[I, V, C]
 	for _, msg := range highestBySource {
 		resp = append(resp, msg)
 	}
@@ -774,26 +850,26 @@ func getFPlus1RoundChanges[I any, V comparable](d Definition[I, V], all []Msg[I,
 }
 
 // preparedKey defines the round and value of set of identical PREPARE messages.
-type preparedKey[I any, V comparable] struct {
+type preparedKey[I any, V comparable, C any] struct {
 	round int64
 	value V
 }
 
 // getPrepareQuorums returns all sets of quorum PREPARE messages
 // with identical rounds and values.
-func getPrepareQuorums[I any, V comparable](d Definition[I, V], all []Msg[I, V]) [][]Msg[I, V] {
-	sets := make(map[preparedKey[I, V]]map[int64]Msg[I, V]) // map[preparedKey]map[process]Msg
+func getPrepareQuorums[I any, V comparable, C any](d Definition[I, V, C], all []Msg[I, V, C]) [][]Msg[I, V, C] {
+	sets := make(map[preparedKey[I, V, C]]map[int64]Msg[I, V, C]) // map[preparedKey]map[process]Msg
 
 	for _, msg := range all { // Flatten to get PREPARES included as ROUND-CHANGE justifications.
 		if msg.Type() != MsgPrepare {
 			continue
 		}
 
-		key := preparedKey[I, V]{round: msg.Round(), value: msg.Value()}
+		key := preparedKey[I, V, C]{round: msg.Round(), value: msg.Value()}
 
 		msgs, ok := sets[key]
 		if !ok {
-			msgs = make(map[int64]Msg[I, V])
+			msgs = make(map[int64]Msg[I, V, C])
 		}
 
 		msgs[msg.Source()] = msg
@@ -801,14 +877,14 @@ func getPrepareQuorums[I any, V comparable](d Definition[I, V], all []Msg[I, V])
 	}
 
 	// Return all quorums
-	var quorums [][]Msg[I, V]
+	var quorums [][]Msg[I, V, C]
 
 	for _, msgs := range sets {
 		if len(msgs) < d.Quorum() {
 			continue
 		}
 
-		var quorum []Msg[I, V]
+		var quorum []Msg[I, V, C]
 		for _, msg := range msgs {
 			quorum = append(quorum, msg)
 		}
@@ -821,7 +897,7 @@ func getPrepareQuorums[I any, V comparable](d Definition[I, V], all []Msg[I, V])
 
 // quorumNullPrepared implements condition J1 and returns Qrc and true if a quorum
 // of round changes messages (Qrc) for the round have null prepared round and value.
-func quorumNullPrepared[I any, V comparable](d Definition[I, V], all []Msg[I, V], round int64) ([]Msg[I, V], bool) {
+func quorumNullPrepared[I any, V comparable, C any](d Definition[I, V, C], all []Msg[I, V, C], round int64) ([]Msg[I, V, C], bool) {
 	var (
 		nullPr int64
 		nullPv V
@@ -833,21 +909,21 @@ func quorumNullPrepared[I any, V comparable](d Definition[I, V], all []Msg[I, V]
 }
 
 // filterByRoundAndValue returns the messages matching the type and value.
-func filterByRoundAndValue[I any, V comparable](msgs []Msg[I, V], typ MsgType, round int64, value V) []Msg[I, V] {
+func filterByRoundAndValue[I any, V comparable, C any](msgs []Msg[I, V, C], typ MsgType, round int64, value V) []Msg[I, V, C] {
 	return filterMsgs(msgs, typ, round, &value, nil, nil)
 }
 
 // filterRoundChange returns all round change messages for the provided round.
-func filterRoundChange[I any, V comparable](msgs []Msg[I, V], round int64) []Msg[I, V] {
+func filterRoundChange[I any, V comparable, C any](msgs []Msg[I, V, C], round int64) []Msg[I, V, C] {
 	return filterMsgs(msgs, MsgRoundChange, round, nil, nil, nil)
 }
 
 // filterMsgs returns one message per process matching the provided type and round
 // and optional value, pr, pv.
-func filterMsgs[I any, V comparable](msgs []Msg[I, V], typ MsgType, round int64, value *V, pr *int64, pv *V) []Msg[I, V] {
+func filterMsgs[I any, V comparable, C any](msgs []Msg[I, V, C], typ MsgType, round int64, value *V, pr *int64, pv *V) []Msg[I, V, C] {
 	var (
-		resp []Msg[I, V]
-		uniq = uniqSource[I, V]()
+		resp []Msg[I, V, C]
+		uniq = uniqSource[I, V, C]()
 	)
 
 	for _, msg := range msgs {
@@ -892,8 +968,8 @@ func isZeroVal[V comparable](v V) bool {
 
 // flatten returns the buffer as a list containing all the buffered messages
 // as well as all their justifications.
-func flatten[I any, V comparable](buffer map[int64][]Msg[I, V]) []Msg[I, V] {
-	var resp []Msg[I, V]
+func flatten[I any, V comparable, C any](buffer map[int64][]Msg[I, V, C]) []Msg[I, V, C] {
+	var resp []Msg[I, V, C]
 
 	for _, msgs := range buffer {
 		for _, msg := range msgs {
@@ -911,7 +987,7 @@ func flatten[I any, V comparable](buffer map[int64][]Msg[I, V]) []Msg[I, V] {
 }
 
 // uniqSource returns a function that returns true if the message is from a unique source.
-func uniqSource[I any, V comparable](msgs ...Msg[I, V]) func(Msg[I, V]) bool {
+func uniqSource[I any, V comparable, C any](msgs ...Msg[I, V, C]) func(Msg[I, V, C]) bool {
 	dedup := make(map[int64]bool)
 	for _, msg := range msgs {
 		if dedup[msg.Source()] {
@@ -921,7 +997,7 @@ func uniqSource[I any, V comparable](msgs ...Msg[I, V]) func(Msg[I, V]) bool {
 		dedup[msg.Source()] = true
 	}
 
-	return func(msg Msg[I, V]) bool {
+	return func(msg Msg[I, V, C]) bool {
 		if dedup[msg.Source()] {
 			return false
 		}



Test patch:
diff --git a/core/consensus/controller_test.go b/core/consensus/controller_test.go
index 2a11f454..7be2bcb7 100644
--- a/core/consensus/controller_test.go
+++ b/core/consensus/controller_test.go
@@ -61,7 +61,7 @@ func TestConsensusController(t *testing.T) {
 	debugger := csmocks.NewDebugger(t)
 	ctx := context.Background()
 
-	controller, err := consensus.NewConsensusController(ctx, hosts[0], new(p2p.Sender), peers, p2pkeys[0], deadlineFunc, gaterFunc, debugger)
+	controller, err := consensus.NewConsensusController(ctx, hosts[0], new(p2p.Sender), peers, p2pkeys[0], deadlineFunc, gaterFunc, debugger, false)
 	require.NoError(t, err)
 	require.NotNil(t, controller)
 
diff --git a/core/consensus/qbft/msg_internal_test.go b/core/consensus/qbft/msg_internal_test.go
index 70d95bdc..391218f7 100644
--- a/core/consensus/qbft/msg_internal_test.go
+++ b/core/consensus/qbft/msg_internal_test.go
@@ -105,7 +105,7 @@ func TestPartialLegacyNewMsg(t *testing.T) {
 	require.ErrorContains(t, err, "value hash not found in values")
 }
 
-// NewRandomMsgForT returns a random qbft message.
+// newRandomQBFTMsg returns a random qbft message.
 func newRandomQBFTMsg(t *testing.T) *pbv1.QBFTMsg {
 	t.Helper()
 
diff --git a/core/consensus/qbft/qbft_internal_test.go b/core/consensus/qbft/qbft_internal_test.go
index d52e6139..7ae342ec 100644
--- a/core/consensus/qbft/qbft_internal_test.go
+++ b/core/consensus/qbft/qbft_internal_test.go
@@ -10,6 +10,7 @@ import (
 	k1 "github.com/decred/dcrd/dcrec/secp256k1/v4"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/anypb"
 
 	"github.com/obolnetwork/charon/app/k1util"
@@ -29,7 +30,7 @@ func TestDebugRoundChange(t *testing.T) {
 
 	tests := []struct {
 		name   string
-		msgs   []qbft.Msg[core.Duty, [32]byte]
+		msgs   []qbft.Msg[core.Duty, [32]byte, proto.Message]
 		round  int64
 		leader int
 	}{
@@ -43,7 +44,7 @@ func TestDebugRoundChange(t *testing.T) {
 		},
 		{
 			name: "quorum",
-			msgs: []qbft.Msg[core.Duty, [32]byte]{
+			msgs: []qbft.Msg[core.Duty, [32]byte, proto.Message]{
 				m(0, qbft.MsgRoundChange),
 				m(1, qbft.MsgRoundChange),
 				m(2, qbft.MsgRoundChange),
@@ -111,6 +112,10 @@ func (t testMsg) Value() [32]byte {
 	panic("implement me")
 }
 
+func (t testMsg) ValueSource() (proto.Message, error) {
+	panic("implement me")
+}
+
 func (t testMsg) PreparedRound() int64 {
 	panic("implement me")
 }
@@ -119,7 +124,7 @@ func (t testMsg) PreparedValue() [32]byte {
 	panic("implement me")
 }
 
-func (t testMsg) Justification() []qbft.Msg[core.Duty, [32]byte] {
+func (t testMsg) Justification() []qbft.Msg[core.Duty, [32]byte, proto.Message] {
 	panic("implement me")
 }
 
diff --git a/core/consensus/qbft/qbft_test.go b/core/consensus/qbft/qbft_test.go
index e7ee36fd..30539dd5 100644
--- a/core/consensus/qbft/qbft_test.go
+++ b/core/consensus/qbft/qbft_test.go
@@ -126,7 +126,7 @@ func testQBFTConsensus(t *testing.T, threshold, nodes int) {
 		deadliner := coremocks.NewDeadliner(t)
 		deadliner.On("Add", mock.Anything).Return(true)
 		deadliner.On("C").Return(nil)
-		c, err := qbft.NewConsensus(hosts[i], new(p2p.Sender), peers, p2pkeys[i], deadliner, gaterFunc, sniffer)
+		c, err := qbft.NewConsensus(hosts[i], new(p2p.Sender), peers, p2pkeys[i], deadliner, gaterFunc, sniffer, false)
 		require.NoError(t, err)
 		c.Subscribe(func(_ context.Context, _ core.Duty, set core.UnsignedDataSet) error {
 			results <- set
diff --git a/core/consensus/qbft/sniffed_internal_test.go b/core/consensus/qbft/sniffed_internal_test.go
index e8ae802c..685316b7 100644
--- a/core/consensus/qbft/sniffed_internal_test.go
+++ b/core/consensus/qbft/sniffed_internal_test.go
@@ -26,7 +26,7 @@ import (
 
 var sniffedFile = flag.String("sniffed-file", "", "path to sniffed file")
 
-// TestSniffedInstances simulates all the instances in the sniffed file.
+// TestSniffedFile simulates all the instances in the sniffed file.
 func TestSniffedFile(t *testing.T) {
 	if *sniffedFile == "" {
 		t.Skip("no sniffed file provided")
@@ -79,9 +79,9 @@ func testSniffedInstance(ctx context.Context, t *testing.T, instance *pbv1.Sniff
 
 			return nil
 		}}
-	}, timer.NewIncreasingRoundTimer(), func(qcommit []qbft.Msg[core.Duty, [32]byte]) {})
+	}, timer.NewIncreasingRoundTimer(), func(qcommit []qbft.Msg[core.Duty, [32]byte, proto.Message]) {}, false)
 
-	recvBuffer := make(chan qbft.Msg[core.Duty, [32]byte], len(instance.GetMsgs()))
+	recvBuffer := make(chan qbft.Msg[core.Duty, [32]byte, proto.Message], len(instance.GetMsgs()))
 
 	var duty core.Duty
 
@@ -102,10 +102,10 @@ func testSniffedInstance(ctx context.Context, t *testing.T, instance *pbv1.Sniff
 	}
 
 	// Create a qbft transport from the transport
-	qt := qbft.Transport[core.Duty, [32]byte]{
+	qt := qbft.Transport[core.Duty, [32]byte, proto.Message]{
 		Broadcast: func(context.Context, qbft.MsgType, core.Duty,
 			int64, int64, [32]byte, int64, [32]byte,
-			[]qbft.Msg[core.Duty, [32]byte],
+			[]qbft.Msg[core.Duty, [32]byte, proto.Message],
 		) error {
 			return nil
 		},
@@ -113,7 +113,7 @@ func testSniffedInstance(ctx context.Context, t *testing.T, instance *pbv1.Sniff
 	}
 
 	// Run the algo, blocking until the context is cancelled.
-	err := qbft.Run[core.Duty, [32]byte](ctx, def, qt, duty, instance.GetPeerIdx(), qbft.InputValue([32]byte{1}))
+	err := qbft.Run(ctx, def, qt, duty, instance.GetPeerIdx(), qbft.InputValue([32]byte{1}), qbft.InputValueSource(proto.Message(newRandomQBFTMsg(t))))
 	if expectDecided {
 		require.ErrorIs(t, err, context.Canceled)
 	} else {
diff --git a/core/consensus/qbft/strategysim_internal_test.go b/core/consensus/qbft/strategysim_internal_test.go
index a6ef00e8..a4ebef38 100644
--- a/core/consensus/qbft/strategysim_internal_test.go
+++ b/core/consensus/qbft/strategysim_internal_test.go
@@ -22,6 +22,7 @@ import (
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"go.uber.org/zap/zaptest"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
@@ -358,7 +359,7 @@ func testStrategySimulator(t *testing.T, conf ssConfig, syncer zapcore.WriteSync
 
 	var (
 		peerIDs    []peerID
-		transports []qbft.Transport[core.Duty, [32]byte]
+		transports []qbft.Transport[core.Duty, [32]byte, proto.Message]
 	)
 	for peerIdx := range conf.latencyPerPeer {
 		peerIDs = append(peerIDs, peerID{Idx: peerIdx, OK: true})
@@ -375,7 +376,7 @@ func testStrategySimulator(t *testing.T, conf ssConfig, syncer zapcore.WriteSync
 		def := newSimDefinition(
 			len(conf.latencyPerPeer),
 			conf.roundTimerFunc(clock),
-			func(qcommit []qbft.Msg[core.Duty, [32]byte]) {
+			func(qcommit []qbft.Msg[core.Duty, [32]byte, proto.Message]) {
 				res = result{
 					PeerIdx:  p.Idx,
 					Decided:  true,
@@ -389,6 +390,7 @@ func testStrategySimulator(t *testing.T, conf ssConfig, syncer zapcore.WriteSync
 
 		// Setup unique non-zero value per peer
 		valCh := make(chan [32]byte, 1)
+		valSrcCh := make(chan proto.Message, 1)
 		enqueueValue := func() {
 			var val [32]byte
 
@@ -418,7 +420,7 @@ func testStrategySimulator(t *testing.T, conf ssConfig, syncer zapcore.WriteSync
 
 		log.Debug(ctx, "Starting peer")
 
-		err := qbft.Run(ctx, def, transports[p.Idx], core.Duty{Slot: uint64(conf.seed)}, p.Idx, valCh)
+		err := qbft.Run(ctx, def, transports[p.Idx], core.Duty{Slot: uint64(conf.seed)}, p.Idx, valCh, valSrcCh)
 		if err != nil && !errors.Is(err, context.Canceled) {
 			return res, err
 		}
@@ -483,21 +485,24 @@ func gosched() {
 }
 
 func newSimDefinition(nodes int, roundTimer timer.RoundTimer,
-	decideCallback func(qcommit []qbft.Msg[core.Duty, [32]byte]),
-) qbft.Definition[core.Duty, [32]byte] {
-	quorum := qbft.Definition[int, int]{Nodes: nodes}.Quorum()
+	decideCallback func(qcommit []qbft.Msg[core.Duty, [32]byte, proto.Message]),
+) qbft.Definition[core.Duty, [32]byte, proto.Message] {
+	quorum := qbft.Definition[int, int, int]{Nodes: nodes}.Quorum()
 
-	return qbft.Definition[core.Duty, [32]byte]{
+	return qbft.Definition[core.Duty, [32]byte, proto.Message]{
 		IsLeader: func(duty core.Duty, round, process int64) bool {
 			return leader(duty, round, nodes) == process
 		},
-		Decide: func(ctx context.Context, duty core.Duty, _ [32]byte, qcommit []qbft.Msg[core.Duty, [32]byte]) {
+		Decide: func(ctx context.Context, duty core.Duty, _ [32]byte, qcommit []qbft.Msg[core.Duty, [32]byte, proto.Message]) {
 			decideCallback(qcommit)
 		},
+		Compare: func(ctx context.Context, qcommit qbft.Msg[core.Duty, [32]byte, proto.Message], inputValueSourceCh <-chan proto.Message, inputValueSource proto.Message, returnErr chan error, returnRes chan proto.Message) {
+			returnErr <- nil
+		},
 		NewTimer:  roundTimer.Timer,
-		LogUnjust: func(context.Context, core.Duty, int64, qbft.Msg[core.Duty, [32]byte]) {},
+		LogUnjust: func(context.Context, core.Duty, int64, qbft.Msg[core.Duty, [32]byte, proto.Message]) {},
 		LogRoundChange: func(ctx context.Context, duty core.Duty, process,
-			round, newRound int64, uponRule qbft.UponRule, msgs []qbft.Msg[core.Duty, [32]byte],
+			round, newRound int64, uponRule qbft.UponRule, msgs []qbft.Msg[core.Duty, [32]byte, proto.Message],
 		) {
 			fields := []z.Field{
 				z.Any("rule", uponRule),
@@ -518,7 +523,7 @@ func newSimDefinition(nodes int, roundTimer timer.RoundTimer,
 		},
 		// LogUponRule logs upon rules at debug level.
 		LogUponRule: func(ctx context.Context, _ core.Duty, _, round int64,
-			_ qbft.Msg[core.Duty, [32]byte], uponRule qbft.UponRule,
+			_ qbft.Msg[core.Duty, [32]byte, proto.Message], uponRule qbft.UponRule,
 		) {
 			log.Debug(ctx, "QBFT upon rule triggered", z.Any("rule", uponRule), z.I64("round", round))
 		},
@@ -538,7 +543,7 @@ type result struct {
 }
 
 type tuple struct {
-	Msg    qbft.Msg[core.Duty, [32]byte]
+	Msg    qbft.Msg[core.Duty, [32]byte, proto.Message]
 	To     int64
 	Arrive time.Time
 }
@@ -566,7 +571,7 @@ type transportSimulator struct {
 	instances map[int64]*transportInstance
 }
 
-func (s *transportSimulator) enqueue(msg qbft.Msg[core.Duty, [32]byte]) {
+func (s *transportSimulator) enqueue(msg qbft.Msg[core.Duty, [32]byte, proto.Message]) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
@@ -612,7 +617,7 @@ func (s *transportSimulator) processBuffer() {
 	s.buffer = remaining
 }
 
-func (s *transportSimulator) instance(peerIdx int64) qbft.Transport[core.Duty, [32]byte] {
+func (s *transportSimulator) instance(peerIdx int64) qbft.Transport[core.Duty, [32]byte, proto.Message] {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
@@ -621,12 +626,12 @@ func (s *transportSimulator) instance(peerIdx int64) qbft.Transport[core.Duty, [
 		inst = &transportInstance{
 			transportSimulator: s,
 			peerIdx:            peerIdx,
-			receive:            make(chan qbft.Msg[core.Duty, [32]byte], 1000),
+			receive:            make(chan qbft.Msg[core.Duty, [32]byte, proto.Message], 1000),
 		}
 		s.instances[peerIdx] = inst
 	}
 
-	return qbft.Transport[core.Duty, [32]byte]{
+	return qbft.Transport[core.Duty, [32]byte, proto.Message]{
 		Broadcast: inst.Broadcast,
 		Receive:   inst.Receive(),
 	}
@@ -636,12 +641,12 @@ type transportInstance struct {
 	*transportSimulator
 
 	peerIdx int64
-	receive chan qbft.Msg[core.Duty, [32]byte]
+	receive chan qbft.Msg[core.Duty, [32]byte, proto.Message]
 }
 
 func (i *transportInstance) Broadcast(_ context.Context, typ qbft.MsgType,
 	duty core.Duty, source int64, round int64, value [32]byte,
-	pr int64, pv [32]byte, justification []qbft.Msg[core.Duty, [32]byte],
+	pr int64, pv [32]byte, justification []qbft.Msg[core.Duty, [32]byte, proto.Message],
 ) error {
 	dummy, _ := anypb.New(timestamppb.Now())
 	values := map[[32]byte]*anypb.Any{
@@ -683,7 +688,7 @@ func (i *transportInstance) Broadcast(_ context.Context, typ qbft.MsgType,
 	return nil
 }
 
-func (i *transportInstance) Receive() <-chan qbft.Msg[core.Duty, [32]byte] {
+func (i *transportInstance) Receive() <-chan qbft.Msg[core.Duty, [32]byte, proto.Message] {
 	return i.receive
 }
 
diff --git a/core/qbft/qbft_internal_test.go b/core/qbft/qbft_internal_test.go
index bbde2196..f39b3395 100644
--- a/core/qbft/qbft_internal_test.go
+++ b/core/qbft/qbft_internal_test.go
@@ -293,15 +293,15 @@ func testQBFT(t *testing.T, test test) {
 	var (
 		ctx, cancel = context.WithCancel(context.Background())
 		clock       = new(fakeClock)
-		receives    = make(map[int64]chan Msg[int64, int64])
-		broadcast   = make(chan Msg[int64, int64])
-		resultChan  = make(chan []Msg[int64, int64], n)
+		receives    = make(map[int64]chan Msg[int64, int64, int64])
+		broadcast   = make(chan Msg[int64, int64, int64])
+		resultChan  = make(chan []Msg[int64, int64, int64], n)
 		runChan     = make(chan error, n)
 	)
 	defer cancel()
 
 	isLeader := makeIsLeader(n)
-	defs := Definition[int64, int64]{
+	defs := Definition[int64, int64, int64]{
 		IsLeader: isLeader,
 		NewTimer: func(round int64) (<-chan time.Time, func()) {
 			d := time.Second
@@ -311,20 +311,23 @@ func testQBFT(t *testing.T, test test) {
 
 			return clock.NewTimer(d)
 		},
-		Decide: func(_ context.Context, instance int64, value int64, qcommit []Msg[int64, int64]) {
+		Decide: func(_ context.Context, instance int64, value int64, qcommit []Msg[int64, int64, int64]) {
 			resultChan <- qcommit
 		},
-		LogRoundChange: func(ctx context.Context, instance int64, process, round, newRound int64, rule UponRule, msgs []Msg[int64, int64]) {
+		Compare: func(ctx context.Context, qcommit Msg[int64, int64, int64], inputValueSourceCh <-chan int64, inputValueSource int64, returnErr chan error, returnRes chan int64) {
+			returnErr <- nil
+		},
+		LogRoundChange: func(ctx context.Context, instance int64, process, round, newRound int64, rule UponRule, msgs []Msg[int64, int64, int64]) {
 			t.Logf("%s %v@%d change to %d ~= %v", clock.NowStr(), process, round, newRound, rule)
 		},
-		LogUponRule: func(_ context.Context, instance int64, process, round int64, msg Msg[int64, int64], rule UponRule) {
+		LogUponRule: func(_ context.Context, instance int64, process, round int64, msg Msg[int64, int64, int64], rule UponRule) {
 			t.Logf("%s %d => %v@%d -> %v@%d ~= %v", clock.NowStr(), msg.Source(), msg.Type(), msg.Round(), process, round, rule)
 
 			if round > maxRound {
 				cancel()
 			}
 		},
-		LogUnjust: func(_ context.Context, instance int64, process int64, msg Msg[int64, int64]) {
+		LogUnjust: func(_ context.Context, instance int64, process int64, msg Msg[int64, int64, int64]) {
 			if test.Fuzz {
 				return // Ignore unjust messages when fuzzing.
 			}
@@ -337,11 +340,11 @@ func testQBFT(t *testing.T, test test) {
 	}
 
 	for i := int64(1); i <= n; i++ {
-		receive := make(chan Msg[int64, int64], 1000)
+		receive := make(chan Msg[int64, int64, int64], 1000)
 		receives[i] = receive
-		trans := Transport[int64, int64]{
+		trans := Transport[int64, int64, int64]{
 			Broadcast: func(ctx context.Context, typ MsgType, instance int64, source int64, round int64, value int64,
-				pr int64, pv int64, justify []Msg[int64, int64],
+				pr int64, pv int64, justify []Msg[int64, int64, int64],
 			) error {
 				if round > maxRound {
 					return errors.New("max round reach")
@@ -354,7 +357,7 @@ func testQBFT(t *testing.T, test test) {
 
 				t.Logf("%s %v => %v@%d", clock.NowStr(), source, typ, round)
 
-				msg := newMsg(typ, instance, source, round, value, pr, pv, justify)
+				msg := newMsg(typ, instance, source, round, value, value, pr, pv, justify)
 				receive <- msg // Always send to self first (no jitter, no drops).
 
 				bcast(t, broadcast, msg, test.BCastJitterMS, clock)
@@ -388,6 +391,7 @@ func testQBFT(t *testing.T, test test) {
 			// - or expect multiple rounds
 			// - or otherwise only the leader of round 1.
 			vChan := make(chan int64, 1)
+			vsChan := make(chan int64, 1)
 
 			if delay, ok := test.ValueDelay[i]; ok {
 				go func() {
@@ -404,7 +408,7 @@ func testQBFT(t *testing.T, test test) {
 				go func() { vChan <- i }()
 			}
 
-			runChan <- Run(ctx, defs, trans, test.Instance, i, vChan)
+			runChan <- Run(ctx, defs, trans, test.Instance, i, vChan, vsChan)
 		}(i)
 	}
 
@@ -413,7 +417,7 @@ func testQBFT(t *testing.T, test test) {
 	}
 
 	var (
-		results = make(map[int64]Msg[int64, int64])
+		results = make(map[int64]Msg[int64, int64, int64])
 		count   int
 		decided bool
 		done    int
@@ -489,7 +493,7 @@ func testQBFT(t *testing.T, test test) {
 }
 
 // fuzz broadcasts random messages from the peer every 100ms (10/round).
-func fuzz(ctx context.Context, clock *fakeClock, broadcast chan Msg[int64, int64], instance, peerIdx int64) {
+func fuzz(ctx context.Context, clock *fakeClock, broadcast chan Msg[int64, int64, int64], instance, peerIdx int64) {
 	for {
 		timer, stop := clock.NewTimer(time.Millisecond * 100)
 		select {
@@ -517,7 +521,7 @@ func randomMsg(instance, peerIdx int64) msg {
 }
 
 // bcast delays the message broadcast by between 1x and 2x jitterMS and drops messages.
-func bcast(t *testing.T, broadcast chan Msg[int64, int64], msg Msg[int64, int64], jitterMS int, clock *fakeClock) {
+func bcast(t *testing.T, broadcast chan Msg[int64, int64, int64], msg Msg[int64, int64, int64], jitterMS int, clock *fakeClock) {
 	t.Helper()
 
 	if jitterMS == 0 {
@@ -537,9 +541,9 @@ func bcast(t *testing.T, broadcast chan Msg[int64, int64], msg Msg[int64, int64]
 }
 
 // newMsg returns a new message to be broadcast.
-func newMsg(typ MsgType, instance int64, source int64, round int64, value int64,
-	pr int64, pv int64, justify []Msg[int64, int64],
-) Msg[int64, int64] {
+func newMsg(typ MsgType, instance int64, source int64, round int64, value int64, valueSource int64,
+	pr int64, pv int64, justify []Msg[int64, int64, int64],
+) Msg[int64, int64, int64] {
 	var msgs []msg
 	for _, j := range justify {
 		m := j.(msg)
@@ -548,28 +552,30 @@ func newMsg(typ MsgType, instance int64, source int64, round int64, value int64,
 	}
 
 	return msg{
-		msgType:  typ,
-		instance: instance,
-		peerIdx:  source,
-		round:    round,
-		value:    value,
-		pr:       pr,
-		pv:       pv,
-		justify:  msgs,
+		msgType:     typ,
+		instance:    instance,
+		peerIdx:     source,
+		round:       round,
+		value:       value,
+		valueSource: valueSource,
+		pr:          pr,
+		pv:          pv,
+		justify:     msgs,
 	}
 }
 
-var _ Msg[int64, int64] = msg{}
+var _ Msg[int64, int64, int64] = msg{}
 
 type msg struct {
-	msgType  MsgType
-	instance int64
-	peerIdx  int64
-	round    int64
-	value    int64
-	pr       int64
-	pv       int64
-	justify  []msg
+	msgType     MsgType
+	instance    int64
+	peerIdx     int64
+	round       int64
+	value       int64
+	valueSource int64
+	pr          int64
+	pv          int64
+	justify     []msg
 }
 
 func (m msg) Type() MsgType {
@@ -592,6 +598,10 @@ func (m msg) Value() int64 {
 	return m.value
 }
 
+func (m msg) ValueSource() (int64, error) {
+	return m.valueSource, nil
+}
+
 func (m msg) PreparedRound() int64 {
 	return m.pr
 }
@@ -600,8 +610,8 @@ func (m msg) PreparedValue() int64 {
 	return m.pv
 }
 
-func (m msg) Justification() []Msg[int64, int64] {
-	var resp []Msg[int64, int64]
+func (m msg) Justification() []Msg[int64, int64, int64] {
+	var resp []Msg[int64, int64, int64]
 	for _, msg := range m.justify {
 		resp = append(resp, msg)
 	}
@@ -626,12 +636,12 @@ func TestIsJustifiedPrePrepare(t *testing.T) {
 		{msgType: 2, instance: 1, peerIdx: 2, round: 2, value: 2, pr: 0, pv: 0},
 	}}
 
-	def := Definition[int64, int64]{
+	def := Definition[int64, int64, int64]{
 		IsLeader: makeIsLeader(n),
 		Nodes:    n,
 	}
 
-	ok := isJustifiedPrePrepare[int64, int64](def, instance, preprepare)
+	ok := isJustifiedPrePrepare(def, instance, preprepare, 0)
 	require.True(t, ok)
 }
 
@@ -640,7 +650,7 @@ func TestFormulas(t *testing.T) {
 	assert := func(t *testing.T, n, q, f int) {
 		t.Helper()
 
-		d := Definition[any, int64]{Nodes: n}
+		d := Definition[any, int64, any]{Nodes: n}
 		require.Equalf(t, q, d.Quorum(), "Quorum given N=%d", n)
 		require.Equalf(t, f, d.Faulty(), "Faulty given N=%d", n)
 	}
@@ -686,7 +696,7 @@ func TestDuplicatePrePreparesRules(t *testing.T) {
 		leader   = 2
 	)
 
-	newPreprepare := func(round int64) Msg[int64, int64] {
+	newPreprepare := func(round int64) Msg[int64, int64, int64] {
 		return msg{
 			msgType: MsgPrePrepare,
 			peerIdx: leader,
@@ -699,7 +709,7 @@ func TestDuplicatePrePreparesRules(t *testing.T) {
 	def.IsLeader = func(_ int64, _ int64, process int64) bool {
 		return process == leader
 	}
-	def.LogUponRule = func(ctx context.Context, instance int64, process, round int64, msg Msg[int64, int64], uponRule UponRule) {
+	def.LogUponRule = func(ctx context.Context, instance int64, process, round int64, msg Msg[int64, int64, int64], uponRule UponRule) {
 		log.Info(ctx, "UponRule", z.Str("rule", uponRule.String()), z.I64("round", msg.Round()))
 		require.Equal(t, uponRule, UponJustifiedPrePrepare)
 
@@ -715,7 +725,7 @@ func TestDuplicatePrePreparesRules(t *testing.T) {
 		require.Fail(t, "unexpected round", "round=%d", round)
 	}
 
-	rChan := make(chan Msg[int64, int64], 2)
+	rChan := make(chan Msg[int64, int64, int64], 2)
 	rChan <- newPreprepare(1)
 
 	rChan <- newPreprepare(2)
@@ -723,21 +733,255 @@ func TestDuplicatePrePreparesRules(t *testing.T) {
 	transport := noopTransport
 	transport.Receive = rChan
 
-	_ = Run(ctx, def, transport, 0, noLeader, InputValue(int64(1)))
+	_ = Run(ctx, def, transport, 0, noLeader, InputValue(int64(1)), InputValueSource(int64(2)))
 }
 
 // noopTransport is a transport that does nothing.
-var noopTransport = Transport[int64, int64]{
-	Broadcast: func(context.Context, MsgType, int64, int64, int64, int64, int64, int64, []Msg[int64, int64]) error {
+var noopTransport = Transport[int64, int64, int64]{
+	Broadcast: func(context.Context, MsgType, int64, int64, int64, int64, int64, int64, []Msg[int64, int64, int64]) error {
 		return nil
 	},
 }
 
 // noopDef is a definition that does nothing.
-var noopDef = Definition[int64, int64]{
+var noopDef = Definition[int64, int64, int64]{
 	IsLeader:       func(int64, int64, int64) bool { return false },
 	NewTimer:       func(int64) (<-chan time.Time, func()) { return nil, func() {} },
-	LogUponRule:    func(context.Context, int64, int64, int64, Msg[int64, int64], UponRule) {},
-	LogRoundChange: func(context.Context, int64, int64, int64, int64, UponRule, []Msg[int64, int64]) {},
-	LogUnjust:      func(context.Context, int64, int64, Msg[int64, int64]) {},
+	LogUponRule:    func(context.Context, int64, int64, int64, Msg[int64, int64, int64], UponRule) {},
+	LogRoundChange: func(context.Context, int64, int64, int64, int64, UponRule, []Msg[int64, int64, int64]) {},
+	LogUnjust:      func(context.Context, int64, int64, Msg[int64, int64, int64]) {},
+}
+
+type testChainSplit struct {
+	ValueSource map[int64]int64 // Use different value source for certain processes (used for chain-split-halt feature).
+	DecideRound int             // Deterministic consensus at specific round.
+	PreparedVal int             // If prepared value decided, as opposed to leader's value.
+	ShouldHalt  bool            // If halt is expected (no consensus reachead).
+}
+
+var errChainSplitHalt = errors.New("chain split halt")
+
+func TestChainSplit(t *testing.T) {
+	t.Run("same value", func(t *testing.T) {
+		testQBFTChainSplit(t, testChainSplit{
+			DecideRound: 1,
+			ValueSource: map[int64]int64{
+				1: 1,
+				2: 1,
+				3: 1,
+				4: 1,
+			},
+			PreparedVal: 1,
+		})
+	})
+
+	t.Run("non-leader peer has different value", func(t *testing.T) {
+		testQBFTChainSplit(t, testChainSplit{
+			DecideRound: 1,
+			ValueSource: map[int64]int64{
+				1: 1,
+				2: 3,
+				3: 1,
+				4: 1,
+			},
+			PreparedVal: 1,
+		})
+	})
+
+	t.Run("first leader has different value, second leader succeeds", func(t *testing.T) {
+		testQBFTChainSplit(t, testChainSplit{
+			DecideRound: 2,
+			ValueSource: map[int64]int64{
+				1: 3,
+				2: 1,
+				3: 1,
+				4: 1,
+			},
+			PreparedVal: 1,
+		})
+	})
+
+	t.Run("no consensus - halt", func(t *testing.T) {
+		testQBFTChainSplit(t, testChainSplit{
+			ValueSource: map[int64]int64{
+				1: 1,
+				2: 1,
+				3: 3,
+				4: 3,
+			},
+			ShouldHalt: true,
+		})
+	})
+}
+
+func testQBFTChainSplit(t *testing.T, test testChainSplit) {
+	t.Helper()
+
+	const (
+		n         = 4
+		maxRound  = 10
+		fifoLimit = 100
+	)
+
+	var (
+		ctx, cancel            = context.WithCancel(context.Background())
+		clock                  = new(fakeClock)
+		receiveChannelsPerNode = make(map[int64]chan Msg[int64, int64, int64])
+		broadcast              = make(chan Msg[int64, int64, int64])
+		resultChan             = make(chan []Msg[int64, int64, int64], n)
+		runChan                = make(chan error, n)
+		instance               = int64(0)
+	)
+	defer cancel()
+
+	isLeader := makeIsLeader(n)
+	defs := Definition[int64, int64, int64]{
+		IsLeader: isLeader,
+		NewTimer: func(round int64) (<-chan time.Time, func()) {
+			return clock.NewTimer(time.Duration(math.Pow(2, float64(round-1))) * time.Second)
+		},
+		Decide: func(_ context.Context, instance int64, value int64, qcommit []Msg[int64, int64, int64]) {
+			resultChan <- qcommit
+		},
+		Compare: func(ctx context.Context, qcommit Msg[int64, int64, int64], inputValueSourceCh <-chan int64, inputValueSource int64, returnCh chan error, returnIVS chan int64) {
+			vs, _ := qcommit.ValueSource()
+
+			if inputValueSource == 0 {
+				inputValueSource = <-inputValueSourceCh
+				returnIVS <- inputValueSource
+			}
+
+			if vs != inputValueSource {
+				returnCh <- errors.New("missmatch", z.I64("leadervalue", vs), z.I64("localvalue", inputValueSource))
+				return
+			}
+
+			returnCh <- nil
+		},
+		LogRoundChange: func(ctx context.Context, instance int64, process, round, newRound int64, rule UponRule, msgs []Msg[int64, int64, int64]) {
+			t.Logf("%s %v@%d change to %d ~= %v", clock.NowStr(), process, round, newRound, rule)
+		},
+		LogUponRule: func(_ context.Context, instance int64, process, round int64, msg Msg[int64, int64, int64], rule UponRule) {
+			t.Logf("%s %d => %v@%d -> %v@%d ~= %v", clock.NowStr(), msg.Source(), msg.Type(), msg.Round(), process, round, rule)
+
+			if round > maxRound {
+				cancel()
+			}
+		},
+		LogUnjust: func(_ context.Context, instance int64, process int64, msg Msg[int64, int64, int64]) {
+			t.Logf("Unjust: %#v", msg)
+		},
+		Nodes:     n,
+		FIFOLimit: fifoLimit,
+	}
+
+	// Start each charon node
+	for i := int64(1); i <= n; i++ {
+		receive := make(chan Msg[int64, int64, int64], 1000)
+		receiveChannelsPerNode[i] = receive
+		transport := Transport[int64, int64, int64]{
+			Broadcast: func(ctx context.Context, typ MsgType, instance int64, source int64, round int64, value int64,
+				pr int64, pv int64, justify []Msg[int64, int64, int64],
+			) error {
+				if round > maxRound {
+					if test.ShouldHalt {
+						return errChainSplitHalt
+					}
+
+					return errors.New("max round reach")
+				}
+
+				t.Logf("%s %v => %v@%d", clock.NowStr(), source, typ, round)
+
+				msg := newMsg(typ, instance, source, round, value, value, pr, pv, justify)
+				receive <- msg // Always send to self first (no jitter, no drops).
+
+				bcast(t, broadcast, msg, 0, clock)
+
+				return nil
+			},
+			Receive: receive,
+		}
+
+		go func(i int64) {
+			// Only enqueue input values for instances that:
+			// - have a value delay
+			// - or expect multiple rounds
+			// - or otherwise only the leader of round 1.
+			vChan := make(chan int64, 1)
+			vsChan := make(chan int64, 1)
+
+			go func() {
+				vChan <- test.ValueSource[i]
+
+				vsChan <- test.ValueSource[i]
+			}()
+
+			runChan <- Run(ctx, defs, transport, instance, i, vChan, vsChan)
+		}(i)
+	}
+
+	var (
+		results = make(map[int64]Msg[int64, int64, int64])
+		count   int
+		decided bool
+		done    int
+	)
+
+	for {
+		select {
+		case msg := <-broadcast:
+			for target, out := range receiveChannelsPerNode {
+				if target == msg.Source() {
+					continue // Do not broadcast to self, we sent to self already.
+				}
+
+				out <- msg
+
+				if rand.Float64() < 0.1 { // Send 10% messages twice
+					out <- msg
+				}
+			}
+		case qCommit := <-resultChan:
+			for _, commit := range qCommit {
+				// Ensure that all results are the same
+				for _, previous := range results {
+					require.Equal(t, previous.Value(), commit.Value(), "commit values")
+				}
+
+				require.EqualValues(t, test.DecideRound, commit.Round(), "wrong decide round")
+
+				if test.PreparedVal != 0 { // Check prepared value if set
+					require.EqualValues(t, test.PreparedVal, commit.Value(), "wrong prepared value")
+				}
+
+				results[commit.Source()] = commit
+			}
+
+			count++
+			if count != n {
+				continue
+			}
+
+			round := qCommit[0].Round()
+			t.Logf("Got all results in round %d after %s: %#v", round, clock.SinceT0(), results)
+
+			// Trigger shutdown
+			decided = true
+
+			cancel()
+		case err := <-runChan:
+			if !decided && !errors.Is(err, errChainSplitHalt) {
+				require.Fail(t, "unexpected run error", err)
+			}
+
+			done++
+			if done == n {
+				return
+			}
+		default:
+			time.Sleep(time.Microsecond)
+			clock.Advance(time.Millisecond * 1)
+		}
+	}
 }