diff --git a/protocol/v2/ssv/queue/message_prioritizer.go b/protocol/v2/ssv/queue/message_prioritizer.go
index d6eaa747b1..e22a198b8e 100644
--- a/protocol/v2/ssv/queue/message_prioritizer.go
+++ b/protocol/v2/ssv/queue/message_prioritizer.go
@@ -5,8 +5,8 @@ import (
 	specqbft "github.com/ssvlabs/ssv-spec/qbft"
 )
 
-// State represents a portion of the current state
-// that is relevant to the prioritization of messages.
+// State represents Runner state that is useful for comparing the priority of various messages (message priority
+// depends on what the current runner state is).
 type State struct {
 	HasRunningInstance bool
 	Height             specqbft.Height
diff --git a/protocol/v2/ssv/queue/messages.go b/protocol/v2/ssv/queue/messages.go
index 1dd734b6ec..d47c5d81f8 100644
--- a/protocol/v2/ssv/queue/messages.go
+++ b/protocol/v2/ssv/queue/messages.go
@@ -134,7 +134,7 @@ func compareHeightOrSlot(state *State, m *SSVMessage) int {
 		if qbftMsg.Height > state.Height {
 			return 1
 		}
-	} else if pms, ok := m.Body.(*spectypes.PartialSignatureMessages); ok && pms != nil { // everyone likes pms
+	} else if pms, ok := m.Body.(*spectypes.PartialSignatureMessages); ok && pms != nil {
 		if pms.Slot == state.Slot {
 			return 0
 		}
diff --git a/protocol/v2/ssv/runner/aggregator.go b/protocol/v2/ssv/runner/aggregator.go
index 0319ab8790..d4814cb9a3 100644
--- a/protocol/v2/ssv/runner/aggregator.go
+++ b/protocol/v2/ssv/runner/aggregator.go
@@ -153,7 +153,7 @@ func (r *AggregatorRunner) ProcessPreConsensus(ctx context.Context, logger *zap.
 			observability.CommitteeIndexAttribute(duty.CommitteeIndex),
 			observability.ValidatorIndexAttribute(duty.ValidatorIndex)),
 	)
-	res, ver, err := r.GetBeaconNode().SubmitAggregateSelectionProof(ctx, duty.Slot, duty.CommitteeIndex, duty.CommitteeLength, duty.ValidatorIndex, fullSig)
+	res, ver, err := r.beacon.SubmitAggregateSelectionProof(ctx, duty.Slot, duty.CommitteeIndex, duty.CommitteeLength, duty.ValidatorIndex, fullSig)
 	if err != nil {
 		return fmt.Errorf("failed to submit aggregate and proof: %w", err)
 	}
@@ -256,7 +256,7 @@ func (r *AggregatorRunner) ProcessConsensus(ctx context.Context, logger *zap.Log
 
 	r.measurements.StartPostConsensus()
 	span.AddEvent("broadcasting post consensus partial signature message")
-	if err := r.GetNetwork().Broadcast(msgID, msgToBroadcast); err != nil {
+	if err := r.network.Broadcast(msgID, msgToBroadcast); err != nil {
 		return fmt.Errorf("can't broadcast partial post consensus sig: %w", err)
 	}
 	const broadcastedPostConsensusMsgEvent = "broadcasted post-consensus partial signature message"
@@ -325,7 +325,7 @@ func (r *AggregatorRunner) ProcessPostConsensus(ctx context.Context, logger *zap
 	span.AddEvent(submittingSignedAggregateProofEvent)
 
 	start := time.Now()
-	if err := r.GetBeaconNode().SubmitSignedAggregateSelectionProof(ctx, msg); err != nil {
+	if err := r.beacon.SubmitSignedAggregateSelectionProof(ctx, msg); err != nil {
 		recordFailedSubmission(ctx, spectypes.BNRoleAggregator)
 		const errMsg = "could not submit to Beacon chain reconstructed contribution and proof"
 		logger.Error(errMsg, fields.Took(time.Since(start)), zap.Error(err))
@@ -431,7 +431,7 @@ func (r *AggregatorRunner) executeDuty(ctx context.Context, logger *zap.Logger,
 
 	r.measurements.StartPreConsensus()
 	span.AddEvent("broadcasting signed SSV message")
-	if err := r.GetNetwork().Broadcast(msgID, msgToBroadcast); err != nil {
+	if err := r.network.Broadcast(msgID, msgToBroadcast); err != nil {
 		return fmt.Errorf("can't broadcast partial selection proof sig: %w", err)
 	}
 
@@ -457,6 +457,7 @@ func (r *AggregatorRunner) GetShare() *spectypes.Share {
 func (r *AggregatorRunner) GetSigner() ekm.BeaconSigner {
 	return r.signer
 }
+
 func (r *AggregatorRunner) GetOperatorSigner() ssvtypes.OperatorSigner {
 	return r.operatorSigner
 }
diff --git a/protocol/v2/ssv/runner/committee.go b/protocol/v2/ssv/runner/committee.go
index e1f2add43a..fe6bcb7e7d 100644
--- a/protocol/v2/ssv/runner/committee.go
+++ b/protocol/v2/ssv/runner/committee.go
@@ -1030,14 +1030,14 @@ func (r *CommitteeRunner) GetSigner() ekm.BeaconSigner {
 	return r.signer
 }
 
-func (r *CommitteeRunner) GetDoppelgangerHandler() DoppelgangerProvider {
-	return r.doppelgangerHandler
-}
-
 func (r *CommitteeRunner) GetOperatorSigner() ssvtypes.OperatorSigner {
 	return r.operatorSigner
 }
 
+func (r *CommitteeRunner) GetDoppelgangerHandler() DoppelgangerProvider {
+	return r.doppelgangerHandler
+}
+
 func constructAttestationData(vote *spectypes.BeaconVote, duty *spectypes.ValidatorDuty, version spec.DataVersion) *phase0.AttestationData {
 	attData := &phase0.AttestationData{
 		Slot:            duty.Slot,
diff --git a/protocol/v2/ssv/runner/committee_test.go b/protocol/v2/ssv/runner/committee_test.go
index 5316459d1d..cc1298e705 100644
--- a/protocol/v2/ssv/runner/committee_test.go
+++ b/protocol/v2/ssv/runner/committee_test.go
@@ -259,7 +259,7 @@ func TestCommitteeRunnerExecuteDuty_FetchesAttestationDataAndStartsConsensus(t *
 	env := newCommitteeRunnerEnv(t, []int{1}, &committeeDutyGuardStub{}, &doppelgangerStub{})
 	duty := spectestingutils.TestingAttesterDuty(spec.DataVersionElectra)
 
-	env.runner.baseSetupForNewDuty(duty, env.sampleKey.Threshold)
+	env.runner.State = NewRunnerState(env.sampleKey.Threshold, duty)
 
 	require.NoError(t, env.runner.executeDuty(context.Background(), env.logger, duty))
 	require.NotNil(t, env.runner.ValCheck)
diff --git a/protocol/v2/ssv/runner/proposer_test.go b/protocol/v2/ssv/runner/proposer_test.go
index 48490cda38..6bd603e459 100644
--- a/protocol/v2/ssv/runner/proposer_test.go
+++ b/protocol/v2/ssv/runner/proposer_test.go
@@ -431,7 +431,7 @@ func setupRunnerForPostConsensus(
 	t.Helper()
 
 	duty := spectestingutils.TestingProposerDutyV(consensusData.Version)
-	runner.baseSetupForNewDuty(duty, keySet.Threshold)
+	runner.State = NewRunnerState(keySet.Threshold, duty)
 	runner.measurements.StartDutyFlow()
 	runner.measurements.StartConsensus()
 	runner.measurements.EndConsensus()
diff --git a/protocol/v2/ssv/runner/runner.go b/protocol/v2/ssv/runner/runner.go
index e469423fea..54fca1e5d3 100644
--- a/protocol/v2/ssv/runner/runner.go
+++ b/protocol/v2/ssv/runner/runner.go
@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"maps"
 	"slices"
-	"sync"
 
 	"github.com/attestantio/go-eth2-client/spec/phase0"
 	ssz "github.com/ferranbt/fastssz"
@@ -28,17 +27,19 @@ import (
 )
 
 type Getters interface {
+	HasRunningDuty() bool
 	HasRunningQBFTInstance() bool
 	HasAcceptedProposalForCurrentRound() bool
 	GetShares() map[phase0.ValidatorIndex]*spectypes.Share
 	GetRole() spectypes.RunnerRole
+	GetCurrentDutySlot() phase0.Slot
 	GetLastHeight() specqbft.Height
 	GetLastRound() specqbft.Round
 	GetStateRoot() ([32]byte, error)
-	GetBeaconNode() beacon.BeaconNode
 	GetSigner() ekm.BeaconSigner
 	GetOperatorSigner() ssvtypes.OperatorSigner
 	GetNetwork() specqbft.Network
+	GetBeaconNode() beacon.BeaconNode
 }
 
 type Setters interface {
@@ -54,8 +55,6 @@ type Runner interface {
 
 	// StartNewDuty starts a new duty for the runner, returns error if can't
 	StartNewDuty(ctx context.Context, logger *zap.Logger, duty spectypes.Duty, quorum uint64) error
-	// HasRunningDuty returns true if it has a running duty
-	HasRunningDuty() bool
 	// ProcessPreConsensus processes all pre-consensus msgs, returns error if can't process
 	ProcessPreConsensus(ctx context.Context, logger *zap.Logger, signedMsg *spectypes.PartialSignatureMessages) error
 	// ProcessConsensus processes all consensus msgs, returns error if can't process
@@ -81,8 +80,17 @@ type DoppelgangerProvider interface {
 var _ Runner = new(CommitteeRunner)
 
 type BaseRunner struct {
-	mtx            sync.RWMutex
-	State          *State
+	// State stores the current runner state, this state corresponds to 1 particular duty the runner is
+	// currently busy with at the moment. The BaseRunner is not responsible for synchronizing any updates
+	// State might need to record - the caller is responsible to ensure the updates/reads (these can happen
+	// whenever runner's method is called to process a p2p message, or an event) are applied sequentially,
+	// plus the caller is also responsible for ensuring there is no race with moving on to the next duty
+	// (the baseSetupForNewDuty call).
+	// Note, the current implementation achieves concurrent safety by making sure every State read/update
+	// is done by the same go-routine, handling all the messages in queue.SSVMessage (p2p messages and events)
+	// sequentially.
+	State *State
+
 	Share          map[phase0.ValidatorIndex]*spectypes.Share
 	QBFTController *controller.Controller
 	NetworkConfig  *networkconfig.Network
@@ -96,21 +104,22 @@ type BaseRunner struct {
 	highestDecidedSlot phase0.Slot
 }
 
+func (b *BaseRunner) HasRunningDuty() bool {
+	return b.hasDutyRunning()
+}
+
+func (b *BaseRunner) HasStartedQBFTInstance() bool {
+	return b.hasDutyAssigned() && b.State.RunningInstance != nil
+}
+
 func (b *BaseRunner) HasRunningQBFTInstance() bool {
-	var runningInstance *instance.Instance
-	if b.HasRunningDuty() {
-		runningInstance = b.State.RunningInstance
-		if runningInstance != nil {
-			decided, _ := runningInstance.IsDecided()
-			return !decided
-		}
-	}
-	return false
+	// Note: RunningInstance.State cannot be nil for existing RunningInstance by construction.
+	return b.hasDutyRunning() && b.HasStartedQBFTInstance() && !b.State.RunningInstance.State.Decided
 }
 
 func (b *BaseRunner) HasAcceptedProposalForCurrentRound() bool {
 	var runningInstance *instance.Instance
-	if b.HasRunningDuty() {
+	if b.hasDutyRunning() {
 		runningInstance = b.State.RunningInstance
 		if runningInstance != nil {
 			return runningInstance.State.ProposalAcceptedForCurrentRound != nil
@@ -123,21 +132,18 @@ func (b *BaseRunner) GetShares() map[phase0.ValidatorIndex]*spectypes.Share {
 	return b.Share
 }
 
-func (b *BaseRunner) HasRunningDuty() bool {
-	b.mtx.RLock() // reads b.State
-	defer b.mtx.RUnlock()
-
-	if b.State == nil {
-		return false
-	}
-
-	return !b.State.Finished
-}
-
 func (b *BaseRunner) GetRole() spectypes.RunnerRole {
 	return b.RunnerRoleType
 }
 
+func (b *BaseRunner) GetCurrentDutySlot() phase0.Slot {
+	if !b.hasDutyAssigned() {
+		return 0
+	}
+	// State.CurrentDuty cannot be nil for non-nil State by construction.
+	return b.State.CurrentDuty.DutySlot()
+}
+
 func (b *BaseRunner) GetLastHeight() specqbft.Height {
 	if ctrl := b.QBFTController; ctrl != nil {
 		return ctrl.Height
@@ -146,7 +152,7 @@ func (b *BaseRunner) GetLastHeight() specqbft.Height {
 }
 
 func (b *BaseRunner) GetLastRound() specqbft.Round {
-	if b.HasRunningDuty() {
+	if b.hasDutyRunning() {
 		inst := b.State.RunningInstance
 		if inst != nil {
 			return inst.State.Round
@@ -205,32 +211,13 @@ func (b *BaseRunner) MarshalJSON() ([]byte, error) {
 	return byts, err
 }
 
-// SetHighestDecidedSlot set highestDecidedSlot for base runner
-func (b *BaseRunner) SetHighestDecidedSlot(slot phase0.Slot) {
-	b.highestDecidedSlot = slot
-}
-
-// baseSetupForNewDuty is sets the runner for a new duty
-func (b *BaseRunner) baseSetupForNewDuty(duty spectypes.Duty, quorum uint64) {
-	// start new state
-	// start new state
-	// TODO nicer way to get quorum
-	state := NewRunnerState(quorum, duty)
-
-	// TODO: potentially incomplete locking of b.State. runner.Execute(duty) has access to
-	// b.State but currently does not write to it
-	b.mtx.Lock() // writes to b.State
-	b.State = state
-	b.mtx.Unlock()
-}
-
 // baseStartNewDuty is a base func that all runner implementation can call to start a duty
 func (b *BaseRunner) baseStartNewDuty(ctx context.Context, logger *zap.Logger, runner Runner, duty spectypes.Duty, quorum uint64) error {
 	if err := b.ShouldProcessDuty(duty); err != nil {
 		return fmt.Errorf("can't start duty: %w", err)
 	}
 
-	b.baseSetupForNewDuty(duty, quorum)
+	b.State = NewRunnerState(quorum, duty)
 
 	if err := runner.executeDuty(ctx, logger, duty); err != nil {
 		return fmt.Errorf("failed to execute duty: %w", err)
@@ -243,7 +230,7 @@ func (b *BaseRunner) baseStartNewNonBeaconDuty(ctx context.Context, logger *zap.
 	if err := b.ShouldProcessNonBeaconDuty(duty); err != nil {
 		return fmt.Errorf("can't start non-beacon duty: %w", err)
 	}
-	b.baseSetupForNewDuty(duty, quorum)
+	b.State = NewRunnerState(quorum, duty)
 	return runner.executeDuty(ctx, logger, duty)
 }
 
@@ -285,7 +272,7 @@ func (b *BaseRunner) baseConsensusMsgProcessing(ctx context.Context, logger *zap
 	span := trace.SpanFromContext(ctx)
 
 	prevDecided := false
-	if b.HasRunningDuty() && b.State != nil && b.State.RunningInstance != nil {
+	if b.hasDutyRunning() && b.HasStartedQBFTInstance() {
 		prevDecided, _ = b.State.RunningInstance.IsDecided()
 	}
 	if prevDecided {
@@ -300,7 +287,7 @@ func (b *BaseRunner) baseConsensusMsgProcessing(ctx context.Context, logger *zap
 		return false, nil, err
 	}
 
-	if !b.HasRunningDuty() {
+	if !b.hasDutyRunning() {
 		logger.Debug("no running duty, applied consensus message but cannot progress further")
 		return false, nil, nil
 	}
@@ -421,7 +408,7 @@ func (b *BaseRunner) didDecideCorrectly(prevDecided bool, signedMessage *spectyp
 		return false, nil
 	}
 
-	if b.State.RunningInstance == nil {
+	if !b.HasStartedQBFTInstance() {
 		return false, spectypes.NewError(spectypes.DecidedWrongInstanceErrorCode, "decided wrong instance (running instance is nil)")
 	}
 
@@ -483,21 +470,15 @@ func (b *BaseRunner) decide(
 }
 
 func (b *BaseRunner) hasDutyAssigned() bool {
-	b.mtx.RLock() // reads b.State
-	defer b.mtx.RUnlock()
-
 	return b.State != nil
 }
 
-func (b *BaseRunner) hasDutyFinished() bool {
-	b.mtx.RLock() // reads b.State
-	defer b.mtx.RUnlock()
-
-	if b.State == nil {
-		return false
-	}
+func (b *BaseRunner) hasDutyRunning() bool {
+	return b.hasDutyAssigned() && !b.State.Finished
+}
 
-	return b.State.Finished
+func (b *BaseRunner) hasDutyFinished() bool {
+	return b.hasDutyAssigned() && b.State.Finished
 }
 
 func (b *BaseRunner) ShouldProcessDuty(duty spectypes.Duty) error {
@@ -511,8 +492,8 @@ func (b *BaseRunner) ShouldProcessDuty(duty spectypes.Duty) error {
 }
 
 func (b *BaseRunner) ShouldProcessNonBeaconDuty(duty spectypes.Duty) error {
-	// assume CurrentDuty is not nil if state is not nil
-	if b.State != nil && b.State.CurrentDuty.DutySlot() >= duty.DutySlot() {
+	// CurrentDuty is not nil if State is not nil by construction.
+	if b.hasDutyAssigned() && b.State.CurrentDuty.DutySlot() >= duty.DutySlot() {
 		return spectypes.NewError(
 			spectypes.DutyAlreadyPassedErrorCode,
 			fmt.Sprintf("duty for slot %d already passed. Current slot is %d", duty.DutySlot(), b.State.CurrentDuty.DutySlot()),
@@ -522,5 +503,19 @@ func (b *BaseRunner) ShouldProcessNonBeaconDuty(duty spectypes.Duty) error {
 }
 
 func (b *BaseRunner) OnTimeoutQBFT(ctx context.Context, logger *zap.Logger, timeoutData *ssvtypes.TimeoutData) error {
+	if !b.hasDutyRunning() {
+		// Duties terminate eventually, timeout-event issuer is unaware of that - that's why we can end up here.
+		return nil
+	}
+
+	if timeoutData.Height != specqbft.Height(b.GetCurrentDutySlot()) {
+		// Validator-Runners are re-used to process duties targeting different slots (unlike Committee-Runners that
+		// are working with exactly one slot), thus for Validator-Runners timeout events can be delayed in the queue
+		// until the runner has already moved on to a new duty/slot - this is why timeout-event height(== slot)
+		// might be different from the actual current slot the runner is working with, and we just skip these delayed
+		// events as no longer relevant (the duty those are targeting has already expired).
+		return nil
+	}
+
 	return b.QBFTController.OnTimeout(ctx, logger, timeoutData)
 }
diff --git a/protocol/v2/ssv/runner/runner_decode_test.go b/protocol/v2/ssv/runner/runner_decode_test.go
index 15362617b5..d246fd875f 100644
--- a/protocol/v2/ssv/runner/runner_decode_test.go
+++ b/protocol/v2/ssv/runner/runner_decode_test.go
@@ -74,9 +74,9 @@ func TestAggregatorRunnerDecodeIgnoresValCheck(t *testing.T) {
 
 	require.Equal(t, beforeRoot, afterRoot)
 	require.Equal(t, spectypes.RoleAggregator, decoded.GetRole())
-	require.False(t, decoded.HasRunningDuty())
 	require.Len(t, decoded.GetShares(), 1)
 	require.Nil(t, decoded.ValCheck)
+	require.False(t, decoded.hasDutyRunning())
 }
 
 func TestProposerRunnerDecodeIgnoresValCheck(t *testing.T) {
@@ -119,9 +119,9 @@ func TestProposerRunnerDecodeIgnoresValCheck(t *testing.T) {
 
 	require.Equal(t, beforeRoot, afterRoot)
 	require.Equal(t, spectypes.RoleProposer, decoded.GetRole())
-	require.False(t, decoded.HasRunningDuty())
 	require.Len(t, decoded.GetShares(), 1)
 	require.Nil(t, decoded.ValCheck)
+	require.False(t, decoded.hasDutyRunning())
 }
 
 func TestSyncCommitteeAggregatorRunnerDecodeIgnoresValCheck(t *testing.T) {
@@ -160,7 +160,7 @@ func TestSyncCommitteeAggregatorRunnerDecodeIgnoresValCheck(t *testing.T) {
 
 	require.Equal(t, beforeRoot, afterRoot)
 	require.Equal(t, spectypes.RoleSyncCommitteeContribution, decoded.GetRole())
-	require.False(t, decoded.HasRunningDuty())
 	require.Len(t, decoded.GetShares(), 1)
 	require.Nil(t, decoded.ValCheck)
+	require.False(t, decoded.hasDutyRunning())
 }
diff --git a/protocol/v2/ssv/runner/runner_delegator_test.go b/protocol/v2/ssv/runner/runner_delegator_test.go
index 5c7c0b6a18..3239d27302 100644
--- a/protocol/v2/ssv/runner/runner_delegator_test.go
+++ b/protocol/v2/ssv/runner/runner_delegator_test.go
@@ -42,9 +42,9 @@ func TestVoluntaryExitRunnerDecodePreservesEmbeddedBaseRunnerMethods(t *testing.
 
 	require.Equal(t, beforeRoot, afterRoot)
 	require.Equal(t, spectypes.RoleVoluntaryExit, decoded.GetRole())
-	require.False(t, decoded.HasRunningDuty())
 	require.Len(t, decoded.GetShares(), 1)
 	require.Equal(t, share.ValidatorIndex, decoded.GetShare().ValidatorIndex)
+	require.False(t, decoded.hasDutyRunning())
 }
 
 func TestVoluntaryExitRunnerUsesReplacedBaseRunner(t *testing.T) {
@@ -100,6 +100,6 @@ func TestCommitteeRunnerDecodePreservesEmbeddedBaseRunnerMethods(t *testing.T) {
 
 	require.Equal(t, beforeRoot, afterRoot)
 	require.Equal(t, spectypes.RoleCommittee, decoded.GetRole())
-	require.False(t, decoded.HasRunningDuty())
 	require.Len(t, decoded.GetShares(), 1)
+	require.False(t, decoded.hasDutyRunning())
 }
diff --git a/protocol/v2/ssv/runner/runner_state.go b/protocol/v2/ssv/runner/runner_state.go
index dea49b898f..8235583e71 100644
--- a/protocol/v2/ssv/runner/runner_state.go
+++ b/protocol/v2/ssv/runner/runner_state.go
@@ -84,18 +84,15 @@ func (pcs *State) MarshalJSON() ([]byte, error) {
 		Finished:               pcs.Finished,
 	}
 
-	if pcs.CurrentDuty != nil {
-		if ValidatorDuty, ok := pcs.CurrentDuty.(*spectypes.ValidatorDuty); ok {
-			alias.ValidatorDuty = ValidatorDuty
-		} else if committeeDuty, ok := pcs.CurrentDuty.(*spectypes.CommitteeDuty); ok {
-			alias.CommitteeDuty = committeeDuty
-		} else {
-			return nil, errors.New("can't marshal because BaseRunner.State.CurrentDuty isn't ValidatorDuty or CommitteeDuty")
-		}
+	if ValidatorDuty, ok := pcs.CurrentDuty.(*spectypes.ValidatorDuty); ok {
+		alias.ValidatorDuty = ValidatorDuty
+	} else if committeeDuty, ok := pcs.CurrentDuty.(*spectypes.CommitteeDuty); ok {
+		alias.CommitteeDuty = committeeDuty
+	} else {
+		return nil, errors.New("can't marshal because BaseRunner.State.CurrentDuty isn't ValidatorDuty or CommitteeDuty")
 	}
-	byts, err := json.Marshal(alias)
 
-	return byts, err
+	return json.Marshal(alias)
 }
 
 func (pcs *State) UnmarshalJSON(data []byte) error {
diff --git a/protocol/v2/ssv/runner/runner_validations.go b/protocol/v2/ssv/runner/runner_validations.go
index 9968531b75..c54b54267a 100644
--- a/protocol/v2/ssv/runner/runner_validations.go
+++ b/protocol/v2/ssv/runner/runner_validations.go
@@ -90,7 +90,7 @@ func (b *BaseRunner) ValidatePostConsensusMsg(ctx context.Context, runner Runner
 		return err
 	}
 
-	if b.State.RunningInstance == nil {
+	if !b.HasStartedQBFTInstance() {
 		return NewRetryableError(spectypes.WrapError(spectypes.NoRunningConsensusInstanceErrorCode, ErrInstanceNotFound))
 	}
 
diff --git a/protocol/v2/ssv/runner/sync_committee_contribution.go b/protocol/v2/ssv/runner/sync_committee_contribution.go
index c76ed3e1fa..9c57fcfb8e 100644
--- a/protocol/v2/ssv/runner/sync_committee_contribution.go
+++ b/protocol/v2/ssv/runner/sync_committee_contribution.go
@@ -560,6 +560,7 @@ func (r *SyncCommitteeAggregatorRunner) GetShare() *spectypes.Share {
 func (r *SyncCommitteeAggregatorRunner) GetSigner() ekm.BeaconSigner {
 	return r.signer
 }
+
 func (r *SyncCommitteeAggregatorRunner) GetOperatorSigner() ssvtypes.OperatorSigner {
 	return r.operatorSigner
 }
diff --git a/protocol/v2/ssv/runner/validator_registration.go b/protocol/v2/ssv/runner/validator_registration.go
index 23bf45fc3d..70595c6fa0 100644
--- a/protocol/v2/ssv/runner/validator_registration.go
+++ b/protocol/v2/ssv/runner/validator_registration.go
@@ -159,7 +159,7 @@ func (r *ValidatorRegistrationRunner) ProcessPostConsensus(ctx context.Context,
 }
 
 func (r *ValidatorRegistrationRunner) expectedPreConsensusRootsAndDomain() ([]ssz.HashRoot, phase0.DomainType, error) {
-	if r.State == nil || r.State.CurrentDuty == nil {
+	if !r.hasDutyAssigned() {
 		return nil, spectypes.DomainError, fmt.Errorf("no running duty to compute preconsensus roots and domain")
 	}
 	vr, err := r.buildValidatorRegistration(r.State.CurrentDuty.DutySlot())
@@ -279,6 +279,7 @@ func (r *ValidatorRegistrationRunner) GetShare() *spectypes.Share {
 func (r *ValidatorRegistrationRunner) GetSigner() ekm.BeaconSigner {
 	return r.signer
 }
+
 func (r *ValidatorRegistrationRunner) GetOperatorSigner() ssvtypes.OperatorSigner {
 	return r.operatorSigner
 }
diff --git a/protocol/v2/ssv/runner/voluntary_exit.go b/protocol/v2/ssv/runner/voluntary_exit.go
index 48f22fa9ea..486fee49cf 100644
--- a/protocol/v2/ssv/runner/voluntary_exit.go
+++ b/protocol/v2/ssv/runner/voluntary_exit.go
@@ -248,6 +248,7 @@ func (r *VoluntaryExitRunner) GetShare() *spectypes.Share {
 func (r *VoluntaryExitRunner) GetSigner() ekm.BeaconSigner {
 	return r.signer
 }
+
 func (r *VoluntaryExitRunner) GetOperatorSigner() ssvtypes.OperatorSigner {
 	return r.operatorSigner
 }
diff --git a/protocol/v2/ssv/spectest/multi_start_new_runner_duty_type.go b/protocol/v2/ssv/spectest/multi_start_new_runner_duty_type.go
index 956c48a156..92e79292a0 100644
--- a/protocol/v2/ssv/spectest/multi_start_new_runner_duty_type.go
+++ b/protocol/v2/ssv/spectest/multi_start_new_runner_duty_type.go
@@ -96,28 +96,28 @@ func (test *StartNewRunnerDutySpecTest) RunAsPartOfMultiTest(t *testing.T, logge
 		for _, inst := range r.QBFTController.StoredInstances {
 			inst.ValueChecker = protocoltesting.TestingValueChecker{}
 		}
-		if r.State.RunningInstance != nil {
+		if r.HasStartedQBFTInstance() {
 			r.State.RunningInstance.ValueChecker = protocoltesting.TestingValueChecker{}
 		}
 	case *runner.AggregatorRunner:
 		for _, inst := range r.QBFTController.StoredInstances {
 			inst.ValueChecker = protocoltesting.TestingValueChecker{}
 		}
-		if r.State.RunningInstance != nil {
+		if r.HasStartedQBFTInstance() {
 			r.State.RunningInstance.ValueChecker = protocoltesting.TestingValueChecker{}
 		}
 	case *runner.ProposerRunner:
 		for _, inst := range r.QBFTController.StoredInstances {
 			inst.ValueChecker = protocoltesting.TestingValueChecker{}
 		}
-		if r.State.RunningInstance != nil {
+		if r.HasStartedQBFTInstance() {
 			r.State.RunningInstance.ValueChecker = protocoltesting.TestingValueChecker{}
 		}
 	case *runner.SyncCommitteeAggregatorRunner:
 		for _, inst := range r.QBFTController.StoredInstances {
 			inst.ValueChecker = protocoltesting.TestingValueChecker{}
 		}
-		if r.State.RunningInstance != nil {
+		if r.HasStartedQBFTInstance() {
 			r.State.RunningInstance.ValueChecker = protocoltesting.TestingValueChecker{}
 		}
 	}
diff --git a/protocol/v2/ssv/spectest/ssv_mapping_test.go b/protocol/v2/ssv/spectest/ssv_mapping_test.go
index a60e7afb86..810e4be61e 100644
--- a/protocol/v2/ssv/spectest/ssv_mapping_test.go
+++ b/protocol/v2/ssv/spectest/ssv_mapping_test.go
@@ -389,11 +389,9 @@ func fixRunnerForRun(t *testing.T, runnerMap map[string]any, ks *spectestingutil
 
 	if baseRunner.QBFTController != nil {
 		baseRunner.QBFTController = fixControllerForRun(logger, baseRunner.QBFTController, ks)
-		if baseRunner.State != nil {
-			if baseRunner.State.RunningInstance != nil {
-				operator := spectestingutils.TestingCommitteeMember(ks)
-				baseRunner.State.RunningInstance = fixInstanceForRun(logger, ks, baseRunner.State.RunningInstance, baseRunner.QBFTController, operator)
-			}
+		if baseRunner.HasStartedQBFTInstance() {
+			operator := spectestingutils.TestingCommitteeMember(ks)
+			baseRunner.State.RunningInstance = fixInstanceForRun(logger, ks, baseRunner.State.RunningInstance, baseRunner.QBFTController, operator)
 		}
 	}
 
diff --git a/protocol/v2/ssv/spectest/util.go b/protocol/v2/ssv/spectest/util.go
index c3dd50e433..e5039b8580 100644
--- a/protocol/v2/ssv/spectest/util.go
+++ b/protocol/v2/ssv/spectest/util.go
@@ -49,7 +49,7 @@ func runnerForTest(t *testing.T, runnerType runner.Runner, name string, testType
 		for _, inst := range cr.QBFTController.StoredInstances {
 			inst.ValueChecker = valCheck
 		}
-		if cr.State != nil && cr.State.RunningInstance != nil {
+		if cr.HasStartedQBFTInstance() {
 			cr.State.RunningInstance.ValueChecker = valCheck
 		}
 	case *runner.AggregatorRunner:
@@ -60,7 +60,7 @@ func runnerForTest(t *testing.T, runnerType runner.Runner, name string, testType
 		for _, inst := range ar.QBFTController.StoredInstances {
 			inst.ValueChecker = valCheck
 		}
-		if ar.State != nil && ar.State.RunningInstance != nil {
+		if ar.HasStartedQBFTInstance() {
 			ar.State.RunningInstance.ValueChecker = valCheck
 		}
 	case *runner.ProposerRunner:
@@ -71,7 +71,7 @@ func runnerForTest(t *testing.T, runnerType runner.Runner, name string, testType
 		for _, inst := range pr.QBFTController.StoredInstances {
 			inst.ValueChecker = valCheck
 		}
-		if pr.State != nil && pr.State.RunningInstance != nil {
+		if pr.HasStartedQBFTInstance() {
 			pr.State.RunningInstance.ValueChecker = valCheck
 		}
 	case *runner.SyncCommitteeAggregatorRunner:
@@ -82,7 +82,7 @@ func runnerForTest(t *testing.T, runnerType runner.Runner, name string, testType
 		for _, inst := range scr.QBFTController.StoredInstances {
 			inst.ValueChecker = valCheck
 		}
-		if scr.State != nil && scr.State.RunningInstance != nil {
+		if scr.HasStartedQBFTInstance() {
 			scr.State.RunningInstance.ValueChecker = valCheck
 		}
 	case *runner.ValidatorRegistrationRunner:
@@ -102,7 +102,7 @@ func normalizeExpectedProposerStartValues(pr *runner.ProposerRunner) {
 	}
 	if state := pr.State; state != nil {
 		state.DecidedValue = normalizeProposerConsensusValue(state.DecidedValue)
-		if state.RunningInstance != nil {
+		if pr.HasStartedQBFTInstance() {
 			state.RunningInstance.StartValue = normalizeProposerConsensusValue(state.RunningInstance.StartValue)
 			if state.RunningInstance.State != nil {
 				state.RunningInstance.State.LastPreparedValue = normalizeProposerConsensusValue(state.RunningInstance.State.LastPreparedValue)
diff --git a/protocol/v2/ssv/validator/committee.go b/protocol/v2/ssv/validator/committee.go
index d0977279ca..a673ca423a 100644
--- a/protocol/v2/ssv/validator/committee.go
+++ b/protocol/v2/ssv/validator/committee.go
@@ -35,7 +35,7 @@ type Committee struct {
 
 	// mtx syncs access to Queues, Runners, Shares.
 	mtx     sync.RWMutex
-	Queues  map[phase0.Slot]queueContainer
+	Queues  map[phase0.Slot]queue.Queue
 	Runners map[phase0.Slot]*runner.CommitteeRunner
 	Shares  map[phase0.ValidatorIndex]*spectypes.Share
 
@@ -65,7 +65,7 @@ func NewCommittee(
 	return &Committee{
 		logger:          logger,
 		networkConfig:   networkConfig,
-		Queues:          make(map[phase0.Slot]queueContainer),
+		Queues:          make(map[phase0.Slot]queue.Queue),
 		Runners:         make(map[phase0.Slot]*runner.CommitteeRunner),
 		Shares:          shares,
 		CommitteeMember: operator,
@@ -92,7 +92,7 @@ func (c *Committee) RemoveShare(validatorIndex phase0.ValidatorIndex) {
 // StartDuty starts a new duty for the given slot.
 func (c *Committee) StartDuty(ctx context.Context, logger *zap.Logger, duty *spectypes.CommitteeDuty) (
 	*runner.CommitteeRunner,
-	queueContainer,
+	queue.Queue,
 	error,
 ) {
 	ctx, span := tracer.Start(ctx,
@@ -106,13 +106,13 @@ func (c *Committee) StartDuty(ctx context.Context, logger *zap.Logger, duty *spe
 	span.AddEvent("prepare duty and runner")
 	r, q, runnableDuty, err := c.prepareDutyAndRunner(ctx, logger, duty)
 	if err != nil {
-		return nil, queueContainer{}, traces.Errorf(span, "prepare duty and runner: %w", err)
+		return nil, nil, traces.Errorf(span, "prepare duty and runner: %w", err)
 	}
 
 	logger.Info("ℹ️ starting duty processing")
 	err = r.StartNewDuty(ctx, logger, runnableDuty, c.CommitteeMember.GetQuorum())
 	if err != nil {
-		return nil, queueContainer{}, traces.Errorf(span, "runner failed to start duty: %w", err)
+		return nil, nil, traces.Errorf(span, "runner failed to start duty: %w", err)
 	}
 
 	span.SetStatus(codes.Ok, "")
@@ -121,7 +121,7 @@ func (c *Committee) StartDuty(ctx context.Context, logger *zap.Logger, duty *spe
 
 func (c *Committee) prepareDutyAndRunner(ctx context.Context, logger *zap.Logger, duty *spectypes.CommitteeDuty) (
 	r *runner.CommitteeRunner,
-	q queueContainer,
+	q queue.Queue,
 	runnableDuty *spectypes.CommitteeDuty,
 	err error,
 ) {
@@ -137,18 +137,18 @@ func (c *Committee) prepareDutyAndRunner(ctx context.Context, logger *zap.Logger
 	defer c.mtx.Unlock()
 
 	if _, exists := c.Runners[duty.Slot]; exists {
-		return nil, queueContainer{}, nil, traces.Errorf(span, "CommitteeRunner for slot %d already exists", duty.Slot)
+		return nil, nil, nil, traces.Errorf(span, "CommitteeRunner for slot %d already exists", duty.Slot)
 	}
 
 	shares, attesters, runnableDuty, err := c.prepareDuty(logger, duty)
 	if err != nil {
-		return nil, queueContainer{}, nil, traces.Error(span, err)
+		return nil, nil, nil, traces.Error(span, err)
 	}
 
 	// Create the corresponding runner.
 	r, err = c.CreateRunnerFn(duty.Slot, shares, attesters, c.dutyGuard)
 	if err != nil {
-		return nil, queueContainer{}, nil, traces.Errorf(span, "could not create CommitteeRunner: %w", err)
+		return nil, nil, nil, traces.Errorf(span, "could not create CommitteeRunner: %w", err)
 	}
 	r.SetTimeoutFunc(c.onTimeout)
 	c.Runners[duty.Slot] = r
@@ -165,26 +165,18 @@ func (c *Committee) prepareDutyAndRunner(ctx context.Context, logger *zap.Logger
 
 // getQueue returns queue for the provided slot, lazily initializing it if it didn't exist previously.
 // MUST be called with c.mtx locked!
-func (c *Committee) getQueue(logger *zap.Logger, slot phase0.Slot) queueContainer {
+func (c *Committee) getQueue(logger *zap.Logger, slot phase0.Slot) queue.Queue {
 	q, exists := c.Queues[slot]
 	if !exists {
-		q = queueContainer{
-			Q: queue.New(
-				logger,
-				1000,
-				queue.WithInboxSizeMetric(
-					queue.InboxSizeMetric,
-					queue.CommitteeQueueMetricType,
-					queue.CommitteeMetricID(slot),
-				),
+		q = queue.New(
+			logger,
+			1000,
+			queue.WithInboxSizeMetric(
+				queue.InboxSizeMetric,
+				queue.CommitteeQueueMetricType,
+				queue.CommitteeMetricID(slot),
 			),
-			queueState: &queue.State{
-				HasRunningInstance: false,
-				Height:             specqbft.Height(slot),
-				Slot:               slot,
-				Quorum:             c.CommitteeMember.GetQuorum(),
-			},
-		}
+		)
 		c.Queues[slot] = q
 	}
 
@@ -325,7 +317,9 @@ func (c *Committee) ProcessMessage(ctx context.Context, logger *zap.Logger, msg
 			dutyRunner, found := c.Runners[slot]
 			c.mtx.RUnlock()
 			if !found {
-				return fmt.Errorf("event message: no committee runner found for slot %d", slot)
+				// Old runners are pruned, timeout-event issuer is unaware of that - that's why we can end up here
+				logger.Debug("event message: timeout event arrived, but targeted runner not found (likely was pruned)")
+				return nil
 			}
 
 			timeoutData, err := eventMsg.GetTimeoutData()
diff --git a/protocol/v2/ssv/validator/committee_queue.go b/protocol/v2/ssv/validator/committee_queue.go
index 757c8754ed..30e1ed9ffb 100644
--- a/protocol/v2/ssv/validator/committee_queue.go
+++ b/protocol/v2/ssv/validator/committee_queue.go
@@ -23,12 +23,6 @@ import (
 	"github.com/ssvlabs/ssv/protocol/v2/types"
 )
 
-// queueContainer wraps a queue with its corresponding state
-type queueContainer struct {
-	Q          queue.Queue
-	queueState *queue.State
-}
-
 // EnqueueMessage enqueues a spectypes.SSVMessage for processing.
 // TODO: accept DecodedSSVMessage once p2p is upgraded to decode messages during validation.
 func (c *Committee) EnqueueMessage(ctx context.Context, msg *queue.SSVMessage) {
@@ -67,7 +61,7 @@ func (c *Committee) EnqueueMessage(ctx context.Context, msg *queue.SSVMessage) {
 	c.mtx.Unlock()
 
 	span.AddEvent("pushing message to the queue")
-	if pushed := q.Q.TryPush(msg); !pushed {
+	if pushed := q.TryPush(msg); !pushed {
 		const errMsg = "❗ dropping message because the queue is full"
 		logger.Warn(errMsg)
 		span.SetStatus(codes.Error, errMsg)
@@ -82,16 +76,13 @@ func (c *Committee) EnqueueMessage(ctx context.Context, msg *queue.SSVMessage) {
 func (c *Committee) ConsumeQueue(
 	ctx context.Context,
 	logger *zap.Logger,
-	q queueContainer,
+	q queue.Queue,
 	handler MessageHandler, // should be c.ProcessMessage, it is a param so can be mocked out for testing
-	rnr *runner.CommitteeRunner,
+	r *runner.CommitteeRunner,
 ) {
 	logger.Debug("📬 queue consumer is running")
 	defer logger.Debug("📪 queue consumer is closed")
 
-	// Construct a representation of the current state.
-	state := *q.queueState
-
 	// msgStates keeps track of in-flight processing state (retry count + span context) per message.
 	// Since this map grows over time, we need to clean it up automatically. There is no specific TTL value
 	// to use for its entries - it just needs to be large enough to prevent unnecessary (but non-harmful)
@@ -102,11 +93,21 @@ func (c *Committee) ConsumeQueue(
 	go msgStates.Start()
 	defer msgStates.Stop()
 
+	// rState defines current runner state that will be used for deciding which messages we want to process
+	// sooner (vs which ones can wait till later).
+	rState := queue.State{
+		Quorum: c.CommitteeMember.GetQuorum(), // never changes for duty runner
+		Slot:   r.GetCurrentDutySlot(),
+	}
+
 	for ctx.Err() == nil {
-		state.HasRunningInstance = rnr.HasRunningQBFTInstance()
+		// Update rState to incorporate the effects previously handled message might have had on the runner state.
+		rState.HasRunningInstance = r.HasRunningQBFTInstance()
+		rState.Height = r.GetLastHeight()
+		rState.Round = r.GetLastRound()
 
 		filter := queue.FilterAny
-		if state.HasRunningInstance && !rnr.HasAcceptedProposalForCurrentRound() {
+		if rState.HasRunningInstance && !r.HasAcceptedProposalForCurrentRound() {
 			// If no proposal was accepted for the current round, skip prepare & commit messages
 			// for the current round.
 			filter = func(m *queue.SSVMessage) bool {
@@ -115,13 +116,13 @@ func (c *Committee) ConsumeQueue(
 					return m.MsgType != spectypes.SSVPartialSignatureMsgType
 				}
 
-				if sm.Round != state.Round { // allow next round or change round messages.
+				if sm.Round != rState.Round { // allow next round or change round messages.
 					return true
 				}
 
 				return sm.MsgType != specqbft.PrepareMsgType && sm.MsgType != specqbft.CommitMsgType
 			}
-		} else if state.HasRunningInstance {
+		} else if rState.HasRunningInstance {
 			filter = func(ssvMessage *queue.SSVMessage) bool {
 				// don't read post consensus until decided
 				return ssvMessage.MsgType != spectypes.SSVPartialSignatureMsgType
@@ -129,8 +130,7 @@ func (c *Committee) ConsumeQueue(
 		}
 
 		// Pop the highest priority message for the current state.
-		// TODO: (Alan) bring back filter
-		msg := q.Q.Pop(ctx, queue.NewCommitteeQueuePrioritizer(&state), filter)
+		msg := q.Pop(ctx, queue.NewCommitteeQueuePrioritizer(&rState), filter)
 		if ctx.Err() != nil {
 			// Optimization: terminate fast if we can.
 			return
@@ -243,7 +243,7 @@ func (c *Committee) ConsumeQueue(
 					case <-msgState.ctx.Done():
 						return
 					}
-					if pushed := q.Q.TryPush(msg); !pushed {
+					if pushed := q.TryPush(msg); !pushed {
 						const droppingMsgDueToQueueIsFullEvent = "❗ not gonna replay message because the queue is full"
 						msgLogger.Error(droppingMsgDueToQueueIsFullEvent)
 						msgState.span.AddEvent(droppingMsgDueToQueueIsFullEvent, trace.WithAttributes(
diff --git a/protocol/v2/ssv/validator/committee_queue_test.go b/protocol/v2/ssv/validator/committee_queue_test.go
index aed796bf8f..95be2e033a 100644
--- a/protocol/v2/ssv/validator/committee_queue_test.go
+++ b/protocol/v2/ssv/validator/committee_queue_test.go
@@ -22,6 +22,7 @@ import (
 	"github.com/ssvlabs/ssv/networkconfig"
 	"github.com/ssvlabs/ssv/observability/log"
 	"github.com/ssvlabs/ssv/protocol/v2/message"
+	"github.com/ssvlabs/ssv/protocol/v2/qbft/controller"
 	"github.com/ssvlabs/ssv/protocol/v2/qbft/instance"
 	"github.com/ssvlabs/ssv/protocol/v2/ssv/queue"
 	"github.com/ssvlabs/ssv/protocol/v2/ssv/runner"
@@ -80,7 +81,7 @@ func runConsumeQueueAsync(
 	t *testing.T,
 	ctx context.Context,
 	committee *Committee,
-	q queueContainer,
+	q queue.Queue,
 	logger *zap.Logger,
 	handler MessageHandler,
 	committeeRunner *runner.CommitteeRunner,
@@ -146,6 +147,43 @@ func setupMessageCollection(capacity int) (chan *queue.SSVMessage, MessageHandle
 	return msgChannel, handler
 }
 
+func newCommitteeQueueStateForTest(slot phase0.Slot, round specqbft.Round, hasRunningInstance bool, quorum uint64) *queue.State {
+	return &queue.State{
+		HasRunningInstance: hasRunningInstance,
+		Height:             specqbft.Height(slot),
+		Slot:               slot,
+		Round:              round,
+		Quorum:             quorum,
+	}
+}
+
+func newCommitteeRunnerForTest(
+	slot phase0.Slot,
+	round specqbft.Round,
+	decided bool,
+	proposal *specqbft.ProcessingMessage,
+) *runner.CommitteeRunner {
+	return &runner.CommitteeRunner{
+		BaseRunner: &runner.BaseRunner{
+			QBFTController: &controller.Controller{
+				Height: specqbft.Height(slot),
+			},
+			State: &runner.State{
+				RunningInstance: &instance.Instance{
+					State: &specqbft.State{
+						Decided:                         decided,
+						ProposalAcceptedForCurrentRound: proposal,
+						Round:                           round,
+					},
+				},
+				CurrentDuty: &spectypes.CommitteeDuty{
+					Slot: slot,
+				},
+			},
+		},
+	}
+}
+
 // TestHandleMessageCreatesQueue verifies that the HandleMessage method correctly
 // initializes a new queue when receiving a message for a slot that doesn't have
 // an associated queue yet.
@@ -170,7 +208,7 @@ func TestHandleMessageCreatesQueue(t *testing.T) {
 	committee := &Committee{
 		logger:          logger,
 		networkConfig:   networkconfig.TestNetwork,
-		Queues:          make(map[phase0.Slot]queueContainer),
+		Queues:          make(map[phase0.Slot]queue.Queue),
 		Runners:         make(map[phase0.Slot]*runner.CommitteeRunner),
 		CommitteeMember: &spectypes.CommitteeMember{},
 	}
@@ -189,13 +227,18 @@ func TestHandleMessageCreatesQueue(t *testing.T) {
 
 	require.True(t, ok)
 
-	assert.NotNil(t, q.Q)
-	assert.Equal(t, slot, q.queueState.Slot)
-	assert.False(t, q.queueState.HasRunningInstance)
-	assert.Equal(t, specqbft.Height(slot), q.queueState.Height)
+	assert.NotNil(t, q)
+	assert.Equal(t, 1, q.Len())
 
-	// default, the queueState.Round is not explicitly initialized from the incoming message
-	assert.Equal(t, specqbft.Round(0), q.queueState.Round)
+	queuedMsg := q.TryPop(
+		queue.NewCommitteeQueuePrioritizer(
+			newCommitteeQueueStateForTest(slot, 0, false, committee.CommitteeMember.GetQuorum()),
+		),
+		queue.FilterAny,
+	)
+	require.NotNil(t, queuedMsg)
+	assert.Equal(t, testMsg.MsgID, queuedMsg.MsgID)
+	assert.Equal(t, testMsg.MsgType, queuedMsg.MsgType)
 }
 
 // TestConsumeQueueBasic tests the fundamental queue consumption functionality
@@ -222,7 +265,7 @@ func TestConsumeQueueBasic(t *testing.T) {
 	committee := &Committee{
 		logger:          logger,
 		networkConfig:   networkconfig.TestNetwork,
-		Queues:          make(map[phase0.Slot]queueContainer),
+		Queues:          make(map[phase0.Slot]queue.Queue),
 		Runners:         make(map[phase0.Slot]*runner.CommitteeRunner),
 		CommitteeMember: &spectypes.CommitteeMember{},
 	}
@@ -245,35 +288,15 @@ func TestConsumeQueueBasic(t *testing.T) {
 	}
 	testMsg2 := makeTestSSVMessage(t, spectypes.SSVConsensusMsgType, msgID2, qbftMsg2)
 
-	q := queueContainer{
-		Q: queue.New(logger, 1000),
-		queueState: &queue.State{
-			HasRunningInstance: true,
-			Height:             specqbft.Height(slot),
-			Slot:               slot,
-			Round:              1,
-		},
-	}
-	q.Q.TryPush(testMsg1)
-	q.Q.TryPush(testMsg2)
+	q := queue.New(logger, 1000)
+	q.TryPush(testMsg1)
+	q.TryPush(testMsg2)
 
 	proposalMsg := &specqbft.ProcessingMessage{
 		QBFTMessage: qbftMsg1,
 	}
 
-	committeeRunner := &runner.CommitteeRunner{
-		BaseRunner: &runner.BaseRunner{
-			State: &runner.State{
-				RunningInstance: &instance.Instance{
-					State: &specqbft.State{
-						Decided:                         false,
-						ProposalAcceptedForCurrentRound: proposalMsg,
-						Round:                           1,
-					},
-				},
-			},
-		},
-	}
+	committeeRunner := newCommitteeRunnerForTest(slot, 1, false, proposalMsg)
 
 	msgChannel, handler := setupMessageCollection(2)
 	runConsumeQueueAsync(t, ctx, committee, q, logger, handler, committeeRunner)
@@ -307,7 +330,7 @@ func TestFilterNoProposalAccepted(t *testing.T) {
 
 	committee := &Committee{
 		networkConfig:   networkconfig.TestNetwork,
-		Queues:          make(map[phase0.Slot]queueContainer),
+		Queues:          make(map[phase0.Slot]queue.Queue),
 		Runners:         make(map[phase0.Slot]*runner.CommitteeRunner),
 		CommitteeMember: &spectypes.CommitteeMember{},
 	}
@@ -351,34 +374,14 @@ func TestFilterNoProposalAccepted(t *testing.T) {
 		combinedMessages[i], combinedMessages[j] = combinedMessages[j], combinedMessages[i]
 	})
 
-	q := queueContainer{
-		Q: queue.New(logger, 1000),
-		queueState: &queue.State{
-			HasRunningInstance: true,
-			Height:             specqbft.Height(slot),
-			Slot:               slot,
-			Round:              currentRound,
-		},
-	}
+	q := queue.New(logger, 1000)
 
 	for _, combined := range combinedMessages {
 		testMsg := makeTestSSVMessage(t, spectypes.SSVConsensusMsgType, combined.ID, combined.Message)
-		q.Q.TryPush(testMsg)
+		q.TryPush(testMsg)
 	}
 
-	committeeRunner := &runner.CommitteeRunner{
-		BaseRunner: &runner.BaseRunner{
-			State: &runner.State{
-				RunningInstance: &instance.Instance{
-					State: &specqbft.State{
-						Decided:                         false,
-						ProposalAcceptedForCurrentRound: nil,
-						Round:                           currentRound,
-					},
-				},
-			},
-		},
-	}
+	committeeRunner := newCommitteeRunnerForTest(slot, currentRound, false, nil)
 
 	msgChannel, handler := setupMessageCollection(4)
 	runConsumeQueueAsync(t, ctx, committee, q, logger, handler, committeeRunner)
@@ -427,7 +430,7 @@ func TestFilterNotDecidedSkipsPartialSignatures(t *testing.T) {
 
 	committee := &Committee{
 		networkConfig:   networkconfig.TestNetwork,
-		Queues:          make(map[phase0.Slot]queueContainer),
+		Queues:          make(map[phase0.Slot]queue.Queue),
 		Runners:         make(map[phase0.Slot]*runner.CommitteeRunner),
 		CommitteeMember: &spectypes.CommitteeMember{},
 	}
@@ -458,36 +461,16 @@ func TestFilterNotDecidedSkipsPartialSignatures(t *testing.T) {
 	testMsg1 := makeTestSSVMessage(t, spectypes.SSVConsensusMsgType, msgID1, qbftMsg)
 	testMsg2 := makeTestSSVMessage(t, spectypes.SSVPartialSignatureMsgType, msgID2, partialSigMsg)
 
-	q := queueContainer{
-		Q: queue.New(logger, 1000),
-		queueState: &queue.State{
-			HasRunningInstance: true,
-			Height:             specqbft.Height(slot),
-			Slot:               slot,
-			Round:              1,
-		},
-	}
+	q := queue.New(logger, 1000)
 
-	q.Q.TryPush(testMsg1)
-	q.Q.TryPush(testMsg2)
+	q.TryPush(testMsg1)
+	q.TryPush(testMsg2)
 
 	proposalMsg := &specqbft.ProcessingMessage{
 		QBFTMessage: qbftMsg,
 	}
 
-	committeeRunner := &runner.CommitteeRunner{
-		BaseRunner: &runner.BaseRunner{
-			State: &runner.State{
-				RunningInstance: &instance.Instance{
-					State: &specqbft.State{
-						Decided:                         false,
-						ProposalAcceptedForCurrentRound: proposalMsg,
-						Round:                           1,
-					},
-				},
-			},
-		},
-	}
+	committeeRunner := newCommitteeRunnerForTest(slot, 1, false, proposalMsg)
 
 	msgChannel, handler := setupMessageCollection(2)
 	runConsumeQueueAsync(t, ctx, committee, q, logger, handler, committeeRunner)
@@ -507,7 +490,7 @@ func TestFilterDecidedAllowsAll(t *testing.T) {
 
 	committee := &Committee{
 		networkConfig:   networkconfig.TestNetwork,
-		Queues:          make(map[phase0.Slot]queueContainer),
+		Queues:          make(map[phase0.Slot]queue.Queue),
 		Runners:         make(map[phase0.Slot]*runner.CommitteeRunner),
 		CommitteeMember: &spectypes.CommitteeMember{},
 	}
@@ -538,36 +521,16 @@ func TestFilterDecidedAllowsAll(t *testing.T) {
 	testMsg1 := makeTestSSVMessage(t, spectypes.SSVConsensusMsgType, msgID1, qbftMsg)
 	testMsg2 := makeTestSSVMessage(t, spectypes.SSVPartialSignatureMsgType, msgID2, partialSigMsg)
 
-	q := queueContainer{
-		Q: queue.New(logger, 1000),
-		queueState: &queue.State{
-			HasRunningInstance: true,
-			Height:             specqbft.Height(slot),
-			Slot:               slot,
-			Round:              1,
-		},
-	}
+	q := queue.New(logger, 1000)
 
-	q.Q.TryPush(testMsg1)
-	q.Q.TryPush(testMsg2)
+	q.TryPush(testMsg1)
+	q.TryPush(testMsg2)
 
 	proposalMsg := &specqbft.ProcessingMessage{
 		QBFTMessage: qbftMsg,
 	}
 
-	committeeRunner := &runner.CommitteeRunner{
-		BaseRunner: &runner.BaseRunner{
-			State: &runner.State{
-				RunningInstance: &instance.Instance{
-					State: &specqbft.State{
-						Decided:                         true,
-						ProposalAcceptedForCurrentRound: proposalMsg,
-						Round:                           1,
-					},
-				},
-			},
-		},
-	}
+	committeeRunner := newCommitteeRunnerForTest(slot, 1, true, proposalMsg)
 
 	msgChannel, handler := setupMessageCollection(2)
 	runConsumeQueueAsync(t, ctx, committee, q, logger, handler, committeeRunner)
@@ -619,16 +582,8 @@ func TestChangingFilterState(t *testing.T) {
 			return fmt.Errorf("intentionally stopping ConsumeQueue after first message")
 		}
 
-		q := queueContainer{
-			Q: queue.New(logger, 1),
-			queueState: &queue.State{
-				HasRunningInstance: true,
-				Height:             specqbft.Height(slot),
-				Slot:               slot,
-				Round:              round,
-			},
-		}
-		q.Q.TryPush(prepareMsg)
+		q := queue.New(logger, 1)
+		q.TryPush(prepareMsg)
 
 		c := &Committee{
 			networkConfig:   networkconfig.TestNetwork,
@@ -639,36 +594,12 @@ func TestChangingFilterState(t *testing.T) {
 	}
 
 	// 1) No proposal accepted => Prepare should be filtered out
-	r1 := &runner.CommitteeRunner{
-		BaseRunner: &runner.BaseRunner{
-			State: &runner.State{
-				RunningInstance: &instance.Instance{
-					State: &specqbft.State{
-						Decided:                         false,
-						ProposalAcceptedForCurrentRound: nil,
-						Round:                           round,
-					},
-				},
-			},
-		},
-	}
+	r1 := newCommitteeRunnerForTest(slot, round, false, nil)
 	seen1 := runOnce(r1)
 	assert.Nil(t, seen1)
 
 	// 2) Proposal accepted => now we should see exactly one Prepare
-	r2 := &runner.CommitteeRunner{
-		BaseRunner: &runner.BaseRunner{
-			State: &runner.State{
-				RunningInstance: &instance.Instance{
-					State: &specqbft.State{
-						Decided:                         false,
-						ProposalAcceptedForCurrentRound: &specqbft.ProcessingMessage{QBFTMessage: prepareBody},
-						Round:                           round,
-					},
-				},
-			},
-		},
-	}
+	r2 := newCommitteeRunnerForTest(slot, round, false, &specqbft.ProcessingMessage{QBFTMessage: prepareBody})
 	seen2 := runOnce(r2)
 
 	require.NotNil(t, seen2)
@@ -740,22 +671,14 @@ func TestCommitteeQueueFilteringScenarios(t *testing.T) {
 
 			committee := &Committee{
 				networkConfig:   networkconfig.TestNetwork,
-				Queues:          make(map[phase0.Slot]queueContainer),
+				Queues:          make(map[phase0.Slot]queue.Queue),
 				Runners:         make(map[phase0.Slot]*runner.CommitteeRunner),
 				CommitteeMember: &spectypes.CommitteeMember{},
 			}
 
 			slot := phase0.Slot(123)
 
-			q := queueContainer{
-				Q: queue.New(logger, 10),
-				queueState: &queue.State{
-					HasRunningInstance: tc.hasRunningDuty,
-					Height:             specqbft.Height(slot),
-					Slot:               slot,
-					Round:              1,
-				},
-			}
+			q := queue.New(logger, 10)
 
 			var proposalMsg *specqbft.ProcessingMessage
 			if tc.proposalAccepted {
@@ -764,23 +687,11 @@ func TestCommitteeQueueFilteringScenarios(t *testing.T) {
 				}
 			}
 
-			committeeRunner := &runner.CommitteeRunner{
-				BaseRunner: &runner.BaseRunner{
-					State: &runner.State{
-						RunningInstance: &instance.Instance{
-							State: &specqbft.State{
-								Decided:                         tc.decided,
-								ProposalAcceptedForCurrentRound: proposalMsg,
-								Round:                           1,
-							},
-						},
-					},
-				},
-			}
+			committeeRunner := newCommitteeRunnerForTest(slot, 1, tc.decided, proposalMsg)
 
 			// Set runner state based on hasRunningDuty parameter
 			if !tc.hasRunningDuty {
-				committeeRunner.State.Finished = true // This makes HasRunningDuty() return false
+				committeeRunner.State.Finished = true // This makes hasDutyRunning() return false
 			}
 
 			msgChannel := make(chan *queue.SSVMessage, len(tc.messagesTypes))
@@ -799,7 +710,7 @@ func TestCommitteeQueueFilteringScenarios(t *testing.T) {
 					MsgType: msgType,
 				}
 				testMsg := makeTestSSVMessage(t, spectypes.SSVConsensusMsgType, msgID, qbftMsg)
-				pushed := q.Q.TryPush(testMsg)
+				pushed := q.TryPush(testMsg)
 				require.True(t, pushed)
 			}
 
@@ -902,36 +813,16 @@ func TestFilterPartialSignatureMessages(t *testing.T) {
 
 			committee := &Committee{
 				networkConfig:   networkconfig.TestNetwork,
-				Queues:          make(map[phase0.Slot]queueContainer),
+				Queues:          make(map[phase0.Slot]queue.Queue),
 				Runners:         make(map[phase0.Slot]*runner.CommitteeRunner),
 				CommitteeMember: &spectypes.CommitteeMember{},
 			}
 
 			slot := phase0.Slot(123)
 
-			q := queueContainer{
-				Q: queue.New(logger, 10),
-				queueState: &queue.State{
-					HasRunningInstance: true,
-					Height:             specqbft.Height(slot),
-					Slot:               slot,
-					Round:              1,
-				},
-			}
+			q := queue.New(logger, 10)
 
-			committeeRunner := &runner.CommitteeRunner{
-				BaseRunner: &runner.BaseRunner{
-					State: &runner.State{
-						RunningInstance: &instance.Instance{
-							State: &specqbft.State{
-								Decided:                         tc.decided,
-								ProposalAcceptedForCurrentRound: &specqbft.ProcessingMessage{},
-								Round:                           1,
-							},
-						},
-					},
-				},
-			}
+			committeeRunner := newCommitteeRunnerForTest(slot, 1, tc.decided, &specqbft.ProcessingMessage{})
 
 			msgID := spectypes.MessageID{0x10}
 			partialSigMsg := &spectypes.PartialSignatureMessages{
@@ -947,7 +838,7 @@ func TestFilterPartialSignatureMessages(t *testing.T) {
 			}
 			testMsg := makeTestSSVMessage(t, spectypes.SSVPartialSignatureMsgType, msgID, partialSigMsg)
 
-			pushed := q.Q.TryPush(testMsg)
+			pushed := q.TryPush(testMsg)
 			require.True(t, pushed)
 
 			if tc.shouldBeFiltered {
@@ -989,7 +880,7 @@ func TestConsumeQueuePrioritization(t *testing.T) {
 
 	committee := &Committee{
 		networkConfig:   networkconfig.TestNetwork,
-		Queues:          make(map[phase0.Slot]queueContainer),
+		Queues:          make(map[phase0.Slot]queue.Queue),
 		Runners:         make(map[phase0.Slot]*runner.CommitteeRunner),
 		CommitteeMember: &spectypes.CommitteeMember{},
 	}
@@ -1018,30 +909,14 @@ func TestConsumeQueuePrioritization(t *testing.T) {
 		makeTestSSVMessage(t, message.SSVEventMsgType, spectypes.MessageID{5}, eventMsgBody),
 	}
 
-	q := queueContainer{
-		Q: queue.New(logger, 10),
-		queueState: &queue.State{
-			HasRunningInstance: true,
-			Height:             specqbft.Height(slot),
-			Slot:               slot,
-			Round:              currentRound,
-		},
-	}
+	q := queue.New(logger, 10)
 	for _, msg := range testMessages {
-		q.Q.TryPush(msg)
+		q.TryPush(msg)
 	}
 
 	// Runner with a proposal already accepted, not yet decided
 	acceptedProposal := &specqbft.ProcessingMessage{QBFTMessage: proposalMsgBody}
-	committeeRunner := &runner.CommitteeRunner{
-		BaseRunner: &runner.BaseRunner{
-			State: &runner.State{RunningInstance: &instance.Instance{State: &specqbft.State{
-				Decided:                         false,
-				ProposalAcceptedForCurrentRound: acceptedProposal,
-				Round:                           currentRound,
-			}}},
-		},
-	}
+	committeeRunner := newCommitteeRunnerForTest(slot, currentRound, false, acceptedProposal)
 
 	msgChannel := make(chan *queue.SSVMessage, len(testMessages))
 
@@ -1116,19 +991,13 @@ func TestHandleMessageQueueFullAndDropping(t *testing.T) {
 	committee := &Committee{
 		logger:          logger,
 		networkConfig:   networkconfig.TestNetwork,
-		Queues:          make(map[phase0.Slot]queueContainer),
+		Queues:          make(map[phase0.Slot]queue.Queue),
 		CommitteeMember: &spectypes.CommitteeMember{},
 	}
 
 	// Step 0: Create the queue container with the desired small capacity and add it to the committee
-	qContainer := queueContainer{
-		Q: queue.New(logger, queueCapacity),
-		queueState: &queue.State{
-			HasRunningInstance: false,
-			Height:             specqbft.Height(slot),
-			Slot:               slot,
-		},
-	}
+	qContainer := queue.New(logger, queueCapacity)
+	qState := newCommitteeQueueStateForTest(slot, 0, false, committee.CommitteeMember.GetQuorum())
 	committee.Queues[slot] = qContainer
 
 	// Step 1: Fill the pre-made queue to its capacity by calling HandleMessage
@@ -1142,7 +1011,7 @@ func TestHandleMessageQueueFullAndDropping(t *testing.T) {
 		committee.EnqueueMessage(ctx, testMsg)
 	}
 
-	require.Equal(t, queueCapacity, qContainer.Q.Len())
+	require.Equal(t, queueCapacity, qContainer.Len())
 
 	// Step 2: Clear log buffer and attempt to push one more message (this one should be dropped)
 	droppedMsgID := msgIDBase
@@ -1152,7 +1021,7 @@ func TestHandleMessageQueueFullAndDropping(t *testing.T) {
 
 	committee.EnqueueMessage(ctx, testMsgDrop)
 
-	assert.Equal(t, queueCapacity, qContainer.Q.Len())
+	assert.Equal(t, queueCapacity, qContainer.Len())
 
 	// Step 3: Verify that the dropped message is not in the queue and original messages are intact.
 	// Pop messages one by one and check their MsgID and Type.
@@ -1164,7 +1033,7 @@ func TestHandleMessageQueueFullAndDropping(t *testing.T) {
 		popCtx, popCancel := context.WithTimeout(t.Context(), 200*time.Millisecond)
 		// Use FilterAny since we are just checking the contents, not a live consumption scenario.
 		// The prioritizer does not matter here as we drain the queue completely.
-		msg := qContainer.Q.Pop(popCtx, queue.NewCommitteeQueuePrioritizer(qContainer.queueState), queue.FilterAny)
+		msg := qContainer.Pop(popCtx, queue.NewCommitteeQueuePrioritizer(qState), queue.FilterAny)
 		popCancel()
 
 		require.NotNil(t, msg)
@@ -1194,7 +1063,7 @@ func TestHandleMessageQueueFullAndDropping(t *testing.T) {
 	finalPopCtx, finalPopCancel := context.WithTimeout(t.Context(), 200*time.Millisecond)
 	defer finalPopCancel()
 
-	assert.Nil(t, qContainer.Q.Pop(finalPopCtx, queue.NewCommitteeQueuePrioritizer(qContainer.queueState), queue.FilterAny))
+	assert.Nil(t, qContainer.Pop(finalPopCtx, queue.NewCommitteeQueuePrioritizer(qState), queue.FilterAny))
 }
 
 // TestConsumeQueueStopsOnErrNoValidDuties verifies that ConsumeQueue stops
@@ -1220,27 +1089,17 @@ func TestConsumeQueueStopsOnErrNoValidDuties(t *testing.T) {
 	}
 
 	slot := phase0.Slot(123)
-	q := queueContainer{
-		Q: queue.New(logger, 10),
-		queueState: &queue.State{
-			HasRunningInstance: true,
-			Height:             specqbft.Height(slot),
-			Slot:               slot,
-			Round:              1,
-		},
-	}
+	q := queue.New(logger, 10)
 
 	// Add multiple messages
-	msg1 := makeTestSSVMessage(t, spectypes.SSVConsensusMsgType, spectypes.MessageID{1}, &specqbft.Message{Height: specqbft.Height(slot), MsgType: specqbft.ProposalMsgType})
-	msg2 := makeTestSSVMessage(t, spectypes.SSVConsensusMsgType, spectypes.MessageID{2}, &specqbft.Message{Height: specqbft.Height(slot), MsgType: specqbft.PrepareMsgType})
-	msg3 := makeTestSSVMessage(t, spectypes.SSVConsensusMsgType, spectypes.MessageID{3}, &specqbft.Message{Height: specqbft.Height(slot), MsgType: specqbft.CommitMsgType})
-	q.Q.TryPush(msg1)
-	q.Q.TryPush(msg2)
-	q.Q.TryPush(msg3)
-
-	committeeRunner := &runner.CommitteeRunner{
-		BaseRunner: &runner.BaseRunner{State: &runner.State{RunningInstance: &instance.Instance{State: &specqbft.State{}}}},
-	}
+	msg1 := makeTestSSVMessage(t, spectypes.SSVConsensusMsgType, spectypes.MessageID{1}, &specqbft.Message{Height: specqbft.Height(slot), Round: 1, MsgType: specqbft.ProposalMsgType})
+	msg2 := makeTestSSVMessage(t, spectypes.SSVConsensusMsgType, spectypes.MessageID{2}, &specqbft.Message{Height: specqbft.Height(slot), Round: 1, MsgType: specqbft.PrepareMsgType})
+	msg3 := makeTestSSVMessage(t, spectypes.SSVConsensusMsgType, spectypes.MessageID{3}, &specqbft.Message{Height: specqbft.Height(slot), Round: 1, MsgType: specqbft.CommitMsgType})
+	q.TryPush(msg1)
+	q.TryPush(msg2)
+	q.TryPush(msg3)
+
+	committeeRunner := newCommitteeRunnerForTest(slot, 1, false, nil)
 
 	var processedMessagesCount int32
 	handler := func(ctx context.Context, _ *zap.Logger, msg *queue.SSVMessage) error {
@@ -1259,7 +1118,7 @@ func TestConsumeQueueStopsOnErrNoValidDuties(t *testing.T) {
 	committee.ConsumeQueue(ctx, logger, q, handler, committeeRunner)
 
 	assert.Equal(t, int32(1), atomic.LoadInt32(&processedMessagesCount))
-	assert.Equal(t, 2, q.Q.Len())
+	assert.Equal(t, 2, q.Len())
 }
 
 // TestConsumeQueueBurstTraffic verifies that under a burst of interleaved messages,
@@ -1283,19 +1142,11 @@ func TestConsumeQueueBurstTraffic(t *testing.T) {
 	slot := phase0.Slot(42)
 	committee := &Committee{
 		networkConfig:   networkconfig.TestNetwork,
-		Queues:          make(map[phase0.Slot]queueContainer),
+		Queues:          make(map[phase0.Slot]queue.Queue),
 		Runners:         make(map[phase0.Slot]*runner.CommitteeRunner),
 		CommitteeMember: &spectypes.CommitteeMember{},
 	}
-	qc := queueContainer{
-		Q: queue.New(logger, 1000),
-		queueState: &queue.State{
-			HasRunningInstance: true,
-			Height:             specqbft.Height(slot),
-			Slot:               slot,
-			Round:              1,
-		},
-	}
+	qc := queue.New(logger, 1000)
 	committee.Queues[slot] = qc
 
 	// Mark that consensus is already decided & proposal accepted → partial-sigs allowed
@@ -1306,19 +1157,7 @@ func TestConsumeQueueBurstTraffic(t *testing.T) {
 			MsgType: specqbft.ProposalMsgType,
 		},
 	}
-	committee.Runners[slot] = &runner.CommitteeRunner{
-		BaseRunner: &runner.BaseRunner{
-			State: &runner.State{
-				RunningInstance: &instance.Instance{
-					State: &specqbft.State{
-						Decided:                         true,
-						ProposalAcceptedForCurrentRound: acceptedProposal,
-						Round:                           1,
-					},
-				},
-			},
-		},
-	}
+	committee.Runners[slot] = newCommitteeRunnerForTest(slot, 1, true, acceptedProposal)
 
 	// --- Build 200 randomized messages and count expected per priority bucket ---
 	var (
@@ -1404,7 +1243,7 @@ func TestConsumeQueueBurstTraffic(t *testing.T) {
 		allMsgs[i], allMsgs[j] = allMsgs[j], allMsgs[i]
 	})
 	for _, m := range allMsgs {
-		require.True(t, qc.Q.TryPush(m))
+		require.True(t, qc.TryPush(m))
 	}
 
 	// --- Drain the queue, capturing the priority bucket of each popped message ---
@@ -1497,7 +1336,7 @@ func TestQueueLoadAndSaturationScenarios(t *testing.T) {
 		committee := &Committee{
 			logger:          logger,
 			networkConfig:   networkconfig.TestNetwork,
-			Queues:          make(map[phase0.Slot]queueContainer),
+			Queues:          make(map[phase0.Slot]queue.Queue),
 			Runners:         make(map[phase0.Slot]*runner.CommitteeRunner),
 			CommitteeMember: &spectypes.CommitteeMember{},
 		}
@@ -1506,15 +1345,8 @@ func TestQueueLoadAndSaturationScenarios(t *testing.T) {
 		nextRound := specqbft.Round(2)
 		queueCapacity := 3
 
-		qContainer := queueContainer{
-			Q: queue.New(logger, queueCapacity),
-			queueState: &queue.State{
-				HasRunningInstance: true,
-				Height:             specqbft.Height(slot),
-				Slot:               slot,
-				Round:              currentRound,
-			},
-		}
+		qContainer := queue.New(logger, queueCapacity)
+		qState := newCommitteeQueueStateForTest(slot, currentRound, true, committee.CommitteeMember.GetQuorum())
 		committee.Queues[slot] = qContainer
 
 		// 1. Fill the queue's inbox channel to capacity using HandleMessage.
@@ -1524,7 +1356,7 @@ func TestQueueLoadAndSaturationScenarios(t *testing.T) {
 			testMsg := makeTestSSVMessage(t, spectypes.SSVConsensusMsgType, msgID, prepareMsgBody)
 			committee.EnqueueMessage(ctx, testMsg)
 		}
-		require.Equal(t, queueCapacity, qContainer.Q.Len())
+		require.Equal(t, queueCapacity, qContainer.Len())
 
 		// 2. Attempt to HandleMessage a new Prepare message for the *nextRound*.
 		poppableMsgBody := &specqbft.Message{Height: specqbft.Height(slot), Round: nextRound, MsgType: specqbft.PrepareMsgType}
@@ -1532,13 +1364,13 @@ func TestQueueLoadAndSaturationScenarios(t *testing.T) {
 		committee.EnqueueMessage(ctx, poppableTestMsg)
 
 		// 3. Verify the poppable message was dropped.
-		assert.Equal(t, queueCapacity, qContainer.Q.Len())
+		assert.Equal(t, queueCapacity, qContainer.Len())
 
 		// 4. Verify the content of the queue.
 		drainedMessages := make([]*queue.SSVMessage, 0, queueCapacity)
 		for i := 0; i < queueCapacity; i++ {
 			popCtx, popCancel := context.WithTimeout(t.Context(), 200*time.Millisecond)
-			msg := qContainer.Q.Pop(popCtx, queue.NewCommitteeQueuePrioritizer(qContainer.queueState), queue.FilterAny)
+			msg := qContainer.Pop(popCtx, queue.NewCommitteeQueuePrioritizer(qState), queue.FilterAny)
 			popCancel() // Ensure cancellation happens after Pop or timeout
 			require.NotNil(t, msg)
 			drainedMessages = append(drainedMessages, msg)
@@ -1547,7 +1379,7 @@ func TestQueueLoadAndSaturationScenarios(t *testing.T) {
 		finalPopCtx, finalPopCancel := context.WithTimeout(t.Context(), 200*time.Millisecond)
 		defer finalPopCancel()
 
-		assert.Nil(t, qContainer.Q.Pop(finalPopCtx, queue.NewCommitteeQueuePrioritizer(qContainer.queueState), queue.FilterAny), "Queue should be empty after draining initial messages")
+		assert.Nil(t, qContainer.Pop(finalPopCtx, queue.NewCommitteeQueuePrioritizer(qState), queue.FilterAny), "Queue should be empty after draining initial messages")
 
 		foundNextRoundMessage := false
 		for _, msg := range drainedMessages {
@@ -1573,7 +1405,7 @@ func TestQueueLoadAndSaturationScenarios(t *testing.T) {
 		committee := &Committee{
 			logger:          logger,
 			networkConfig:   networkconfig.TestNetwork,
-			Queues:          make(map[phase0.Slot]queueContainer),
+			Queues:          make(map[phase0.Slot]queue.Queue),
 			Runners:         make(map[phase0.Slot]*runner.CommitteeRunner),
 			CommitteeMember: &spectypes.CommitteeMember{},
 		}
@@ -1581,15 +1413,8 @@ func TestQueueLoadAndSaturationScenarios(t *testing.T) {
 		currentRound := specqbft.Round(1)
 		queueCapacity := 3
 
-		qContainer := queueContainer{
-			Q: queue.New(logger, queueCapacity),
-			queueState: &queue.State{
-				HasRunningInstance: true,
-				Height:             specqbft.Height(slot),
-				Slot:               slot,
-				Round:              currentRound,
-			},
-		}
+		qContainer := queue.New(logger, queueCapacity)
+		qState := newCommitteeQueueStateForTest(slot, currentRound, true, committee.CommitteeMember.GetQuorum())
 		committee.Queues[slot] = qContainer
 
 		// 1. Fill the queue with low-priority consensus messages
@@ -1603,7 +1428,7 @@ func TestQueueLoadAndSaturationScenarios(t *testing.T) {
 			testMsg := makeTestSSVMessage(t, spectypes.SSVConsensusMsgType, msgID, commitMsgBody)
 			committee.EnqueueMessage(ctx, testMsg)
 		}
-		require.Equal(t, queueCapacity, qContainer.Q.Len(), "Queue should be at capacity")
+		require.Equal(t, queueCapacity, qContainer.Len(), "Queue should be at capacity")
 
 		// 2. Try to add a high-priority proposal message (proposals are higher priority than commits)
 		highPriorityMsgBody := &specqbft.Message{
@@ -1623,13 +1448,13 @@ func TestQueueLoadAndSaturationScenarios(t *testing.T) {
 		committee.EnqueueMessage(ctx, highPriorityMsg)
 
 		// 3. Verify queue length still at capacity
-		assert.Equal(t, queueCapacity, qContainer.Q.Len())
+		assert.Equal(t, queueCapacity, qContainer.Len())
 
 		// 4. Verify only the original messages are in the queue
 		drainedMessages := make([]*queue.SSVMessage, 0, queueCapacity)
 		for i := 0; i < queueCapacity; i++ {
 			popCtx, popCancel := context.WithTimeout(t.Context(), 200*time.Millisecond)
-			msg := qContainer.Q.Pop(popCtx, queue.NewCommitteeQueuePrioritizer(qContainer.queueState), queue.FilterAny)
+			msg := qContainer.Pop(popCtx, queue.NewCommitteeQueuePrioritizer(qState), queue.FilterAny)
 			popCancel()
 			require.NotNil(t, msg)
 			drainedMessages = append(drainedMessages, msg)
@@ -1667,39 +1492,18 @@ func TestQueueLoadAndSaturationScenarios(t *testing.T) {
 
 		committee := &Committee{
 			networkConfig:   networkconfig.TestNetwork,
-			Queues:          make(map[phase0.Slot]queueContainer),
+			Queues:          make(map[phase0.Slot]queue.Queue),
 			Runners:         make(map[phase0.Slot]*runner.CommitteeRunner),
 			CommitteeMember: &spectypes.CommitteeMember{},
 		}
 
 		queueCapacity := 5
 		currentRound := specqbft.Round(1)
-
-		committeeRunner := &runner.CommitteeRunner{
-			BaseRunner: &runner.BaseRunner{
-				State: &runner.State{
-					RunningInstance: &instance.Instance{
-						State: &specqbft.State{
-							Decided:                         false,
-							ProposalAcceptedForCurrentRound: nil,
-							Round:                           currentRound,
-						},
-					},
-				},
-			},
-		}
-
 		slot := phase0.Slot(456)
 
-		q := queueContainer{
-			Q: queue.New(logger, queueCapacity),
-			queueState: &queue.State{
-				HasRunningInstance: true,
-				Height:             specqbft.Height(slot),
-				Slot:               slot,
-				Round:              currentRound,
-			},
-		}
+		committeeRunner := newCommitteeRunnerForTest(slot, currentRound, false, nil)
+
+		q := queue.New(logger, queueCapacity)
 
 		var (
 			processedMsgs    []*queue.SSVMessage
@@ -1728,7 +1532,7 @@ func TestQueueLoadAndSaturationScenarios(t *testing.T) {
 		for i := 0; i < queueCapacity; i++ {
 			msgID := spectypes.MessageID{byte(i + 1)}
 			prepare := &specqbft.Message{Height: specqbft.Height(slot), Round: currentRound, MsgType: specqbft.PrepareMsgType}
-			require.True(t, q.Q.TryPush(makeTestSSVMessage(t, spectypes.SSVConsensusMsgType, msgID, prepare)))
+			require.True(t, q.TryPush(makeTestSSVMessage(t, spectypes.SSVConsensusMsgType, msgID, prepare)))
 		}
 
 		time.Sleep(400 * time.Millisecond) // Give time for the consumer to process (and filter) messages
@@ -1742,7 +1546,7 @@ func TestQueueLoadAndSaturationScenarios(t *testing.T) {
 		// Push ExecuteDuty
 		execData, _ := json.Marshal(&types.ExecuteCommitteeDutyData{Duty: &spectypes.CommitteeDuty{Slot: slot}})
 		execMsg := makeTestSSVMessage(t, message.SSVEventMsgType, spectypes.MessageID{0xEE}, &types.EventMsg{Type: types.ExecuteDuty, Data: execData})
-		require.True(t, q.Q.TryPush(execMsg))
+		require.True(t, q.TryPush(execMsg))
 		select {
 		case <-handlerCalled: // Good
 		case <-time.After(1 * time.Second):
@@ -1752,7 +1556,7 @@ func TestQueueLoadAndSaturationScenarios(t *testing.T) {
 		// Push Proposal
 		proposal := &specqbft.Message{Height: specqbft.Height(slot), Round: currentRound, MsgType: specqbft.ProposalMsgType}
 		propMsg := makeTestSSVMessage(t, spectypes.SSVConsensusMsgType, spectypes.MessageID{0xFF}, proposal)
-		require.True(t, q.Q.TryPush(propMsg))
+		require.True(t, q.TryPush(propMsg))
 		select {
 		case <-handlerCalled: // Good
 		case <-time.After(1 * time.Second):
diff --git a/protocol/v2/ssv/validator/queue_validator.go b/protocol/v2/ssv/validator/queue_validator.go
index 337efe74cb..11e870bfcf 100644
--- a/protocol/v2/ssv/validator/queue_validator.go
+++ b/protocol/v2/ssv/validator/queue_validator.go
@@ -110,17 +110,23 @@ func (v *Validator) StartQueueConsumer(
 		go msgStates.Start()
 		defer msgStates.Stop()
 
+		// rState defines current runner state that will be used for deciding which messages we want to process
+		// sooner (vs which ones can wait till later).
+		rState := queue.State{
+			Quorum: v.Operator.GetQuorum(), // never changes for duty runner
+		}
+
 		for ctx.Err() == nil {
-			// Construct a representation of the current state.
-			state := queue.State{}
 			r := v.DutyRunners.DutyRunnerForMsgID(msgID)
 			if r == nil {
 				return fmt.Errorf("could not get duty runner for msg ID %v", msgID)
 			}
-			state.HasRunningInstance = r.HasRunningQBFTInstance()
-			state.Height = r.GetLastHeight()
-			state.Round = r.GetLastRound()
-			state.Quorum = v.Operator.GetQuorum()
+
+			// Update rState to incorporate the effects previously handled message might have had on the runner state.
+			rState.HasRunningInstance = r.HasRunningQBFTInstance()
+			rState.Height = r.GetLastHeight()
+			rState.Round = r.GetLastRound()
+			rState.Slot = r.GetCurrentDutySlot()
 
 			filter := queue.FilterAny
 			if !r.HasRunningDuty() {
@@ -132,7 +138,7 @@ func (v *Validator) StartQueueConsumer(
 					}
 					return e.Type == types.ExecuteDuty
 				}
-			} else if state.HasRunningInstance && !r.HasAcceptedProposalForCurrentRound() {
+			} else if rState.HasRunningInstance && !r.HasAcceptedProposalForCurrentRound() {
 				// If no proposal was accepted for the current round, skip prepare & commit messages
 				// for the current height and round.
 				filter = func(m *queue.SSVMessage) bool {
@@ -141,7 +147,7 @@ func (v *Validator) StartQueueConsumer(
 						return true
 					}
 
-					if qbftMsg.Height != state.Height || qbftMsg.Round != state.Round {
+					if qbftMsg.Height != rState.Height || qbftMsg.Round != rState.Round {
 						return true
 					}
 					return qbftMsg.MsgType != specqbft.PrepareMsgType && qbftMsg.MsgType != specqbft.CommitMsgType
@@ -149,7 +155,7 @@ func (v *Validator) StartQueueConsumer(
 			}
 
 			// Pop the highest priority message for the current state.
-			msg := q.Pop(ctx, queue.NewMessagePrioritizer(&state), filter)
+			msg := q.Pop(ctx, queue.NewMessagePrioritizer(&rState), filter)
 			if ctx.Err() != nil {
 				// Optimization: terminate fast if we can.
 				return nil
diff --git a/protocol/v2/ssv/validator/timer.go b/protocol/v2/ssv/validator/timer.go
index 218a82adf7..1b6186aad2 100644
--- a/protocol/v2/ssv/validator/timer.go
+++ b/protocol/v2/ssv/validator/timer.go
@@ -22,38 +22,27 @@ func (v *Validator) onTimeout(ctx context.Context, logger *zap.Logger, identifie
 		v.mtx.RLock() // read-lock for v.Queues
 		defer v.mtx.RUnlock()
 
-		// The relevant queue might not have been initialized yet, hence we need to check for nil here
+		// If the relevant queue hasn't been initialized yet, there isn't a running duty we can issue a
+		// timeout for, in practice this should never happen - but we need to handle this just in case.
 		q := v.Queues[identifier.GetRoleType()]
 		if q == nil {
-			return
-		}
-
-		dr := v.DutyRunners[identifier.GetRoleType()]
-		if dr == nil {
-			// runner can be nil: expired committee runners are removed, but timeout event can still be. in this case we should just skip it
-			logger.Warn("❗no duty runner found for role", fields.RunnerRole(identifier.GetRoleType()))
-			return
-		}
-		hasDuty := dr.HasRunningDuty()
-		if !hasDuty {
+			logger.Error("❗ couldn't schedule timeout event due to missing queue")
 			return
 		}
 
 		msg, err := v.createTimerMessage(identifier, height, round)
 		if err != nil {
-			logger.Debug("❗ failed to create timer msg", zap.Error(err))
+			logger.Error("❌ failed to create timer msg", zap.Error(err))
 			return
 		}
 		dec, err := queue.DecodeSSVMessage(msg)
 		if err != nil {
-			logger.Debug("❌ failed to decode timer msg", zap.Error(err))
+			logger.Error("❌ failed to decode timer msg", zap.Error(err))
 			return
 		}
 
 		if pushed := q.TryPush(dec); !pushed {
-			logger.Warn("❗️ dropping timeout message because the queue is full",
-				fields.RunnerRole(identifier.GetRoleType()),
-			)
+			logger.Error("❗️ dropping timeout message because the queue is full", fields.RunnerRole(identifier.GetRoleType()))
 			return
 		}
 	}
@@ -86,33 +75,31 @@ func (v *Validator) createTimerMessage(identifier spectypes.MessageID, height sp
 
 func (c *Committee) onTimeout(ctx context.Context, logger *zap.Logger, identifier spectypes.MessageID, height specqbft.Height) roundtimer.OnRoundTimeoutF {
 	return func(round specqbft.Round) {
-		c.mtx.RLock() // read-lock for c.Queues, c.Runners
+		c.mtx.RLock() // read-lock for c.Queues
 		defer c.mtx.RUnlock()
 
-		dr := c.Runners[phase0.Slot(height)]
-		if dr == nil { // only happens when we prune expired runners
-			logger.Debug("❗no committee runner found for slot")
-			return
-		}
-
-		hasDuty := dr.HasRunningDuty()
-		if !hasDuty {
+		// If the relevant queue hasn't been initialized yet, there isn't a running duty we can issue a
+		// timeout for, in practice this should never happen - but we need to handle this just in case.
+		// This is also possible if the queue got pruned already (due to becoming old and irrelevant).
+		q := c.Queues[phase0.Slot(height)]
+		if q == nil {
+			logger.Debug("couldn't schedule timeout event due to missing queue (likely was pruned)")
 			return
 		}
 
 		msg, err := c.createTimerMessage(identifier, height, round)
 		if err != nil {
-			logger.Debug("❗ failed to create timer msg", zap.Error(err))
+			logger.Error("❌ failed to create timer msg", zap.Error(err))
 			return
 		}
 		dec, err := queue.DecodeSSVMessage(msg)
 		if err != nil {
-			logger.Debug("❌ failed to decode timer msg", zap.Error(err))
+			logger.Error("❌ failed to decode timer msg", zap.Error(err))
 			return
 		}
 
-		if pushed := c.Queues[phase0.Slot(height)].Q.TryPush(dec); !pushed {
-			logger.Warn("❗️ dropping timeout message because the queue is full", fields.RunnerRole(identifier.GetRoleType()))
+		if pushed := q.TryPush(dec); !pushed {
+			logger.Error("❗️ dropping timeout message because the queue is full", fields.RunnerRole(identifier.GetRoleType()))
 		}
 	}
 }
diff --git a/protocol/v2/ssv/validator/validator.go b/protocol/v2/ssv/validator/validator.go
index 4be9981aa1..70e72c21c2 100644
--- a/protocol/v2/ssv/validator/validator.go
+++ b/protocol/v2/ssv/validator/validator.go
@@ -212,7 +212,7 @@ func (v *Validator) ProcessMessage(ctx context.Context, logger *zap.Logger, msg
 
 			timeoutData, err := eventMsg.GetTimeoutData()
 			if err != nil {
-				return fmt.Errorf("get event message timeout data: %w", err)
+				return fmt.Errorf("event message: get timeout data: %w", err)
 			}
 
 			if err := dutyRunner.OnTimeoutQBFT(ctx, logger, timeoutData); err != nil {
diff --git a/scripts/differ/differ_test.go b/scripts/differ/differ_test.go
index 1bb6e26b4d..3d58090a8f 100644
--- a/scripts/differ/differ_test.go
+++ b/scripts/differ/differ_test.go
@@ -14,7 +14,7 @@ func TestDiffer(t *testing.T) {
 	input := `package main
 
 		func (r *ProposerRunner) ProcessPostConsensus(signedMsg *types.SignedPartialSignatureMessage) error {
-			quorum, roots, err := r.BaseRunner.basePostConsensusMsgProcessing(r, signedMsg)
+			quorum, roots, err := r.basePostConsensusMsgProcessing(r, signedMsg)
 			if err != nil {
 				return errors.Wrap(err, "failed processing post consensus message")
 			}
@@ -28,7 +28,7 @@ func TestDiffer(t *testing.T) {
 				if err != nil {
 					// If the reconstructed signature verification failed, fall back to verifying each partial signature
 					for _, root := range roots {
-						r.BaseRunner.FallBackAndVerifyEachSignature(r.GetState().PostConsensusContainer, root)
+						r.FallBackAndVerifyEachSignature(r.GetState().PostConsensusContainer, root)
 					}
 					return errors.Wrap(err, "got post-consensus quorum but it has invalid signatures")
 				}
@@ -49,7 +49,7 @@ func TestDiffer(t *testing.T) {
 		}`
 
 	expectedOutput := `func (r *ProposerRunner) ProcessPostConsensus(signedMsg *SignedPartialSignatureMessage) error {
-	quorum, roots, err := r.BaseRunner.basePostConsensusMsgProcessing(r, signedMsg)
+	quorum, roots, err := r.basePostConsensusMsgProcessing(r, signedMsg)
 	if err != nil {
 		return errors.Wrap(err, "failed processing post consensus message")
 	}
@@ -60,7 +60,7 @@ func TestDiffer(t *testing.T) {
 		sig, err := r.GetState().ReconstructBeaconSig(r.GetState().PostConsensusContainer, root, r.GetShare().ValidatorPubKey)
 		if err != nil {
 			for _, root := range roots {
-				r.BaseRunner.FallBackAndVerifyEachSignature(r.GetState().PostConsensusContainer, root)
+				r.FallBackAndVerifyEachSignature(r.GetState().PostConsensusContainer, root)
 			}
 			return errors.Wrap(err, "got post-consensus quorum but it has invalid signatures")
 		}
diff --git a/scripts/differ/transformers_test.go b/scripts/differ/transformers_test.go
index dfbc87681e..6f50ed5153 100644
--- a/scripts/differ/transformers_test.go
+++ b/scripts/differ/transformers_test.go
@@ -52,7 +52,7 @@ One qbft.Two`
 
 func TestAll(t *testing.T) {
 	input := `func (r *ProposerRunner) ProcessPreConsensus(signedMsg *types.SignedPartialSignatureMessage) error {
-	quorum, roots, err := r.BaseRunner.basePreConsensusMsgProcessing(r, signedMsg)
+	quorum, roots, err := r.basePreConsensusMsgProcessing(r, signedMsg)
 	if err != nil {
 		return errors.Wrap(err, "failed processing randao message")
 	}
@@ -68,7 +68,7 @@ func TestAll(t *testing.T) {
 	fullSig, err := r.GetState().ReconstructBeaconSig(r.GetState().PreConsensusContainer, root, r.GetShare().ValidatorPubKey)
 	if err != nil {
 		// If the reconstructed signature verification failed, fall back to verifying each partial signature
-		r.BaseRunner.FallBackAndVerifyEachSignature(r.GetState().PreConsensusContainer, root)
+		r.FallBackAndVerifyEachSignature(r.GetState().PreConsensusContainer, root)
 		return errors.Wrap(err, "got pre-consensus quorum but it has invalid signatures")
 	}
 
@@ -101,7 +101,7 @@ func TestAll(t *testing.T) {
 		DataSSZ: byts,
 	}
 
-	if err := r.BaseRunner.decide(r, input); err != nil {
+	if err := r.decide(r, input); err != nil {
 		return errors.Wrap(err, "qbft-decide")
 	}
 
@@ -109,7 +109,7 @@ func TestAll(t *testing.T) {
 }`
 
 	expected := `func (r *ProposerRunner) ProcessPreConsensus(signedMsg *SignedPartialSignatureMessage) error {
-	quorum, roots, err := r.BaseRunner.basePreConsensusMsgProcessing(r, signedMsg)
+	quorum, roots, err := r.basePreConsensusMsgProcessing(r, signedMsg)
 	if err != nil {
 		return errors.Wrap(err, "failed processing randao message")
 	}
@@ -119,7 +119,7 @@ func TestAll(t *testing.T) {
 	root := roots[0]
 	fullSig, err := r.GetState().ReconstructBeaconSig(r.GetState().PreConsensusContainer, root, r.GetShare().ValidatorPubKey)
 	if err != nil {
-		r.BaseRunner.FallBackAndVerifyEachSignature(r.GetState().PreConsensusContainer, root)
+		r.FallBackAndVerifyEachSignature(r.GetState().PreConsensusContainer, root)
 		return errors.Wrap(err, "got pre-consensus quorum but it has invalid signatures")
 	}
 	duty := r.GetState().CurrentDuty
@@ -145,7 +145,7 @@ func TestAll(t *testing.T) {
 		Version: ver,
 		DataSSZ: byts,
 	}
-	if err := r.BaseRunner.decide(r, input); err != nil {
+	if err := r.decide(r, input); err != nil {
 		return errors.Wrap(err, "qbft-decide")
 	}
 	return nil