From 59bae88b88ebe6a53eea1342104c2db7b1f7d57c Mon Sep 17 00:00:00 2001
From: Toralf Wittner <tw@dtex.org>
Date: Thu, 16 Jul 2026 14:09:47 +0200
Subject: [PATCH] Move proposal/VID-share pairing into `Consensus`.

The coordinator used to join validated proposals with this node's VID
share in ad-hoc caches and only hand the completed pair to consensus.

Consensus now receives the halves as separate inputs (`Proposal`,
`VidShare`), parks the first arrival keyed by (view, payload
commitment), and processes the pair when it completes. The coordinator
effects formerly performed at pairing time, i.e. persisting the proposal,
VID share and state cert, and seeding the VID reconstructor, are
driven by a new `ProposalPaired` output.

Behavior is unchanged: `ProposalPaired` is emitted before the safety
and DRB checks, matching the previous persist-before-checks order, and
the parked halves are dropped on pairing and on local view GC exactly
as the old caches were.
---
 .../hotshot/new-protocol/bench/src/metrics.rs |   2 +-
 crates/hotshot/new-protocol/src/consensus.rs  | 101 +++++++++++++--
 .../hotshot/new-protocol/src/coordinator.rs   | 108 +++++-----------
 .../new-protocol/src/tests/common/utils.rs    |  57 ++++++---
 .../new-protocol/src/tests/consensus.rs       | 120 +++++++++---------
 .../new-protocol/src/tests/epoch_change.rs    |  47 ++++---
 6 files changed, 245 insertions(+), 190 deletions(-)

diff --git a/crates/hotshot/new-protocol/bench/src/metrics.rs b/crates/hotshot/new-protocol/bench/src/metrics.rs
index 0166ba53caa..195c687f3c2 100644
--- a/crates/hotshot/new-protocol/bench/src/metrics.rs
+++ b/crates/hotshot/new-protocol/bench/src/metrics.rs
@@ -70,7 +70,7 @@ impl MetricsCollector {
                 self.view_mut(v).vid_disperse_ns = Some(ts);
             },
             // Replica: proposal processing
-            ConsensusInput::ProposalWithVidShare(_sender, p, _) => {
+            ConsensusInput::Proposal(_sender, p) => {
                 let v = *p.view_number();
                 self.view_mut(v).proposal_recv_ns = Some(ts);
             },
diff --git a/crates/hotshot/new-protocol/src/consensus.rs b/crates/hotshot/new-protocol/src/consensus.rs
index e8655735ca0..9503bd1423a 100644
--- a/crates/hotshot/new-protocol/src/consensus.rs
+++ b/crates/hotshot/new-protocol/src/consensus.rs
@@ -102,11 +102,12 @@ pub enum ConsensusInput<T: NodeType> {
     },
     EpochChange(EpochChangeMessage<T>),
     HeaderCreated(ViewNumber, Commitment<Leaf2<T>>, T::BlockHeader),
-    ProposalWithVidShare(
-        T::SignatureKey,
-        ProposalMessage<T, Validated>,
-        VidDisperseShare2<T>,
-    ),
+    /// A validated proposal. Consensus parks it until this node's VID share
+    /// for the same payload arrives ([`ConsensusInput::VidShare`]) and only
+    /// processes the two together.
+    Proposal(T::SignatureKey, ProposalMessage<T, Validated>),
+    /// This node's validated VID share.
+    VidShare(VidDisperseShare2<T>),
     FetchedProposal(ProposalMessage<T, Validated>),
     StateValidated(StateResponse<T>),
     StateValidationFailed(StateResponse<T>),
@@ -156,6 +157,11 @@ pub enum ConsensusOutput<T: NodeType> {
     ViewChanged(ViewNumber, EpochNumber),
     /// A view timed out with a timeout certificate.
     ViewTimedOut(ViewNumber),
+    /// A validated proposal met this node's VID share.
+    ProposalPaired {
+        proposal: SignedProposal<T, Proposal<T>>,
+        vid_share: VidDisperseShare2<T>,
+    },
     ProposalValidated {
         proposal: SignedProposal<T, Proposal<T>>,
         sender: T::SignatureKey,
@@ -178,6 +184,13 @@ pub enum ConsensusOutput<T: NodeType> {
     BroadcastVidShare(VidDisperseShare2<T>),
 }
 
+type UnpairedProposals<T> = BTreeMap<
+    (ViewNumber, VidCommitment2),
+    (<T as NodeType>::SignatureKey, ProposalMessage<T, Validated>),
+>;
+
+type UnpairedVidShares<T> = BTreeMap<(ViewNumber, VidCommitment2), VidDisperseShare2<T>>;
+
 /// Views to retain decide inputs (`proposals`, `certs`, `certs2`) behind the
 /// decided view, letting a late-broadcast Cert2 decide an older gap view.
 pub(crate) const DECIDE_BUFFER: u64 = 20;
@@ -190,6 +203,8 @@ pub struct Consensus<T: NodeType> {
     signed_proposals: BTreeMap<ViewNumber, SignedProposal<T, Proposal<T>>>,
     proposed_views: BTreeSet<ViewNumber>,
     vid_shares: BTreeMap<ViewNumber, VidDisperseShare2<T>>,
+    unpaired_proposals: UnpairedProposals<T>,
+    unpaired_vid_shares: UnpairedVidShares<T>,
     states_verified: BTreeMap<ViewNumber, Commitment<Leaf2<T>>>,
     blocks_reconstructed: BTreeSet<(ViewNumber, VidCommitment2)>,
     blocks: BTreeMap<(ViewNumber, VidCommitment2), T::BlockPayload>,
@@ -369,6 +384,8 @@ impl<T: NodeType> Consensus<T> {
             state_certs: BTreeMap::new(),
             upgrade_lock,
             vid_shares: BTreeMap::new(),
+            unpaired_proposals: BTreeMap::new(),
+            unpaired_vid_shares: BTreeMap::new(),
             epoch_height: epoch_height.into(),
         }
     }
@@ -659,14 +676,18 @@ impl<T: NodeType> Consensus<T> {
             input.view_number()
         };
         let proto = match input {
-            ConsensusInput::ProposalWithVidShare(sender, proposal, vid_share) => {
+            ConsensusInput::Proposal(sender, proposal) => {
                 debug!(
                     sender = %KeyPrefix::from(&sender),
                     block = %proposal.proposal.data.block_header.block_number(),
                     epoch = %proposal.proposal.data.epoch,
-                    "apply: proposal+vid share"
+                    "apply: proposal"
                 );
-                self.handle_proposal_with_vid_share(sender, proposal, vid_share, outbox)
+                self.pair_proposal(sender, proposal, outbox)
+            },
+            ConsensusInput::VidShare(vid_share) => {
+                debug!("apply: vid share");
+                self.pair_vid_share(vid_share, outbox)
             },
             ConsensusInput::FetchedProposal(message) => {
                 debug!(
@@ -946,7 +967,10 @@ impl<T: NodeType> Consensus<T> {
         match scope {
             GcScope::Local(view) => {
                 let c = Commitment::default_commitment_no_preimage();
+                let vc = VidCommitment2::default();
                 self.headers = self.headers.split_off(&(view, c));
+                self.unpaired_proposals = self.unpaired_proposals.split_off(&(view, vc));
+                self.unpaired_vid_shares = self.unpaired_vid_shares.split_off(&(view, vc));
                 self.proposed_views = self.proposed_views.split_off(&view);
                 self.states_verified = self.states_verified.split_off(&view);
                 self.timeout_certs = self.timeout_certs.split_off(&view);
@@ -1009,6 +1033,64 @@ impl<T: NodeType> Consensus<T> {
         self.proposals.insert(view, proposal);
     }
 
+    /// Pair a validated proposal with this node's VID share for the same payload.
+    ///
+    /// The half arriving first is parked, keyed by (view, payload commitment).
+    fn pair_proposal(
+        &mut self,
+        sender: T::SignatureKey,
+        proposal: ProposalMessage<T, Validated>,
+        outbox: &mut Outbox<ConsensusOutput<T>>,
+    ) -> Protocol {
+        let view = proposal.view_number();
+        let VidCommitment::V2(commit) = proposal.proposal.data.block_header.payload_commitment()
+        else {
+            warn!(%view, "proposal payload commitment is not V2, discarding");
+            return Protocol::Abort;
+        };
+        let Some(vid_share) = self.unpaired_vid_shares.remove(&(view, commit)) else {
+            self.unpaired_proposals
+                .insert((view, commit), (sender, proposal));
+            return Protocol::Abort;
+        };
+        self.on_proposal_paired(sender, proposal, vid_share, outbox)
+    }
+
+    /// Pair this node's VID share with a validated proposal for the same payload.
+    ///
+    /// The half arriving first is parked, keyed by (view, payload commitment).
+    fn pair_vid_share(
+        &mut self,
+        vid_share: VidDisperseShare2<T>,
+        outbox: &mut Outbox<ConsensusOutput<T>>,
+    ) -> Protocol {
+        let key = (vid_share.view_number(), vid_share.payload_commitment);
+        let Some((sender, proposal)) = self.unpaired_proposals.remove(&key) else {
+            self.unpaired_vid_shares.insert(key, vid_share);
+            return Protocol::Abort;
+        };
+        self.on_proposal_paired(sender, proposal, vid_share, outbox)
+    }
+
+    fn on_proposal_paired(
+        &mut self,
+        sender: T::SignatureKey,
+        proposal: ProposalMessage<T, Validated>,
+        vid_share: VidDisperseShare2<T>,
+        outbox: &mut Outbox<ConsensusOutput<T>>,
+    ) -> Protocol {
+        // Parked halves for this and older views can no longer pair.
+        let view = proposal.view_number();
+        let vc = VidCommitment2::default();
+        self.unpaired_proposals = self.unpaired_proposals.split_off(&(view + 1, vc));
+        self.unpaired_vid_shares = self.unpaired_vid_shares.split_off(&(view + 1, vc));
+        outbox.push_back(ConsensusOutput::ProposalPaired {
+            proposal: proposal.proposal.clone(),
+            vid_share: vid_share.clone(),
+        });
+        self.handle_proposal_with_vid_share(sender, proposal, vid_share, outbox)
+    }
+
     #[instrument(level = "debug", skip_all)]
     fn handle_proposal_with_vid_share(
         &mut self,
@@ -2625,7 +2707,8 @@ impl<T: NodeType> ConsensusInput<T> {
             ConsensusInput::AdvanceView(cert) => cert.view_number() + 1,
             ConsensusInput::EpochRootCertificates { cert1, .. } => cert1.view_number(),
             ConsensusInput::HeaderCreated(view, ..) => *view,
-            ConsensusInput::ProposalWithVidShare(_, prop, _) => prop.view_number(),
+            ConsensusInput::Proposal(_, prop) => prop.view_number(),
+            ConsensusInput::VidShare(share) => share.view_number(),
             ConsensusInput::FetchedProposal(prop) => prop.view_number(),
             ConsensusInput::StateValidated(response) => response.view,
             ConsensusInput::StateValidationFailed(request) => request.view,
diff --git a/crates/hotshot/new-protocol/src/coordinator.rs b/crates/hotshot/new-protocol/src/coordinator.rs
index cfbfd578927..ea151909a51 100644
--- a/crates/hotshot/new-protocol/src/coordinator.rs
+++ b/crates/hotshot/new-protocol/src/coordinator.rs
@@ -14,7 +14,7 @@ use hotshot::{HotShotInitializer, traits::BlockPayload, types::SignatureKey};
 use hotshot_types::{
     consensus::{ConsensusMetricsValue, ParticipationTracker},
     data::{
-        EpochNumber, Leaf2, VidCommitment, VidCommitment2, VidDisperseShare2, ViewNumber,
+        EpochNumber, Leaf2, VidCommitment, VidCommitment2, ViewNumber,
         vid_disperse::vid_total_weight,
     },
     epoch_membership::EpochMembershipCoordinator,
@@ -51,7 +51,7 @@ use crate::{
     },
     network::Cliquenet,
     outbox::Outbox,
-    proposal::{ProposalValidator, ValidatedProposal, VidShareValidator},
+    proposal::{ProposalValidator, VidShareValidator},
     state::{HeaderRequest, StateEntry, StateManager, StateManagerOutput},
     storage::{NewProtocolStorage, Storage},
     vid::{VidDisperseRequest, VidDisperser, VidFragmentAccumulator, VidReconstructor},
@@ -112,10 +112,6 @@ pub struct Coordinator<T: NodeType, S> {
     pending_proposal_fetches: PendingProposalFetches<T>,
     #[builder(skip)]
     requested_missing_proposals: HashSet<ProposalFetchKey<T>>,
-    #[builder(default)]
-    cached_validated_proposals: BTreeMap<(ViewNumber, VidCommitment2), ValidatedProposal<T>>,
-    #[builder(default)]
-    cached_vid_shares: BTreeMap<(ViewNumber, VidCommitment2), VidDisperseShare2<T>>,
     #[builder(skip)]
     da_payloads: BTreeMap<(ViewNumber, VidCommitment2), PendingDa<T>>,
     metrics: Option<metrics::Metrics>,
@@ -475,14 +471,7 @@ where
                 }
                 Some(item) = self.share_validator.next() => match item {
                     Ok(vid_share) => {
-                        let view = vid_share.view_number();
-                        let key = (view, vid_share.payload_commitment);
-                        let Some(validated) = self.cached_validated_proposals.remove(&key) else {
-                            // Wait for the proposal
-                            self.cached_vid_shares.insert(key, vid_share);
-                            continue;
-                        };
-                        return self.on_proposal_and_vid_share(validated, vid_share)
+                        return Ok(ConsensusInput::VidShare(vid_share))
                     },
                     Err(e) => {
                         return Err(CoordinatorError::regular(e).context("vid share validation"))
@@ -501,21 +490,7 @@ where
                         {
                             error!(%epoch, %err, "network apply_epoch failed");
                         }
-
-                        let view = validated.message.proposal.data.view_number();
-                        let VidCommitment::V2(commit) =
-                            validated.message.proposal.data.block_header.payload_commitment()
-                        else {
-                            warn!(%view, "proposal payload commitment is not V2, discarding");
-                            continue;
-                        };
-                        let key = (view, commit);
-                        let Some(vid_share) = self.cached_vid_shares.remove(&key) else {
-                            // Wait for the vid share describing this payload.
-                            self.cached_validated_proposals.insert(key, validated);
-                            continue;
-                        };
-                        return self.on_proposal_and_vid_share(validated, vid_share)
+                        return Ok(ConsensusInput::Proposal(validated.sender, validated.message))
                     }
                     Err(e) => {
                         return Err(CoordinatorError::regular(e).context("proposal validation"))
@@ -751,6 +726,31 @@ where
                     }
                 }
             },
+            ConsensusOutput::ProposalPaired {
+                proposal,
+                vid_share,
+            } => {
+                let view = proposal.data.view_number;
+                debug!(%node, %view, "proposal paired with vid share");
+                self.storage.append_vid(vid_share.clone());
+                self.storage.append_proposal(proposal.data.clone());
+                if let Some(state_cert) = &proposal.data.state_cert {
+                    self.storage.append_state_cert(
+                        ViewNumber::new(state_cert.light_client_state.view_number),
+                        state_cert.clone(),
+                    );
+                }
+                let expected_param = self.expected_vid_param(vid_share.target_epoch);
+                self.vid_reconstructor.handle_proposal(
+                    view,
+                    vid_share.payload_commitment,
+                    proposal.data.block_header.metadata().clone(),
+                    proposal.data.epoch,
+                    expected_param,
+                );
+                self.vid_reconstructor
+                    .handle_vid_share(self.public_key.clone(), vid_share);
+            },
             ConsensusOutput::SendProposal(proposal) => {
                 let view = proposal.data.view_number;
                 let epoch = proposal.data.epoch;
@@ -1309,52 +1309,6 @@ where
         init_avidm_gf2_param(total_weight).ok()
     }
 
-    fn on_proposal_and_vid_share(
-        &mut self,
-        validated: ValidatedProposal<T>,
-        vid_share: VidDisperseShare2<T>,
-    ) -> Result<ConsensusInput<T>, CoordinatorError> {
-        self.storage.append_vid(vid_share.clone());
-        self.storage
-            .append_proposal(validated.message.proposal.data.clone());
-
-        if let Some(state_cert) = &validated.message.proposal.data.state_cert {
-            self.storage.append_state_cert(
-                ViewNumber::new(state_cert.light_client_state.view_number),
-                state_cert.clone(),
-            );
-        }
-
-        let expected_param = self.expected_vid_param(vid_share.target_epoch);
-        let proposal = &validated.message.proposal.data;
-        self.vid_reconstructor.handle_proposal(
-            proposal.view_number(),
-            vid_share.payload_commitment,
-            proposal.block_header.metadata().clone(),
-            proposal.epoch,
-            expected_param,
-        );
-        // This is our own share, addressed to us by the leader and already
-        // verified by the share validator.
-        self.vid_reconstructor
-            .handle_vid_share(self.public_key.clone(), vid_share.clone());
-
-        // GC for the cache
-        let view = validated.message.proposal.data.view_number();
-        self.cached_vid_shares = self
-            .cached_vid_shares
-            .split_off(&(view + 1, VidCommitment2::default()));
-        self.cached_validated_proposals = self
-            .cached_validated_proposals
-            .split_off(&(view + 1, VidCommitment2::default()));
-
-        Ok(ConsensusInput::ProposalWithVidShare(
-            validated.sender,
-            validated.message,
-            vid_share,
-        ))
-    }
-
     fn broadcast(
         &self,
         message_type: ConsensusMessage<T, Validated>,
@@ -1718,11 +1672,7 @@ where
         self.consensus.gc(scope);
         match scope {
             GcScope::Local(view) => {
-                let vc = VidCommitment2::default();
                 self.block_builder.gc(view);
-                self.cached_validated_proposals =
-                    self.cached_validated_proposals.split_off(&(view, vc));
-                self.cached_vid_shares = self.cached_vid_shares.split_off(&(view, vc));
                 self.vid_disperser.gc(view);
                 self.vid_fragment_accumulator.gc(view);
                 // When we enter a new view, we do not want to GC certain data
diff --git a/crates/hotshot/new-protocol/src/tests/common/utils.rs b/crates/hotshot/new-protocol/src/tests/common/utils.rs
index 0860975c128..621904bad11 100644
--- a/crates/hotshot/new-protocol/src/tests/common/utils.rs
+++ b/crates/hotshot/new-protocol/src/tests/common/utils.rs
@@ -147,11 +147,17 @@ impl TestView {
             )),
         }
     }
-    pub fn proposal_input_consensus(&self, recipient_key: &BLSPubKey) -> ConsensusInput<TestTypes> {
-        ConsensusInput::ProposalWithVidShare(
-            self.leader_public_key,
-            self.proposal_message(),
-            self.vid_share_for(recipient_key),
+    /// Build the proposal and VID share inputs a node receives for this view.
+    ///
+    /// Consensus pairs the two internally, so both must be applied
+    /// (cf. [`ConsensusHarness::apply_pair`]).
+    pub fn proposal_input_consensus(
+        &self,
+        recipient_key: &BLSPubKey,
+    ) -> (ConsensusInput<TestTypes>, ConsensusInput<TestTypes>) {
+        (
+            ConsensusInput::Proposal(self.leader_public_key, self.proposal_message()),
+            ConsensusInput::VidShare(self.vid_share_for(recipient_key)),
         )
     }
 
@@ -1067,24 +1073,20 @@ impl ConsensusHarness {
     /// actions that consensus expects feedback for.
     pub async fn apply(&mut self, input: ConsensusInput<TestTypes>) {
         let mut outbox = Outbox::new();
-        // The coordinator persists incoming proposals and VID shares before
-        // consensus sees them; simulate those storage confirmations.
-        if let ConsensusInput::ProposalWithVidShare(_, msg, vid_share) = &input {
-            let view = msg.proposal.data.view_number;
-            let commitment = proposal_commitment(&msg.proposal.data);
-            self.consensus.apply(
-                ConsensusInput::Stored(StorageOutput::Proposal(view, commitment)),
-                &mut outbox,
-            );
-            self.consensus.apply(
-                ConsensusInput::Stored(StorageOutput::Vid(vid_share.view_number)),
-                &mut outbox,
-            );
-        }
         self.consensus.apply(input, &mut outbox);
         self.drain_outbox(&mut outbox).await;
     }
 
+    /// Apply a proposal/VID-share input pair
+    /// (cf. `TestView::proposal_input_consensus`).
+    pub async fn apply_pair(
+        &mut self,
+        (proposal, vid_share): (ConsensusInput<TestTypes>, ConsensusInput<TestTypes>),
+    ) {
+        self.apply(proposal).await;
+        self.apply(vid_share).await;
+    }
+
     async fn drain_outbox(&mut self, outbox: &mut Outbox<ConsensusOutput<TestTypes>>) {
         while let Some(output) = outbox.pop_front() {
             self.handle_output(&output, outbox).await;
@@ -1122,6 +1124,23 @@ impl ConsensusHarness {
                     outbox,
                 );
             },
+            // The coordinator persists a paired proposal and VID share;
+            // simulate the storage confirmations.
+            ConsensusOutput::ProposalPaired {
+                proposal,
+                vid_share,
+            } => {
+                let view = proposal.data.view_number;
+                let commitment = proposal_commitment(&proposal.data);
+                self.consensus.apply(
+                    ConsensusInput::Stored(StorageOutput::Proposal(view, commitment)),
+                    outbox,
+                );
+                self.consensus.apply(
+                    ConsensusInput::Stored(StorageOutput::Vid(vid_share.view_number)),
+                    outbox,
+                );
+            },
             ConsensusOutput::RequestBlockAndHeader(req) => {
                 let mock_block = MockBlock::new();
                 let parent_leaf = req.parent_proposal.clone().into();
diff --git a/crates/hotshot/new-protocol/src/tests/consensus.rs b/crates/hotshot/new-protocol/src/tests/consensus.rs
index 16483ad7347..69e75a8e4f0 100644
--- a/crates/hotshot/new-protocol/src/tests/consensus.rs
+++ b/crates/hotshot/new-protocol/src/tests/consensus.rs
@@ -39,7 +39,7 @@ async fn test_safety_genesis_no_lock() {
     let node_key = BLSPubKey::generated_from_seed_indexed([0; 32], 0).0;
 
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
 
     assert!(
@@ -67,12 +67,12 @@ async fn test_timeout_filters_vote1_not_processing() {
     // Send stale proposal (view 2, which is <= timeout_view 3).
     // It is still processed (state validation requested) but vote1 is suppressed.
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
 
     // Send fresh proposal (view 4, which is > timeout_view 3)
     harness
-        .apply(test_data.views[3].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[3].proposal_input_consensus(&node_key))
         .await;
 
     // Both proposals are processed.
@@ -94,14 +94,14 @@ async fn test_vote1_for_sequential_views() {
     let node_key = BLSPubKey::generated_from_seed_indexed([0; 32], 0).0;
 
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[0].block_reconstructed_input())
         .await;
 
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
 
     assert_eq!(
@@ -129,7 +129,7 @@ async fn test_vote1_with_seeded_parent_proposal() {
 
     // The view-2 proposal arrives and builds on the re-seeded parent.
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
 
     assert!(
@@ -156,7 +156,7 @@ async fn test_vote1_parent_reconstruction_from_lock() {
         .seed_locked_cert(test_data.views[0].cert1.clone());
 
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
 
     assert!(
@@ -178,7 +178,7 @@ async fn test_vote1_blocked_without_parent_proposal() {
         .await;
 
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
 
     assert!(
@@ -195,7 +195,7 @@ async fn test_vote1_genesis_parent() {
     let node_key = BLSPubKey::generated_from_seed_indexed([0; 32], 0).0;
 
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
 
     assert!(
@@ -213,14 +213,14 @@ async fn test_vote2_missing_cert1() {
     let node_key = BLSPubKey::generated_from_seed_indexed([0; 32], 0).0;
 
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[0].block_reconstructed_input())
         .await;
 
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[1].block_reconstructed_input())
@@ -240,14 +240,14 @@ async fn test_vote2_with_cert1() {
     let node_key = BLSPubKey::generated_from_seed_indexed([0; 32], 0).0;
 
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[0].block_reconstructed_input())
         .await;
 
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[1].block_reconstructed_input())
@@ -268,14 +268,14 @@ async fn test_single_view_decide() {
     let node_key = BLSPubKey::generated_from_seed_indexed([0; 32], 0).0;
 
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[0].block_reconstructed_input())
         .await;
 
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[1].block_reconstructed_input())
@@ -300,13 +300,13 @@ async fn test_cert2_broadcast_once() {
     let node_key = BLSPubKey::generated_from_seed_indexed([0; 32], 0).0;
 
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[0].block_reconstructed_input())
         .await;
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[1].block_reconstructed_input())
@@ -348,7 +348,7 @@ async fn test_late_cert2_decides_after_view_change_gc() {
     // Drive views 1..=4 forward with Cert1 only, so nothing decides.
     for view in &test_data.views[0..4] {
         harness
-            .apply(view.proposal_input_consensus(&node_key))
+            .apply_pair(view.proposal_input_consensus(&node_key))
             .await;
         harness.apply(view.block_reconstructed_input()).await;
         harness.apply(view.cert1_input()).await;
@@ -382,7 +382,7 @@ async fn test_gap_fill_decide_of_older_view() {
 
     // Decide view 4 without ever delivering views 1-3, leaving a gap.
     harness
-        .apply(test_data.views[3].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[3].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[3].block_reconstructed_input())
@@ -437,7 +437,7 @@ async fn test_no_duplicate_vote1() {
     let node_key = BLSPubKey::generated_from_seed_indexed([0; 32], 0).0;
 
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[0].block_reconstructed_input())
@@ -459,14 +459,14 @@ async fn test_no_duplicate_vote2() {
     let node_key = BLSPubKey::generated_from_seed_indexed([0; 32], 0).0;
 
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[0].block_reconstructed_input())
         .await;
 
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[1].block_reconstructed_input())
@@ -489,7 +489,7 @@ async fn test_state_validation_failed_removes_proposal() {
     let node_key = BLSPubKey::generated_from_seed_indexed([0; 32], 0).0;
 
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[0].block_reconstructed_input())
@@ -499,9 +499,10 @@ async fn test_state_validation_failed_removes_proposal() {
     // by directly applying the proposal input, then manually sending
     // StateValidationFailed instead of letting the harness auto-respond.
     // We need to call consensus.apply directly to avoid auto StateVerified.
-    let proposal_input = test_data.views[1].proposal_input_consensus(&node_key);
+    let (proposal_input, vid_share_input) = test_data.views[1].proposal_input_consensus(&node_key);
     let mut outbox = Outbox::new();
     harness.consensus.apply(proposal_input, &mut outbox);
+    harness.consensus.apply(vid_share_input, &mut outbox);
     harness.collected.extend(outbox.take());
 
     // Send StateVerificationFailed — removes proposal
@@ -539,14 +540,14 @@ async fn test_decide_requires_cert2() {
     let node_key = BLSPubKey::generated_from_seed_indexed([0; 32], 0).0;
 
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[0].block_reconstructed_input())
         .await;
 
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[1].block_reconstructed_input())
@@ -569,7 +570,7 @@ async fn test_vote2_missing_block_reconstructed() {
     let node_key = BLSPubKey::generated_from_seed_indexed([0; 32], 0).0;
 
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[0].block_reconstructed_input())
@@ -577,7 +578,7 @@ async fn test_vote2_missing_block_reconstructed() {
 
     // View 2: proposal + cert1, but NO block_reconstructed for view 2
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
     harness.apply(test_data.views[1].cert1_input()).await;
 
@@ -595,7 +596,7 @@ async fn test_vote2_block_reconstructed_arrives_late() {
     let node_key = BLSPubKey::generated_from_seed_indexed([0; 32], 0).0;
 
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[0].block_reconstructed_input())
@@ -603,7 +604,7 @@ async fn test_vote2_block_reconstructed_arrives_late() {
 
     // View 2: proposal + cert1 first (no block_reconstructed yet)
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
     harness.apply(test_data.views[1].cert1_input()).await;
 
@@ -627,7 +628,7 @@ async fn test_multi_view_chain_decide() {
 
     for view in &test_data.views {
         harness
-            .apply(view.proposal_input_consensus(&node_key))
+            .apply_pair(view.proposal_input_consensus(&node_key))
             .await;
         harness.apply(view.block_reconstructed_input()).await;
         harness.apply(view.cert1_input()).await;
@@ -647,7 +648,7 @@ async fn test_timeout_prevents_vote1_but_allows_vote2() {
 
     // Process view 1 to establish state.
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[0].block_reconstructed_input())
@@ -671,7 +672,7 @@ async fn test_timeout_prevents_vote1_but_allows_vote2() {
     // The proposal is still stored (inputs are processed), but vote1 is
     // suppressed because view 2 <= timeout_view.
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[1].block_reconstructed_input())
@@ -702,7 +703,7 @@ async fn test_leader_sends_proposal() {
     let mut harness = ConsensusHarness::new(leader_index).await;
 
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&leader_for_view_2))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&leader_for_view_2))
         .await;
     harness.apply(test_data.views[0].cert1_input()).await;
 
@@ -740,7 +741,7 @@ async fn test_propose_refuses_when_stored_proposal_differs_from_cert() {
     // triggers RequestBlockAndHeader for view 2 (auto-built by the harness
     // against the canonical view-1 proposal as parent).
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&leader_for_view_2))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&leader_for_view_2))
         .await;
 
     // Sanity: header for view 2 has been built off the canonical view-1
@@ -814,7 +815,7 @@ async fn test_leader_proposes_after_timeout() {
 
     // Build up locked_cert: process view 1 so cert1 sets locked_cert
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&leader_for_view_3))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&leader_for_view_3))
         .await;
     harness
         .apply(test_data.views[0].block_reconstructed_input())
@@ -849,7 +850,7 @@ async fn test_non_leader_does_not_propose() {
     let mut harness = ConsensusHarness::new(non_leader_index).await;
 
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&non_leader_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&non_leader_key))
         .await;
     harness.apply(test_data.views[0].cert1_input()).await;
 
@@ -870,7 +871,7 @@ async fn test_safety_rejects_proposal_below_lock() {
 
     // Process view 1 fully: proposal + block_reconstructed + cert1 → locked_qc = view 1
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[0].block_reconstructed_input())
@@ -879,7 +880,7 @@ async fn test_safety_rejects_proposal_below_lock() {
 
     // Process view 2 fully → locked_qc = view 2
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[1].block_reconstructed_input())
@@ -894,7 +895,7 @@ async fn test_safety_rejects_proposal_below_lock() {
     // Safety:   genesis commitment != cert1(view 2) commitment = false
     // → Proposal rejected by is_safe, no RequestState emitted.
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
 
     let state_requests_after = count_matching(harness.outputs(), is_request_state);
@@ -915,7 +916,7 @@ async fn test_vote_after_timeout_cert() {
 
     // Process view 1 fully → locked_qc = view 1
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[0].block_reconstructed_input())
@@ -924,7 +925,7 @@ async fn test_vote_after_timeout_cert() {
 
     // Process view 2 proposal + block_reconstructed (need parent data for view 3)
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[1].block_reconstructed_input())
@@ -953,7 +954,7 @@ async fn test_vote_after_timeout_cert() {
     // Process proposal for view 3. Its justify_qc is at view 2, which is
     // >= locked_qc at view 1 (liveness passes). Parent data for view 2 exists.
     harness
-        .apply(test_data.views[2].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[2].proposal_input_consensus(&node_key))
         .await;
 
     assert!(
@@ -979,7 +980,7 @@ async fn test_no_vote_after_timeout_for_proposal_below_lock() {
     // Process views 1-3 fully → locked_qc = cert1 for view 3
     for i in 0..3 {
         harness
-            .apply(test_data.views[i].proposal_input_consensus(&node_key))
+            .apply_pair(test_data.views[i].proposal_input_consensus(&node_key))
             .await;
         harness
             .apply(test_data.views[i].block_reconstructed_input())
@@ -1000,7 +1001,7 @@ async fn test_no_vote_after_timeout_for_proposal_below_lock() {
     //          justify_qc commitment ≠ locked_qc commitment → false (safety fails)
     // → Proposal rejected by is_safe.
     harness
-        .apply(test_data.views[2].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[2].proposal_input_consensus(&node_key))
         .await;
 
     assert_eq!(
@@ -1019,13 +1020,13 @@ async fn test_decide_not_repeated_for_same_view() {
 
     // Full round for view 2
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[0].block_reconstructed_input())
         .await;
     harness
-        .apply(test_data.views[1].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[1].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[1].block_reconstructed_input())
@@ -1072,7 +1073,7 @@ async fn test_restart_does_not_redecide_anchor() {
 
     // Drive the first decide after the restart at the next view.
     harness
-        .apply(decider.proposal_input_consensus(&node_key))
+        .apply_pair(decider.proposal_input_consensus(&node_key))
         .await;
     harness.apply(decider.block_reconstructed_input()).await;
     harness.apply(decider.cert1_input()).await;
@@ -1296,10 +1297,9 @@ async fn test_vote1_gated_on_storage() {
     let consensus = &mut harness.consensus;
     let mut outbox = Outbox::new();
 
-    consensus.apply(
-        test_data.views[0].proposal_input_consensus(&node_key),
-        &mut outbox,
-    );
+    let (proposal_input, vid_share_input) = test_data.views[0].proposal_input_consensus(&node_key);
+    consensus.apply(proposal_input, &mut outbox);
+    consensus.apply(vid_share_input, &mut outbox);
     consensus.apply(
         state_verified_input(&test_data.views[0].proposal.data, view),
         &mut outbox,
@@ -1344,10 +1344,9 @@ async fn test_vote2_gated_on_vid_storage() {
     let consensus = &mut harness.consensus;
     let mut outbox = Outbox::new();
 
-    consensus.apply(
-        test_data.views[0].proposal_input_consensus(&node_key),
-        &mut outbox,
-    );
+    let (proposal_input, vid_share_input) = test_data.views[0].proposal_input_consensus(&node_key);
+    consensus.apply(proposal_input, &mut outbox);
+    consensus.apply(vid_share_input, &mut outbox);
     consensus.apply(
         state_verified_input(&test_data.views[0].proposal.data, view),
         &mut outbox,
@@ -1401,10 +1400,9 @@ async fn test_pending_vote1_dropped_on_timeout() {
     let consensus = &mut harness.consensus;
     let mut outbox = Outbox::new();
 
-    consensus.apply(
-        test_data.views[0].proposal_input_consensus(&node_key),
-        &mut outbox,
-    );
+    let (proposal_input, vid_share_input) = test_data.views[0].proposal_input_consensus(&node_key);
+    consensus.apply(proposal_input, &mut outbox);
+    consensus.apply(vid_share_input, &mut outbox);
     consensus.apply(
         state_verified_input(&test_data.views[0].proposal.data, view),
         &mut outbox,
@@ -1439,7 +1437,7 @@ async fn test_proposal_release_follows_storage() {
     let mut harness = ConsensusHarness::new(leader_index).await;
 
     harness
-        .apply(test_data.views[0].proposal_input_consensus(&leader_for_view_2))
+        .apply_pair(test_data.views[0].proposal_input_consensus(&leader_for_view_2))
         .await;
     harness.apply(test_data.views[0].cert1_input()).await;
 
diff --git a/crates/hotshot/new-protocol/src/tests/epoch_change.rs b/crates/hotshot/new-protocol/src/tests/epoch_change.rs
index 5f5ad5712f3..24811260a32 100644
--- a/crates/hotshot/new-protocol/src/tests/epoch_change.rs
+++ b/crates/hotshot/new-protocol/src/tests/epoch_change.rs
@@ -49,7 +49,7 @@ async fn run_views_full(
 
     for i in range {
         harness
-            .apply(test_data.views[i].proposal_input_consensus(node_key))
+            .apply_pair(test_data.views[i].proposal_input_consensus(node_key))
             .await;
         harness
             .apply(test_data.views[i].block_reconstructed_input())
@@ -430,10 +430,12 @@ async fn test_epoch_change_votes() {
         .clone();
 
     harness
-        .apply(ConsensusInput::ProposalWithVidShare(
-            first_view.leader_public_key,
-            ProposalMessage::validated(signed_proposal),
-            vid_share,
+        .apply_pair((
+            ConsensusInput::Proposal(
+                first_view.leader_public_key,
+                ProposalMessage::validated(signed_proposal),
+            ),
+            ConsensusInput::VidShare(vid_share),
         ))
         .await;
 
@@ -460,7 +462,7 @@ async fn test_first_epoch_leader_proposes_without_drb() {
     // should propose — but maybe_propose also needs to skip the DRB check.
     for i in 0..7 {
         harness
-            .apply(test_data.views[i].proposal_input_consensus(&node_key))
+            .apply_pair(test_data.views[i].proposal_input_consensus(&node_key))
             .await;
         harness
             .apply(test_data.views[i].block_reconstructed_input())
@@ -529,10 +531,12 @@ async fn test_second_epoch_leader_proposes_without_drb() {
         .expect("VID share not found")
         .clone();
     harness
-        .apply(ConsensusInput::ProposalWithVidShare(
-            first_e2_view.leader_public_key,
-            ProposalMessage::validated(signed),
-            vid_share,
+        .apply_pair((
+            ConsensusInput::Proposal(
+                first_e2_view.leader_public_key,
+                ProposalMessage::validated(signed),
+            ),
+            ConsensusInput::VidShare(vid_share),
         ))
         .await;
     harness
@@ -544,7 +548,7 @@ async fn test_second_epoch_leader_proposes_without_drb() {
     // ---- Epoch 2 views 12-16 (blocks 12-16, before transition window) ----
     for i in 11..16 {
         harness
-            .apply(test_data.views[i].proposal_input_consensus(&node_key))
+            .apply_pair(test_data.views[i].proposal_input_consensus(&node_key))
             .await;
         harness
             .apply(test_data.views[i].block_reconstructed_input())
@@ -558,7 +562,7 @@ async fn test_second_epoch_leader_proposes_without_drb() {
     // Process view 17 (block 17, first block in transition window).
     // After cert1 for view 17, the leader for view 18 should propose.
     harness
-        .apply(test_data.views[16].proposal_input_consensus(&node_key))
+        .apply_pair(test_data.views[16].proposal_input_consensus(&node_key))
         .await;
     harness
         .apply(test_data.views[16].block_reconstructed_input())
@@ -617,10 +621,12 @@ async fn test_epoch3_transition_requests_drb_for_future_epoch() {
         .expect("VID share not found")
         .clone();
     harness
-        .apply(ConsensusInput::ProposalWithVidShare(
-            first_e2_view.leader_public_key,
-            ProposalMessage::validated(signed),
-            vid_share,
+        .apply_pair((
+            ConsensusInput::Proposal(
+                first_e2_view.leader_public_key,
+                ProposalMessage::validated(signed),
+            ),
+            ConsensusInput::VidShare(vid_share),
         ))
         .await;
     harness
@@ -632,7 +638,7 @@ async fn test_epoch3_transition_requests_drb_for_future_epoch() {
     // ---- Epoch 2 views 12-16 (before transition window) ----
     for i in 11..16 {
         harness
-            .apply(test_data.views[i].proposal_input_consensus(&node_key))
+            .apply_pair(test_data.views[i].proposal_input_consensus(&node_key))
             .await;
         harness
             .apply(test_data.views[i].block_reconstructed_input())
@@ -662,10 +668,9 @@ async fn test_epoch3_transition_requests_drb_for_future_epoch() {
         .expect("VID share not found")
         .clone();
     harness
-        .apply(ConsensusInput::ProposalWithVidShare(
-            v17.leader_public_key,
-            ProposalMessage::validated(signed),
-            vid_share,
+        .apply_pair((
+            ConsensusInput::Proposal(v17.leader_public_key, ProposalMessage::validated(signed)),
+            ConsensusInput::VidShare(vid_share),
         ))
         .await;