diff --git a/Cargo.lock b/Cargo.lock
index 10b520c0ad7..93b69a38be1 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1920,6 +1920,7 @@ dependencies = [
  "cumulus-client-network",
  "cumulus-primitives-core",
  "cumulus-test-client",
+ "cumulus-test-relay-sproof-builder",
  "cumulus-test-runtime",
  "futures",
  "parity-scale-codec 3.4.0",
@@ -1977,6 +1978,7 @@ dependencies = [
  "cumulus-primitives-core",
  "cumulus-relay-chain-interface",
  "cumulus-test-client",
+ "cumulus-test-relay-sproof-builder",
  "dyn-clone",
  "futures",
  "futures-timer",
diff --git a/client/collator/Cargo.toml b/client/collator/Cargo.toml
index 19e3578d153..95ecf67acac 100644
--- a/client/collator/Cargo.toml
+++ b/client/collator/Cargo.toml
@@ -42,3 +42,4 @@ polkadot-node-subsystem-test-helpers = { git = "https://github.com/paritytech/po
 # Cumulus
 cumulus-test-client = { path = "../../test/client" }
 cumulus-test-runtime = { path = "../../test/runtime" }
+cumulus-test-relay-sproof-builder = { path = "../../test/relay-sproof-builder" }
diff --git a/client/collator/src/lib.rs b/client/collator/src/lib.rs
index a931201f6cc..13708c4f617 100644
--- a/client/collator/src/lib.rs
+++ b/client/collator/src/lib.rs
@@ -384,10 +384,12 @@ mod tests {
 		Client, ClientBlockImportExt, DefaultTestClientBuilderExt, InitBlockBuilder,
 		TestClientBuilder, TestClientBuilderExt,
 	};
+	use cumulus_test_relay_sproof_builder::RelayStateSproofBuilder;
 	use cumulus_test_runtime::{Block, Header};
 	use futures::{channel::mpsc, executor::block_on, StreamExt};
 	use polkadot_node_subsystem_test_helpers::ForwardSubsystem;
 	use polkadot_overseer::{dummy::dummy_overseer_builder, HeadSupportsParachains};
+	use polkadot_primitives::HeadData;
 	use sp_consensus::BlockOrigin;
 	use sp_core::{testing::TaskExecutor, Pair};
 	use sp_runtime::traits::BlakeTwo256;
@@ -415,10 +417,14 @@ mod tests {
 			_: PHash,
 			validation_data: &PersistedValidationData,
 		) -> Option<ParachainCandidate<Block>> {
+			let mut sproof = RelayStateSproofBuilder::default();
+			sproof.included_para_head = Some(HeadData(parent.encode()));
+			sproof.para_id = cumulus_test_runtime::PARACHAIN_ID.into();
+
 			let builder = self.client.init_block_builder_at(
 				parent.hash(),
 				Some(validation_data.clone()),
-				Default::default(),
+				sproof,
 			);
 
 			let (block, _, proof) = builder.build().expect("Creates block").into_inner();
diff --git a/client/consensus/common/Cargo.toml b/client/consensus/common/Cargo.toml
index 12fa2099a84..c606ff27417 100644
--- a/client/consensus/common/Cargo.toml
+++ b/client/consensus/common/Cargo.toml
@@ -38,3 +38,4 @@ sp-tracing = { git = "https://github.com/paritytech/substrate", branch = "master
 
 # Cumulus
 cumulus-test-client = { path = "../../../test/client" }
+cumulus-test-relay-sproof-builder = { path = "../../../test/relay-sproof-builder" }
diff --git a/client/consensus/common/src/tests.rs b/client/consensus/common/src/tests.rs
index 23516d96388..5e65ffc53d1 100644
--- a/client/consensus/common/src/tests.rs
+++ b/client/consensus/common/src/tests.rs
@@ -28,8 +28,10 @@ use cumulus_test_client::{
 	runtime::{Block, Hash, Header},
 	Backend, Client, InitBlockBuilder, TestClientBuilder, TestClientBuilderExt,
 };
+use cumulus_test_relay_sproof_builder::RelayStateSproofBuilder;
 use futures::{channel::mpsc, executor::block_on, select, FutureExt, Stream, StreamExt};
 use futures_timer::Delay;
+use polkadot_primitives::HeadData;
 use sc_client_api::{blockchain::Backend as _, Backend as _, UsageProvider};
 use sc_consensus::{BlockImport, BlockImportParams, ForkChoiceStrategy};
 use sp_consensus::{BlockOrigin, BlockStatus};
@@ -209,17 +211,36 @@ impl RelayChainInterface for Relaychain {
 	}
 }
 
+fn sproof_with_best_parent(client: &Client) -> RelayStateSproofBuilder {
+	let best_hash = client.chain_info().best_hash;
+	sproof_with_parent_by_hash(client, best_hash)
+}
+
+fn sproof_with_parent_by_hash(client: &Client, hash: PHash) -> RelayStateSproofBuilder {
+	let header = client.header(hash).ok().flatten().expect("No header for parent block");
+	sproof_with_parent(HeadData(header.encode()))
+}
+
+fn sproof_with_parent(parent: HeadData) -> RelayStateSproofBuilder {
+	let mut x = RelayStateSproofBuilder::default();
+	x.para_id = cumulus_test_client::runtime::PARACHAIN_ID.into();
+	x.included_para_head = Some(parent);
+
+	x
+}
+
 fn build_block<B: InitBlockBuilder>(
 	builder: &B,
+	sproof: RelayStateSproofBuilder,
 	at: Option<Hash>,
 	timestamp: Option<u64>,
 ) -> Block {
 	let builder = match at {
 		Some(at) => match timestamp {
-			Some(ts) => builder.init_block_builder_with_timestamp(at, None, Default::default(), ts),
-			None => builder.init_block_builder_at(at, None, Default::default()),
+			Some(ts) => builder.init_block_builder_with_timestamp(at, None, sproof, ts),
+			None => builder.init_block_builder_at(at, None, sproof),
 		},
-		None => builder.init_block_builder(None, Default::default()),
+		None => builder.init_block_builder(None, sproof),
 	};
 
 	let mut block = builder.build().unwrap().block;
@@ -260,15 +281,20 @@ fn import_block_sync<I: BlockImport<Block>>(
 	block_on(import_block(importer, block, origin, import_as_best));
 }
 
-fn build_and_import_block_ext<B: InitBlockBuilder, I: BlockImport<Block>>(
-	builder: &B,
+fn build_and_import_block_ext<I: BlockImport<Block>>(
+	client: &Client,
 	origin: BlockOrigin,
 	import_as_best: bool,
 	importer: &mut I,
 	at: Option<Hash>,
 	timestamp: Option<u64>,
 ) -> Block {
-	let block = build_block(builder, at, timestamp);
+	let sproof = match at {
+		None => sproof_with_best_parent(client),
+		Some(at) => sproof_with_parent_by_hash(client, at),
+	};
+
+	let block = build_block(client, sproof, at, timestamp);
 	import_block_sync(importer, block.clone(), origin, import_as_best);
 	block
 }
@@ -337,7 +363,12 @@ fn follow_new_best_with_dummy_recovery_works() {
 		Some(recovery_chan_tx),
 	);
 
-	let block = build_block(&*client.clone(), None, None);
+	let sproof = {
+		let best = client.chain_info().best_hash;
+		let header = client.header(best).ok().flatten().expect("No header for best");
+		sproof_with_parent(HeadData(header.encode()))
+	};
+	let block = build_block(&*client.clone(), sproof, None, None);
 	let block_clone = block.clone();
 	let client_clone = client.clone();
 
@@ -423,7 +454,8 @@ fn follow_finalized_does_not_stop_on_unknown_block() {
 	let block = build_and_import_block(client.clone(), false);
 
 	let unknown_block = {
-		let block_builder = client.init_block_builder_at(block.hash(), None, Default::default());
+		let sproof = sproof_with_parent_by_hash(&*client, block.hash());
+		let block_builder = client.init_block_builder_at(block.hash(), None, sproof);
 		block_builder.build().unwrap().block
 	};
 
@@ -472,7 +504,8 @@ fn follow_new_best_sets_best_after_it_is_imported() {
 	let block = build_and_import_block(client.clone(), false);
 
 	let unknown_block = {
-		let block_builder = client.init_block_builder_at(block.hash(), None, Default::default());
+		let sproof = sproof_with_parent_by_hash(&*client, block.hash());
+		let block_builder = client.init_block_builder_at(block.hash(), None, sproof);
 		block_builder.build().unwrap().block
 	};
 
diff --git a/pallets/aura-ext/Cargo.toml b/pallets/aura-ext/Cargo.toml
index 1db43697511..df145aad522 100644
--- a/pallets/aura-ext/Cargo.toml
+++ b/pallets/aura-ext/Cargo.toml
@@ -18,6 +18,9 @@ sp-consensus-aura = { git = "https://github.com/paritytech/substrate", default-f
 sp-runtime = { git = "https://github.com/paritytech/substrate", default-features = false, branch = "master" }
 sp-std = { git = "https://github.com/paritytech/substrate", default-features = false, branch = "master" }
 
+# Cumulus
+cumulus-pallet-parachain-system = { path = "../parachain-system", default-features = false }
+
 [dev-dependencies]
 
 # Cumulus
@@ -35,5 +38,6 @@ std = [
 	"sp-consensus-aura/std",
 	"sp-runtime/std",
 	"sp-std/std",
+	"cumulus-pallet-parachain-system/std",
 ]
 try-runtime = ["frame-support/try-runtime"]
diff --git a/pallets/aura-ext/src/consensus_hook.rs b/pallets/aura-ext/src/consensus_hook.rs
new file mode 100644
index 00000000000..c8806b1f4cb
--- /dev/null
+++ b/pallets/aura-ext/src/consensus_hook.rs
@@ -0,0 +1,56 @@
+// Copyright 2023 Parity Technologies (UK) Ltd.
+// This file is part of Cumulus.
+
+// Cumulus is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Cumulus is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Cumulus.  If not, see <http://www.gnu.org/licenses/>.
+
+//! The definition of a [`FixedVelocityConsensusHook`] for consensus logic to manage
+//! block velocity.
+//!
+//! The velocity `V` refers to the rate of block processing by the relay chain.
+
+use super::pallet;
+use cumulus_pallet_parachain_system::{
+	consensus_hook::{ConsensusHook, UnincludedSegmentCapacity},
+	relay_state_snapshot::RelayChainStateProof,
+};
+use frame_support::pallet_prelude::*;
+use sp_std::{marker::PhantomData, num::NonZeroU32};
+
+/// A consensus hook for a fixed block processing velocity and unincluded segment capacity.
+pub struct FixedVelocityConsensusHook<T, const V: u32, const C: u32>(PhantomData<T>);
+
+impl<T: pallet::Config, const V: u32, const C: u32> ConsensusHook
+	for FixedVelocityConsensusHook<T, V, C>
+{
+	// Validates the number of authored blocks within the slot with respect to the `V + 1` limit.
+	fn on_state_proof(_state_proof: &RelayChainStateProof) -> (Weight, UnincludedSegmentCapacity) {
+		// Ensure velocity is non-zero.
+		let velocity = V.max(1);
+
+		let authored = pallet::Pallet::<T>::slot_info()
+			.map(|(_slot, authored)| authored)
+			.expect("slot info is inserted on block initialization");
+		if authored > velocity + 1 {
+			panic!("authored blocks limit is reached for the slot")
+		}
+		let weight = T::DbWeight::get().reads(1);
+
+		(
+			weight,
+			NonZeroU32::new(sp_std::cmp::max(C, 1))
+				.expect("1 is the minimum value and non-zero; qed")
+				.into(),
+		)
+	}
+}
diff --git a/pallets/aura-ext/src/lib.rs b/pallets/aura-ext/src/lib.rs
index 15e82edeefe..5605e2f2ac5 100644
--- a/pallets/aura-ext/src/lib.rs
+++ b/pallets/aura-ext/src/lib.rs
@@ -37,9 +37,12 @@
 
 use frame_support::traits::{ExecuteBlock, FindAuthor};
 use sp_application_crypto::RuntimeAppPublic;
-use sp_consensus_aura::digests::CompatibleDigestItem;
+use sp_consensus_aura::{digests::CompatibleDigestItem, Slot};
 use sp_runtime::traits::{Block as BlockT, Header as HeaderT};
 
+pub mod consensus_hook;
+pub use consensus_hook::FixedVelocityConsensusHook;
+
 type Aura<T> = pallet_aura::Pallet<T>;
 
 pub use pallet::*;
@@ -68,6 +71,19 @@ pub mod pallet {
 			// Fetch the authorities once to get them into the storage proof of the PoV.
 			Authorities::<T>::get();
 
+			let new_slot = Aura::<T>::current_slot();
+
+			let (new_slot, authored) = match SlotInfo::<T>::get() {
+				Some((slot, authored)) if slot == new_slot => (slot, authored + 1),
+				Some((slot, _)) if slot < new_slot => (new_slot, 1),
+				Some(..) => {
+					panic!("slot moved backwards")
+				},
+				None => (new_slot, 1),
+			};
+
+			SlotInfo::<T>::put((new_slot, authored));
+
 			T::DbWeight::get().reads_writes(2, 1)
 		}
 	}
@@ -84,6 +100,13 @@ pub mod pallet {
 		ValueQuery,
 	>;
 
+	/// Current slot paired with a number of authored blocks.
+	///
+	/// Updated on each block initialization.
+	#[pallet::storage]
+	#[pallet::getter(fn slot_info)]
+	pub(crate) type SlotInfo<T: Config> = StorageValue<_, (Slot, u32), OptionQuery>;
+
 	#[pallet::genesis_config]
 	#[derive(Default)]
 	pub struct GenesisConfig;
diff --git a/pallets/parachain-system/src/consensus_hook.rs b/pallets/parachain-system/src/consensus_hook.rs
new file mode 100644
index 00000000000..bdf590a0fd5
--- /dev/null
+++ b/pallets/parachain-system/src/consensus_hook.rs
@@ -0,0 +1,109 @@
+// Copyright 2023 Parity Technologies (UK) Ltd.
+// This file is part of Cumulus.
+
+// Cumulus is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Cumulus is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Cumulus.  If not, see <http://www.gnu.org/licenses/>.
+
+//! The definition of a [`ConsensusHook`] trait for consensus logic to manage the backlog
+//! of parachain blocks ready to submit to the relay chain, as well as some basic implementations.
+
+use super::relay_state_snapshot::RelayChainStateProof;
+use frame_support::weights::Weight;
+use sp_std::num::NonZeroU32;
+
+/// The possible capacity of the unincluded segment.
+#[derive(Clone)]
+pub struct UnincludedSegmentCapacity(UnincludedSegmentCapacityInner);
+
+impl UnincludedSegmentCapacity {
+	pub(crate) fn get(&self) -> u32 {
+		match self.0 {
+			UnincludedSegmentCapacityInner::ExpectParentIncluded => 1,
+			UnincludedSegmentCapacityInner::Value(v) => v.get(),
+		}
+	}
+
+	pub(crate) fn is_expecting_included_parent(&self) -> bool {
+		match self.0 {
+			UnincludedSegmentCapacityInner::ExpectParentIncluded => true,
+			UnincludedSegmentCapacityInner::Value(_) => false,
+		}
+	}
+}
+
+#[derive(Clone)]
+pub(crate) enum UnincludedSegmentCapacityInner {
+	ExpectParentIncluded,
+	Value(NonZeroU32),
+}
+
+impl From<NonZeroU32> for UnincludedSegmentCapacity {
+	fn from(value: NonZeroU32) -> Self {
+		UnincludedSegmentCapacity(UnincludedSegmentCapacityInner::Value(value))
+	}
+}
+
+/// The consensus hook for dealing with the unincluded segment.
+///
+/// Higher-level and user-configurable consensus logic is more informed about the
+/// desired unincluded segment length, as well as any rules for adapting it dynamically
+/// according to the relay-chain state.
+pub trait ConsensusHook {
+	/// This hook is called partway through the `set_validation_data` inherent in parachain-system.
+	///
+	/// The hook is allowed to panic if customized consensus rules aren't met and is required
+	/// to return a maximum capacity for the unincluded segment with weight consumed.
+	fn on_state_proof(state_proof: &RelayChainStateProof) -> (Weight, UnincludedSegmentCapacity);
+}
+
+/// A special consensus hook for handling the migration to asynchronous backing gracefully,
+/// even if collators haven't been updated to provide the last included parent in the state
+/// proof yet.
+///
+/// This behaves as though the parent is included, even if the relay chain state proof doesn't contain
+/// the included para head. If the para head is present in the state proof, this does ensure the
+/// parent is included.
+pub struct ExpectParentIncluded;
+
+impl ConsensusHook for ExpectParentIncluded {
+	fn on_state_proof(_state_proof: &RelayChainStateProof) -> (Weight, UnincludedSegmentCapacity) {
+		(
+			Weight::zero(),
+			UnincludedSegmentCapacity(UnincludedSegmentCapacityInner::ExpectParentIncluded),
+		)
+	}
+}
+
+/// A consensus hook for a fixed unincluded segment length. This hook does nothing but
+/// set the capacity of the unincluded segment to the constant N.
+///
+/// Since it is illegal to provide an unincluded segment length of 0, this sets a minimum of
+/// 1.
+pub struct FixedCapacityUnincludedSegment<const N: u32>;
+
+impl<const N: u32> ConsensusHook for FixedCapacityUnincludedSegment<N> {
+	fn on_state_proof(_state_proof: &RelayChainStateProof) -> (Weight, UnincludedSegmentCapacity) {
+		(
+			Weight::zero(),
+			NonZeroU32::new(sp_std::cmp::max(N, 1))
+				.expect("1 is the minimum value and non-zero; qed")
+				.into(),
+		)
+	}
+}
+
+/// A fixed-capacity unincluded segment hook, which requires that the parent block is
+/// included prior to the current block being authored.
+///
+/// This is a simple type alias around a fixed-capacity unincluded segment with a size of 1.
+pub type RequireParentIncluded = FixedCapacityUnincludedSegment<1>;
diff --git a/pallets/parachain-system/src/lib.rs b/pallets/parachain-system/src/lib.rs
index 3edcc6dfb14..8c8a57107dd 100644
--- a/pallets/parachain-system/src/lib.rs
+++ b/pallets/parachain-system/src/lib.rs
@@ -48,7 +48,7 @@ use frame_system::{ensure_none, ensure_root};
 use polkadot_parachain::primitives::RelayChainBlockNumber;
 use scale_info::TypeInfo;
 use sp_runtime::{
-	traits::{Block as BlockT, BlockNumberProvider, Hash, Zero},
+	traits::{Block as BlockT, BlockNumberProvider, Hash},
 	transaction_validity::{
 		InvalidTransaction, TransactionLongevity, TransactionSource, TransactionValidity,
 		ValidTransaction,
@@ -58,17 +58,21 @@ use sp_std::{cmp, collections::btree_map::BTreeMap, prelude::*};
 use xcm::latest::XcmHash;
 
 mod migration;
-mod relay_state_snapshot;
+#[cfg(test)]
+mod tests;
 mod unincluded_segment;
+
+pub mod consensus_hook;
+pub mod relay_state_snapshot;
 #[macro_use]
 pub mod validate_block;
-#[cfg(test)]
-mod tests;
 
 use unincluded_segment::{
-	Ancestor, HrmpChannelUpdate, SegmentTracker, TotalBandwidthLimits, UsedBandwidth,
+	Ancestor, HrmpChannelUpdate, HrmpWatermarkUpdate, OutboundBandwidthLimits, SegmentTracker,
+	UsedBandwidth,
 };
 
+pub use consensus_hook::ConsensusHook;
 /// Register the `validate_block` function that is used by parachains to validate blocks on a
 /// validator.
 ///
@@ -198,6 +202,18 @@ pub mod pallet {
 
 		/// Something that can check the associated relay parent block number.
 		type CheckAssociatedRelayNumber: CheckAssociatedRelayNumber;
+
+		/// An entry-point for higher-level logic to manage the backlog of unincluded parachain blocks
+		/// and authorship rights for those blocks.
+		///
+		/// Typically, this should be a hook tailored to the collator-selection/consensus mechanism
+		/// that is used for this chain.
+		///
+		/// However, to maintain the same behavior as prior to asynchronous backing, provide the
+		/// [`consensus_hook::ExpectParentIncluded`] here. This is only necessary in the case
+		/// that collators aren't expected to have node versions that supply the included block
+		/// in the relay-chain state proof.
+		type ConsensusHook: ConsensusHook;
 	}
 
 	#[pallet::hooks]
@@ -237,14 +253,25 @@ pub mod pallet {
 				},
 			};
 
-			let (ump_msg_count, ump_total_bytes) = <PendingUpwardMessages<T>>::mutate(|up| {
-				let (count, size) = relevant_messaging_state.relay_dispatch_queue_size;
+			let total_bandwidth_out = OutboundBandwidthLimits::from_relay_chain_state(
+				&relevant_messaging_state,
+				host_config.max_upward_queue_count,
+				host_config.max_upward_queue_size,
+			);
+			let bandwidth_out = AggregatedUnincludedSegment::<T>::get().map(|segment| {
+				let mut bandwidth_out = total_bandwidth_out.clone();
+				bandwidth_out.subtract(segment.used_bandwidth());
+				bandwidth_out
+			});
 
+			let (ump_msg_count, ump_total_bytes) = <PendingUpwardMessages<T>>::mutate(|up| {
+				let bandwidth_out = bandwidth_out.as_ref().unwrap_or(&total_bandwidth_out);
 				let available_capacity = cmp::min(
-					host_config.max_upward_queue_count.saturating_sub(count),
+					bandwidth_out.ump_messages_remaining,
 					host_config.max_upward_message_num_per_candidate,
 				);
-				let available_size = host_config.max_upward_queue_size.saturating_sub(size);
+
+				let available_size = bandwidth_out.ump_bytes_remaining;
 
 				// Count the number of messages we can possibly fit in the given constraints, i.e.
 				// available_capacity and available_size.
@@ -289,23 +316,17 @@ pub mod pallet {
 				.hrmp_max_message_num_per_candidate
 				.min(<AnnouncedHrmpMessagesPerCandidate<T>>::take()) as usize;
 
+			// TODO [now]: the `ChannelInfo` implementation for this pallet is what's
+			// important here for proper limiting.
 			let outbound_messages =
 				T::OutboundXcmpMessageSource::take_outbound_messages(maximum_channels)
 					.into_iter()
 					.map(|(recipient, data)| OutboundHrmpMessage { recipient, data })
 					.collect::<Vec<_>>();
 
-			if MaxUnincludedLen::<T>::get().map_or(false, |max_len| !max_len.is_zero()) {
-				// NOTE: these limits don't account for the amount of processed messages from
-				// downward and horizontal queues.
-				//
-				// This is correct because:
-				// - inherent never contains messages that were previously processed.
-				// - current implementation always attempts to exhaust each message queue.
-				//
-				// <https://github.com/paritytech/cumulus/issues/2472>
-				let limits = TotalBandwidthLimits::new(&relevant_messaging_state);
-
+			// Update the unincluded segment length; capacity checks were done previously in
+			// `set_validation_data`, so this can be done unconditionally.
+			{
 				let hrmp_outgoing = outbound_messages
 					.iter()
 					.map(|msg| {
@@ -321,12 +342,14 @@ pub mod pallet {
 				let ancestor = Ancestor::new_unchecked(used_bandwidth);
 
 				let watermark = HrmpWatermark::<T>::get();
+				let watermark_update =
+					HrmpWatermarkUpdate::new(watermark, LastRelayChainBlockNumber::<T>::get());
 				AggregatedUnincludedSegment::<T>::mutate(|agg| {
 					let agg = agg.get_or_insert_with(SegmentTracker::default);
 					// TODO: In order of this panic to be correct, outbound message source should
 					// respect bandwidth limits as well.
 					// <https://github.com/paritytech/cumulus/issues/2471>
-					agg.append(&ancestor, watermark, &limits)
+					agg.append(&ancestor, watermark_update, &total_bandwidth_out)
 						.expect("unincluded segment limits exceeded");
 				});
 				// Check in `on_initialize` guarantees there's space for this block.
@@ -346,8 +369,8 @@ pub mod pallet {
 				weight += T::DbWeight::get().writes(1);
 			}
 
-			// New para head was unknown during block finalization, update it.
-			if MaxUnincludedLen::<T>::get().map_or(false, |max_len| !max_len.is_zero()) {
+			// The parent hash was unknown during block finalization. Update it here.
+			{
 				<UnincludedSegment<T>>::mutate(|chain| {
 					if let Some(ancestor) = chain.last_mut() {
 						let parent = frame_system::Pallet::<T>::parent_hash();
@@ -359,9 +382,8 @@ pub mod pallet {
 				weight += T::DbWeight::get().reads_writes(1, 1);
 
 				// Weight used during finalization.
-				weight += T::DbWeight::get().reads_writes(2, 2);
+				weight += T::DbWeight::get().reads_writes(3, 2);
 			}
-			weight += T::DbWeight::get().reads(1);
 
 			// Remove the validation from the old block.
 			ValidationData::<T>::kill();
@@ -461,6 +483,10 @@ pub mod pallet {
 			)
 			.expect("Invalid relay chain state proof");
 
+			// Update the desired maximum capacity according to the consensus hook.
+			let (consensus_hook_weight, capacity) =
+				T::ConsensusHook::on_state_proof(&relay_state_proof);
+
 			// initialization logic: we know that this runs exactly once every block,
 			// which means we can put the initialization logic here to remove the
 			// sequencing problem.
@@ -508,7 +534,17 @@ pub mod pallet {
 			<T::OnSystemEvent as OnSystemEvent>::on_validation_data(&vfp);
 
 			// TODO: This is more than zero, but will need benchmarking to figure out what.
-			let mut total_weight = Weight::zero();
+			// NOTE: We don't account for the amount of processed messages from
+			// downward and horizontal channels in the unincluded segment.
+			//
+			// This is correct only because the  current implementation always attempts
+			// to exhaust each message queue and panics if the DMQ head doesn't match.
+			//
+			// If one or more messages were ever "re-processed" in a parachain block before its
+			// ancestor was included, the MQC heads wouldn't match and the block would be invalid.
+			//
+			// <https://github.com/paritytech/cumulus/issues/2472>
+			let mut total_weight = consensus_hook_weight;
 			total_weight += Self::process_inbound_downward_messages(
 				relevant_messaging_state.dmq_mqc_head,
 				downward_messages,
@@ -518,7 +554,7 @@ pub mod pallet {
 				horizontal_messages,
 				vfp.relay_parent_number,
 			);
-			total_weight += Self::maybe_drop_included_ancestors(&relay_state_proof);
+			total_weight += Self::maybe_drop_included_ancestors(&relay_state_proof, capacity);
 
 			Ok(PostDispatchInfo { actual_weight: Some(total_weight), pays_fee: Pays::No })
 		}
@@ -621,18 +657,12 @@ pub mod pallet {
 		Unauthorized,
 	}
 
-	/// Maximum number of latest included block descendants the runtime is allowed to accept. In other words,
-	/// these are ancestor of the block being currently executed, not yet sent to the relay chain runtime.
-	///
-	/// This value is optional, but once set to `Some` by the governance, should never go back to `None`.
-	/// Requires latest included para head to be present in the relay chain storage proof.
-	#[pallet::storage]
-	pub(super) type MaxUnincludedLen<T: Config> = StorageValue<_, T::BlockNumber, OptionQuery>;
-
 	/// Latest included block descendants the runtime accepted. In other words, these are
-	/// ancestors of the block being currently executed, not yet sent to the relay chain runtime.
+	/// ancestors of the currently executing block which have not been included in the observed
+	/// relay-chain state.
 	///
-	/// The segment length is limited by [`MaxUnincludedLen`].
+	/// The segment length is limited by the capacity returned from the [`ConsensusHook`] configured
+	/// in the pallet.
 	#[pallet::storage]
 	pub(super) type UnincludedSegment<T: Config> =
 		StorageValue<_, Vec<Ancestor<T::Hash>>, ValueQuery>;
@@ -1061,53 +1091,57 @@ impl<T: Config> Pallet<T> {
 	}
 
 	/// Drop blocks from the unincluded segment with respect to the latest parachain head.
-	///
-	/// No-op if [`MaxUnincludedLen`] is not set.
-	fn maybe_drop_included_ancestors(relay_state_proof: &RelayChainStateProof) -> Weight {
+	fn maybe_drop_included_ancestors(
+		relay_state_proof: &RelayChainStateProof,
+		capacity: consensus_hook::UnincludedSegmentCapacity,
+	) -> Weight {
 		let mut weight_used = Weight::zero();
-		// If `MaxUnincludedLen` is present in the storage, parachain head
-		// is always expected to be included into the relay storage proof.
-		let para_head_with_len = <MaxUnincludedLen<T>>::get().map(|max_len| {
-			(
-				relay_state_proof
-					.read_included_para_head()
-					.expect("Invalid para head in relay chain state proof"),
-				max_len,
-			)
-		});
+		// If the unincluded segment length is nonzero, then the parachain head must be present.
+		let para_head =
+			relay_state_proof.read_included_para_head().ok().map(|h| T::Hashing::hash(&h.0));
+
+		let unincluded_segment_len = <UnincludedSegment<T>>::decode_len().unwrap_or(0);
 		weight_used += T::DbWeight::get().reads(1);
-		let Some((para_head, max_len)) = para_head_with_len else { return weight_used };
 
-		let para_head_hash = T::Hashing::hash(&para_head.0);
-		if !max_len.is_zero() {
-			let (dropped, left_count): (Vec<Ancestor<T::Hash>>, u32) =
-				<UnincludedSegment<T>>::mutate(|chain| {
-					// Drop everything up to the block with an included para head, if present.
-					let idx = chain
-						.iter()
-						.position(|block| {
-							let head_hash = block.para_head_hash().expect(
-								"para head hash is updated during block initialization; qed",
-							);
-							head_hash == &para_head_hash
-						})
-						.map_or(0, |idx| idx + 1); // inclusive.
-
-					let left_count = (idx..chain.len()).count() as u32;
-					let dropped = chain.drain(..idx).collect();
-					(dropped, left_count)
-				});
-			weight_used += T::DbWeight::get().reads_writes(1, 1);
+		// Clean up unincluded segment if nonempty.
+		let included_head = match (para_head, capacity.is_expecting_included_parent()) {
+			(Some(h), true) => {
+				assert_eq!(
+					h,
+					frame_system::Pallet::<T>::parent_hash(),
+					"expected parent to be included"
+				);
 
-			// sanity-check there's place for the block at finalization phase.
-			//
-			// If this fails, the max segment len is reached and parachain should wait
-			// for ancestor's inclusion.
-			assert!(
-				max_len > left_count.into(),
-				"no space left for the block in the unincluded segment"
-			);
+				h
+			},
+			(Some(h), false) => h,
+			(None, true) => {
+				// All this logic is essentially a workaround to support collators which
+				// might still not provide the included block with the state proof.
+				frame_system::Pallet::<T>::parent_hash()
+			},
+			(None, false) => panic!("included head not present in relay storage proof"),
+		};
+
+		let new_len = {
+			let para_head_hash = included_head;
+			let dropped: Vec<Ancestor<T::Hash>> = <UnincludedSegment<T>>::mutate(|chain| {
+				// Drop everything up to (inclusive) the block with an included para head, if present.
+				let idx = chain
+					.iter()
+					.position(|block| {
+						let head_hash = block
+							.para_head_hash()
+							.expect("para head hash is updated during block initialization; qed");
+						head_hash == &para_head_hash
+					})
+					.map_or(0, |idx| idx + 1); // inclusive.
+
+				chain.drain(..idx).collect()
+			});
+			weight_used += T::DbWeight::get().reads_writes(1, 1);
 
+			let new_len = unincluded_segment_len - dropped.len();
 			if !dropped.is_empty() {
 				<AggregatedUnincludedSegment<T>>::mutate(|agg| {
 					let agg = agg.as_mut().expect(
@@ -1119,7 +1153,15 @@ impl<T: Config> Pallet<T> {
 				});
 				weight_used += T::DbWeight::get().reads_writes(1, 1);
 			}
-		}
+
+			new_len as u32
+		};
+
+		// Current block validity check: ensure there is space in the unincluded segment.
+		//
+		// If this fails, the parachain needs to wait for ancestors to be included before
+		// a new block is allowed.
+		assert!(new_len < capacity.get(), "no space left for the block in the unincluded segment");
 		weight_used
 	}
 
diff --git a/pallets/parachain-system/src/relay_state_snapshot.rs b/pallets/parachain-system/src/relay_state_snapshot.rs
index 9da5a03ce83..ead077f527d 100644
--- a/pallets/parachain-system/src/relay_state_snapshot.rs
+++ b/pallets/parachain-system/src/relay_state_snapshot.rs
@@ -14,6 +14,8 @@
 // You should have received a copy of the GNU General Public License
 // along with Cumulus.  If not, see <http://www.gnu.org/licenses/>.
 
+//! Relay chain state proof provides means for accessing part of relay chain storage for reads.
+
 use codec::{Decode, Encode};
 use cumulus_primitives_core::{
 	relay_chain, AbridgedHostConfiguration, AbridgedHrmpChannel, ParaId,
diff --git a/pallets/parachain-system/src/tests.rs b/pallets/parachain-system/src/tests.rs
index 8edbabaf5a7..574ab43078d 100755
--- a/pallets/parachain-system/src/tests.rs
+++ b/pallets/parachain-system/src/tests.rs
@@ -38,11 +38,12 @@ use sp_runtime::{
 	traits::{BlakeTwo256, IdentityLookup},
 	DispatchErrorWithPostInfo,
 };
-use sp_std::collections::vec_deque::VecDeque;
+use sp_std::{collections::vec_deque::VecDeque, num::NonZeroU32};
 use sp_version::RuntimeVersion;
 use std::cell::RefCell;
 
 use crate as parachain_system;
+use crate::consensus_hook::UnincludedSegmentCapacity;
 
 type UncheckedExtrinsic = frame_system::mocking::MockUncheckedExtrinsic<Test>;
 type Block = frame_system::mocking::MockBlock<Test>;
@@ -110,6 +111,7 @@ impl Config for Test {
 	type XcmpMessageHandler = SaveIntoThreadLocal;
 	type ReservedXcmpWeight = ReservedXcmpWeight;
 	type CheckAssociatedRelayNumber = RelayNumberStrictlyIncreases;
+	type ConsensusHook = TestConsensusHook;
 }
 
 pub struct FromThreadLocal;
@@ -119,6 +121,16 @@ std::thread_local! {
 	static HANDLED_DMP_MESSAGES: RefCell<Vec<(relay_chain::BlockNumber, Vec<u8>)>> = RefCell::new(Vec::new());
 	static HANDLED_XCMP_MESSAGES: RefCell<Vec<(ParaId, relay_chain::BlockNumber, Vec<u8>)>> = RefCell::new(Vec::new());
 	static SENT_MESSAGES: RefCell<Vec<(ParaId, Vec<u8>)>> = RefCell::new(Vec::new());
+	static CONSENSUS_HOOK: RefCell<Box<dyn Fn(&RelayChainStateProof) -> (Weight, UnincludedSegmentCapacity)>>
+		= RefCell::new(Box::new(|_| (Weight::zero(), NonZeroU32::new(1).unwrap().into())));
+}
+
+pub struct TestConsensusHook;
+
+impl ConsensusHook for TestConsensusHook {
+	fn on_state_proof(s: &RelayChainStateProof) -> (Weight, UnincludedSegmentCapacity) {
+		CONSENSUS_HOOK.with(|f| f.borrow_mut()(s))
+	}
 }
 
 fn send_message(dest: ParaId, message: Vec<u8>) {
@@ -233,7 +245,6 @@ struct BlockTests {
 	inherent_data_hook:
 		Option<Box<dyn Fn(&BlockTests, RelayChainBlockNumber, &mut ParachainInherentData)>>,
 	inclusion_delay: Option<usize>,
-	max_unincluded_len: Option<u64>,
 
 	included_para_head: Option<relay_chain::HeadData>,
 	pending_blocks: VecDeque<relay_chain::HeadData>,
@@ -297,9 +308,8 @@ impl BlockTests {
 		self
 	}
 
-	fn with_unincluded_segment(mut self, inclusion_delay: usize, max_unincluded_len: u64) -> Self {
+	fn with_inclusion_delay(mut self, inclusion_delay: usize) -> Self {
 		self.inclusion_delay.replace(inclusion_delay);
-		self.max_unincluded_len.replace(max_unincluded_len);
 		self
 	}
 
@@ -311,11 +321,8 @@ impl BlockTests {
 				relay_chain::HeadData(header.encode())
 			};
 
-			if let Some(max_unincluded_len) = self.max_unincluded_len {
-				// Initialize included head if the segment is enabled.
-				self.included_para_head.replace(parent_head_data.clone());
-				<MaxUnincludedLen<Test>>::put(max_unincluded_len);
-			}
+			self.included_para_head = Some(parent_head_data.clone());
+
 			for BlockTest { n, within_block, after_block } in self.tests.iter() {
 				// clear pending updates, as applicable
 				if let Some(upgrade_block) = self.pending_upgrade {
@@ -433,8 +440,12 @@ fn block_tests_run_on_drop() {
 
 #[test]
 fn unincluded_segment_works() {
+	CONSENSUS_HOOK.with(|c| {
+		*c.borrow_mut() = Box::new(|_| (Weight::zero(), NonZeroU32::new(10).unwrap().into()))
+	});
+
 	BlockTests::new()
-		.with_unincluded_segment(1, 10)
+		.with_inclusion_delay(1)
 		.add_with_post_test(
 			123,
 			|| {},
@@ -464,10 +475,14 @@ fn unincluded_segment_works() {
 }
 
 #[test]
-#[should_panic]
+#[should_panic = "no space left for the block in the unincluded segment"]
 fn unincluded_segment_is_limited() {
+	CONSENSUS_HOOK.with(|c| {
+		*c.borrow_mut() = Box::new(|_| (Weight::zero(), NonZeroU32::new(1).unwrap().into()))
+	});
+
 	BlockTests::new()
-		.with_unincluded_segment(10, 1)
+		.with_inclusion_delay(2)
 		.add_with_post_test(
 			123,
 			|| {},
@@ -624,7 +639,10 @@ fn send_upward_message_num_per_candidate() {
 		)
 		.add_with_post_test(
 			2,
-			|| { /* do nothing within block */ },
+			|| {
+				assert_eq!(UnincludedSegment::<Test>::get().len(), 0);
+				/* do nothing within block */
+			},
 			|| {
 				let v = UpwardMessages::<Test>::get();
 				assert_eq!(v, vec![b"message 2".to_vec()]);
diff --git a/pallets/parachain-system/src/unincluded_segment.rs b/pallets/parachain-system/src/unincluded_segment.rs
index d0e5dd47f61..a61c5725556 100644
--- a/pallets/parachain-system/src/unincluded_segment.rs
+++ b/pallets/parachain-system/src/unincluded_segment.rs
@@ -27,6 +27,7 @@ use scale_info::TypeInfo;
 use sp_std::{collections::btree_map::BTreeMap, marker::PhantomData};
 
 /// Constraints on outbound HRMP channel.
+#[derive(Clone)]
 pub struct HrmpOutboundLimits {
 	/// The maximum bytes that can be written to the channel.
 	pub bytes_remaining: u32,
@@ -34,8 +35,9 @@ pub struct HrmpOutboundLimits {
 	pub messages_remaining: u32,
 }
 
-/// Constraints imposed on the entire segment, i.e. based on the latest included parablock.
-pub struct TotalBandwidthLimits {
+/// Limits on outbound message bandwidth.
+#[derive(Clone)]
+pub struct OutboundBandwidthLimits {
 	/// The amount of UMP messages remaining.
 	pub ump_messages_remaining: u32,
 	/// The amount of UMP bytes remaining.
@@ -44,11 +46,23 @@ pub struct TotalBandwidthLimits {
 	pub hrmp_outgoing: BTreeMap<ParaId, HrmpOutboundLimits>,
 }
 
-impl TotalBandwidthLimits {
-	/// Creates new limits from the messaging state.
-	pub fn new(messaging_state: &MessagingStateSnapshot) -> Self {
-		let (ump_messages_remaining, ump_bytes_remaining) =
-			messaging_state.relay_dispatch_queue_size;
+impl OutboundBandwidthLimits {
+	/// Creates new limits from the messaging state and upward message queue maximums fetched
+	/// from the host configuration.
+	///
+	/// These will be the total bandwidth limits across the entire unincluded segment.
+	pub fn from_relay_chain_state(
+		messaging_state: &MessagingStateSnapshot,
+		max_upward_queue_count: u32,
+		max_upward_queue_size: u32,
+	) -> Self {
+		let (ump_messages_in_relay, ump_bytes_in_relay) = messaging_state.relay_dispatch_queue_size;
+
+		let (ump_messages_remaining, ump_bytes_remaining) = (
+			max_upward_queue_count.saturating_sub(ump_messages_in_relay),
+			max_upward_queue_size.saturating_sub(ump_bytes_in_relay),
+		);
+
 		let hrmp_outgoing = messaging_state
 			.egress_channels
 			.iter()
@@ -56,8 +70,8 @@ impl TotalBandwidthLimits {
 				(
 					*id,
 					HrmpOutboundLimits {
-						bytes_remaining: channel.max_total_size,
-						messages_remaining: channel.max_capacity,
+						bytes_remaining: channel.max_total_size.saturating_sub(channel.total_size),
+						messages_remaining: channel.max_capacity.saturating_sub(channel.msg_count),
 					},
 				)
 			})
@@ -65,6 +79,21 @@ impl TotalBandwidthLimits {
 
 		Self { ump_messages_remaining, ump_bytes_remaining, hrmp_outgoing }
 	}
+
+	/// Compute the remaining bandwidth when accounting for the used amounts provided.
+	pub fn subtract(&mut self, used: &UsedBandwidth) {
+		self.ump_messages_remaining =
+			self.ump_messages_remaining.saturating_sub(used.ump_msg_count);
+		self.ump_bytes_remaining = self.ump_bytes_remaining.saturating_sub(used.ump_total_bytes);
+		for (para_id, channel_limits) in self.hrmp_outgoing.iter_mut() {
+			if let Some(update) = used.hrmp_outgoing.get(para_id) {
+				channel_limits.bytes_remaining =
+					channel_limits.bytes_remaining.saturating_sub(update.total_bytes);
+				channel_limits.messages_remaining =
+					channel_limits.messages_remaining.saturating_sub(update.msg_count);
+			}
+		}
+	}
 }
 
 /// The error type for updating bandwidth used by a segment.
@@ -131,7 +160,7 @@ impl HrmpChannelUpdate {
 		&self,
 		other: &Self,
 		recipient: ParaId,
-		limits: &TotalBandwidthLimits,
+		limits: &OutboundBandwidthLimits,
 	) -> Result<Self, BandwidthUpdateError> {
 		let limits = limits
 			.hrmp_outgoing
@@ -186,7 +215,7 @@ impl UsedBandwidth {
 	fn append(
 		&self,
 		other: &Self,
-		limits: &TotalBandwidthLimits,
+		limits: &OutboundBandwidthLimits,
 	) -> Result<Self, BandwidthUpdateError> {
 		let mut new = self.clone();
 
@@ -263,6 +292,34 @@ impl<H> Ancestor<H> {
 	}
 }
 
+/// An update to the HRMP watermark. This is always a relay-chain block number,
+/// but the two variants have different semantic meanings.
+pub enum HrmpWatermarkUpdate {
+	/// An update to the HRMP watermark where the new value is set to be equal to the
+	/// relay-parent's block number, i.e. the "head" of the relay chain.
+	/// This is always legal.
+	Head(relay_chain::BlockNumber),
+	/// An update to the HRMP watermark where the new value falls into the "trunk" of the
+	/// relay-chain. In this case, the watermark must be greater than the previous value.
+	Trunk(relay_chain::BlockNumber),
+}
+
+impl HrmpWatermarkUpdate {
+	/// Create a new update based on the desired watermark value and the current
+	/// relay-parent number.
+	pub fn new(
+		watermark: relay_chain::BlockNumber,
+		relay_parent_number: relay_chain::BlockNumber,
+	) -> Self {
+		// Hard constrain the watermark to the relay-parent number.
+		if watermark >= relay_parent_number {
+			HrmpWatermarkUpdate::Head(relay_parent_number)
+		} else {
+			HrmpWatermarkUpdate::Trunk(watermark)
+		}
+	}
+}
+
 /// Struct that keeps track of bandwidth used by the unincluded part of the chain
 /// along with the latest HRMP watermark.
 #[derive(Default, Encode, Decode, TypeInfo)]
@@ -277,23 +334,29 @@ pub struct SegmentTracker<H> {
 
 impl<H> SegmentTracker<H> {
 	/// Tries to append another block to the tracker, respecting given bandwidth limits.
+	/// In practice, the bandwidth limits supplied should be the total allowed within the
+	/// block.
 	pub fn append(
 		&mut self,
 		block: &Ancestor<H>,
-		hrmp_watermark: relay_chain::BlockNumber,
-		limits: &TotalBandwidthLimits,
+		new_watermark: HrmpWatermarkUpdate,
+		limits: &OutboundBandwidthLimits,
 	) -> Result<(), BandwidthUpdateError> {
 		if let Some(watermark) = self.hrmp_watermark.as_ref() {
-			if &hrmp_watermark <= watermark {
-				return Err(BandwidthUpdateError::InvalidHrmpWatermark {
-					submitted: hrmp_watermark,
-					latest: *watermark,
-				})
+			if let HrmpWatermarkUpdate::Trunk(new) = new_watermark {
+				if &new <= watermark {
+					return Err(BandwidthUpdateError::InvalidHrmpWatermark {
+						submitted: new,
+						latest: *watermark,
+					})
+				}
 			}
 		}
 
 		self.used_bandwidth = self.used_bandwidth.append(block.used_bandwidth(), limits)?;
-		self.hrmp_watermark.replace(hrmp_watermark);
+		self.hrmp_watermark.replace(match new_watermark {
+			HrmpWatermarkUpdate::Trunk(w) | HrmpWatermarkUpdate::Head(w) => w,
+		});
 
 		Ok(())
 	}
@@ -304,6 +367,11 @@ impl<H> SegmentTracker<H> {
 		// Watermark doesn't need to be updated since the is always dropped
 		// from the tail of the segment.
 	}
+
+	/// Return a reference to the used bandwidth across the entire segment.
+	pub fn used_bandwidth(&self) -> &UsedBandwidth {
+		&self.used_bandwidth
+	}
 }
 
 #[cfg(test)]
@@ -311,6 +379,65 @@ mod tests {
 	use super::*;
 	use assert_matches::assert_matches;
 
+	#[test]
+	fn outbound_limits_constructed_correctly() {
+		let para_a = ParaId::from(0);
+		let para_a_channel = relay_chain::AbridgedHrmpChannel {
+			max_message_size: 15,
+
+			// Msg count capacity left is 2.
+			msg_count: 5,
+			max_capacity: 7,
+
+			// Bytes capacity left is 10.
+			total_size: 50,
+			max_total_size: 60,
+			mqc_head: None,
+		};
+
+		let para_b = ParaId::from(1);
+		let para_b_channel = relay_chain::AbridgedHrmpChannel {
+			max_message_size: 15,
+
+			// Msg count capacity left is 10.
+			msg_count: 40,
+			max_capacity: 50,
+
+			// Bytes capacity left is 0.
+			total_size: 500,
+			max_total_size: 500,
+			mqc_head: None,
+		};
+		let messaging_state = MessagingStateSnapshot {
+			dmq_mqc_head: relay_chain::Hash::zero(),
+			// (msg_count, bytes)
+			relay_dispatch_queue_size: (10, 100),
+			ingress_channels: Vec::new(),
+
+			egress_channels: vec![(para_a, para_a_channel), (para_b, para_b_channel)],
+		};
+
+		let max_upward_queue_count = 11;
+		let max_upward_queue_size = 150;
+		let limits = OutboundBandwidthLimits::from_relay_chain_state(
+			&messaging_state,
+			max_upward_queue_count,
+			max_upward_queue_size,
+		);
+
+		// UMP.
+		assert_eq!(limits.ump_messages_remaining, 1);
+		assert_eq!(limits.ump_bytes_remaining, 50);
+
+		// HRMP.
+		let para_a_limits = limits.hrmp_outgoing.get(&para_a).expect("channel must be present");
+		let para_b_limits = limits.hrmp_outgoing.get(&para_b).expect("channel must be present");
+		assert_eq!(para_a_limits.bytes_remaining, 10);
+		assert_eq!(para_a_limits.messages_remaining, 2);
+		assert_eq!(para_b_limits.bytes_remaining, 0);
+		assert_eq!(para_b_limits.messages_remaining, 10);
+	}
+
 	#[test]
 	fn hrmp_msg_count_limits() {
 		let para_0 = ParaId::from(0);
@@ -319,7 +446,7 @@ mod tests {
 		let para_1 = ParaId::from(1);
 		let para_1_limits = HrmpOutboundLimits { bytes_remaining: u32::MAX, messages_remaining: 3 };
 		let hrmp_outgoing = [(para_0, para_0_limits), (para_1, para_1_limits)].into();
-		let limits = TotalBandwidthLimits {
+		let limits = OutboundBandwidthLimits {
 			ump_messages_remaining: 0,
 			ump_bytes_remaining: 0,
 			hrmp_outgoing,
@@ -331,7 +458,7 @@ mod tests {
 		for _ in 0..5 {
 			hrmp_update = hrmp_update
 				.append(&HrmpChannelUpdate { msg_count: 1, total_bytes: 10 }, para_0, &limits)
-				.expect("update is withing the limits");
+				.expect("update is within the limits");
 		}
 		assert_matches!(
 			hrmp_update.append(
@@ -349,7 +476,7 @@ mod tests {
 		let mut hrmp_update = HrmpChannelUpdate::default();
 		hrmp_update = hrmp_update
 			.append(&HrmpChannelUpdate { msg_count: 2, total_bytes: 10 }, para_1, &limits)
-			.expect("update is withing the limits");
+			.expect("update is within the limits");
 		assert_matches!(
 			hrmp_update.append(
 				&HrmpChannelUpdate { msg_count: 3, total_bytes: 10 },
@@ -371,7 +498,7 @@ mod tests {
 			HrmpOutboundLimits { bytes_remaining: 25, messages_remaining: u32::MAX };
 
 		let hrmp_outgoing = [(para_0, para_0_limits)].into();
-		let limits = TotalBandwidthLimits {
+		let limits = OutboundBandwidthLimits {
 			ump_messages_remaining: 0,
 			ump_bytes_remaining: 0,
 			hrmp_outgoing,
@@ -383,7 +510,7 @@ mod tests {
 		for _ in 0..5 {
 			hrmp_update = hrmp_update
 				.append(&HrmpChannelUpdate { msg_count: 1, total_bytes: 4 }, para_0, &limits)
-				.expect("update is withing the limits");
+				.expect("update is within the limits");
 		}
 		assert_matches!(
 			hrmp_update.append(
@@ -410,7 +537,7 @@ mod tests {
 		let para_1 = ParaId::from(1);
 		let para_1_limits = HrmpOutboundLimits { bytes_remaining: 20, messages_remaining: 3 };
 		let hrmp_outgoing = [(para_0, para_0_limits), (para_1, para_1_limits)].into();
-		let limits = TotalBandwidthLimits {
+		let limits = OutboundBandwidthLimits {
 			ump_messages_remaining: 0,
 			ump_bytes_remaining: 0,
 			hrmp_outgoing,
@@ -423,7 +550,9 @@ mod tests {
 			used_bandwidth: create_used_hrmp([(para_0, para_0_update)].into()),
 			para_head_hash: None::<relay_chain::Hash>,
 		};
-		segment.append(&ancestor_0, 0, &limits).expect("update is withing the limits");
+		segment
+			.append(&ancestor_0, HrmpWatermarkUpdate::Trunk(0), &limits)
+			.expect("update is within the limits");
 
 		for watermark in 1..5 {
 			let ancestor = Ancestor {
@@ -431,8 +560,8 @@ mod tests {
 				para_head_hash: None::<relay_chain::Hash>,
 			};
 			segment
-				.append(&ancestor, watermark, &limits)
-				.expect("update is withing the limits");
+				.append(&ancestor, HrmpWatermarkUpdate::Trunk(watermark), &limits)
+				.expect("update is within the limits");
 		}
 
 		let para_0_update = HrmpChannelUpdate { msg_count: 1, total_bytes: 1 };
@@ -441,7 +570,7 @@ mod tests {
 			para_head_hash: None::<relay_chain::Hash>,
 		};
 		assert_matches!(
-			segment.append(&ancestor_5, 5, &limits),
+			segment.append(&ancestor_5, HrmpWatermarkUpdate::Trunk(5), &limits),
 			Err(BandwidthUpdateError::HrmpBytesOverflow {
 				recipient,
 				bytes_remaining,
@@ -450,17 +579,21 @@ mod tests {
 		);
 		// Remove the first ancestor from the segment to make space.
 		segment.subtract(&ancestor_0);
-		segment.append(&ancestor_5, 5, &limits).expect("update is withing the limits");
+		segment
+			.append(&ancestor_5, HrmpWatermarkUpdate::Trunk(5), &limits)
+			.expect("update is within the limits");
 
 		let para_1_update = HrmpChannelUpdate { msg_count: 3, total_bytes: 10 };
 		let ancestor = Ancestor {
 			used_bandwidth: create_used_hrmp([(para_1, para_1_update)].into()),
 			para_head_hash: None::<relay_chain::Hash>,
 		};
-		segment.append(&ancestor, 6, &limits).expect("update is withing the limits");
+		segment
+			.append(&ancestor, HrmpWatermarkUpdate::Trunk(6), &limits)
+			.expect("update is within the limits");
 
 		assert_matches!(
-			segment.append(&ancestor, 7, &limits),
+			segment.append(&ancestor, HrmpWatermarkUpdate::Trunk(7), &limits),
 			Err(BandwidthUpdateError::HrmpMessagesOverflow {
 				recipient,
 				messages_remaining,
@@ -477,7 +610,7 @@ mod tests {
 			hrmp_outgoing: BTreeMap::default(),
 		};
 
-		let limits = TotalBandwidthLimits {
+		let limits = OutboundBandwidthLimits {
 			ump_messages_remaining: 5,
 			ump_bytes_remaining: 50,
 			hrmp_outgoing: BTreeMap::default(),
@@ -489,7 +622,9 @@ mod tests {
 			used_bandwidth: create_used_ump((1, 10)),
 			para_head_hash: None::<relay_chain::Hash>,
 		};
-		segment.append(&ancestor_0, 0, &limits).expect("update is withing the limits");
+		segment
+			.append(&ancestor_0, HrmpWatermarkUpdate::Trunk(0), &limits)
+			.expect("update is within the limits");
 
 		for watermark in 1..4 {
 			let ancestor = Ancestor {
@@ -497,8 +632,8 @@ mod tests {
 				para_head_hash: None::<relay_chain::Hash>,
 			};
 			segment
-				.append(&ancestor, watermark, &limits)
-				.expect("update is withing the limits");
+				.append(&ancestor, HrmpWatermarkUpdate::Trunk(watermark), &limits)
+				.expect("update is within the limits");
 		}
 
 		let ancestor_4 = Ancestor {
@@ -506,7 +641,7 @@ mod tests {
 			para_head_hash: None::<relay_chain::Hash>,
 		};
 		assert_matches!(
-			segment.append(&ancestor_4, 4, &limits),
+			segment.append(&ancestor_4, HrmpWatermarkUpdate::Trunk(4), &limits),
 			Err(BandwidthUpdateError::UmpBytesOverflow {
 				bytes_remaining,
 				bytes_submitted,
@@ -517,9 +652,11 @@ mod tests {
 			used_bandwidth: create_used_ump((1, 5)),
 			para_head_hash: None::<relay_chain::Hash>,
 		};
-		segment.append(&ancestor, 4, &limits).expect("update is withing the limits");
+		segment
+			.append(&ancestor, HrmpWatermarkUpdate::Trunk(4), &limits)
+			.expect("update is within the limits");
 		assert_matches!(
-			segment.append(&ancestor, 5, &limits),
+			segment.append(&ancestor, HrmpWatermarkUpdate::Trunk(5), &limits),
 			Err(BandwidthUpdateError::UmpMessagesOverflow {
 				messages_remaining,
 				messages_submitted,
@@ -535,17 +672,17 @@ mod tests {
 			used_bandwidth: UsedBandwidth::default(),
 			para_head_hash: None::<relay_chain::Hash>,
 		};
-		let limits = TotalBandwidthLimits {
+		let limits = OutboundBandwidthLimits {
 			ump_messages_remaining: 0,
 			ump_bytes_remaining: 0,
 			hrmp_outgoing: BTreeMap::default(),
 		};
 
 		segment
-			.append(&ancestor, 0, &limits)
+			.append(&ancestor, HrmpWatermarkUpdate::Head(0), &limits)
 			.expect("nothing to compare the watermark with in default segment");
 		assert_matches!(
-			segment.append(&ancestor, 0, &limits),
+			segment.append(&ancestor, HrmpWatermarkUpdate::Trunk(0), &limits),
 			Err(BandwidthUpdateError::InvalidHrmpWatermark {
 				submitted,
 				latest,
@@ -553,17 +690,23 @@ mod tests {
 		);
 
 		for watermark in 1..5 {
-			segment.append(&ancestor, watermark, &limits).expect("hrmp watermark is valid");
+			segment
+				.append(&ancestor, HrmpWatermarkUpdate::Trunk(watermark), &limits)
+				.expect("hrmp watermark is valid");
 		}
 		for watermark in 0..5 {
 			assert_matches!(
-				segment.append(&ancestor, watermark, &limits),
+				segment.append(&ancestor, HrmpWatermarkUpdate::Trunk(watermark), &limits),
 				Err(BandwidthUpdateError::InvalidHrmpWatermark {
 					submitted,
 					latest,
 				}) if submitted == watermark && latest == 4
 			);
 		}
+
+		segment
+			.append(&ancestor, HrmpWatermarkUpdate::Head(4), &limits)
+			.expect("head updates always valid");
 	}
 
 	#[test]
@@ -579,7 +722,7 @@ mod tests {
 		let para_1_limits =
 			HrmpOutboundLimits { bytes_remaining: u32::MAX, messages_remaining: u32::MAX };
 		let hrmp_outgoing = [(para_0, para_0_limits), (para_1, para_1_limits)].into();
-		let limits = TotalBandwidthLimits {
+		let limits = OutboundBandwidthLimits {
 			ump_messages_remaining: 0,
 			ump_bytes_remaining: 0,
 			hrmp_outgoing,
@@ -592,13 +735,17 @@ mod tests {
 			used_bandwidth: create_used_hrmp([(para_0, para_0_update)].into()),
 			para_head_hash: None::<relay_chain::Hash>,
 		};
-		segment.append(&ancestor_0, 0, &limits).expect("update is withing the limits");
+		segment
+			.append(&ancestor_0, HrmpWatermarkUpdate::Head(0), &limits)
+			.expect("update is within the limits");
 		let para_1_update = HrmpChannelUpdate { msg_count: 3, total_bytes: 10 };
 		let ancestor_1 = Ancestor {
 			used_bandwidth: create_used_hrmp([(para_1, para_1_update)].into()),
 			para_head_hash: None::<relay_chain::Hash>,
 		};
-		segment.append(&ancestor_1, 1, &limits).expect("update is withing the limits");
+		segment
+			.append(&ancestor_1, HrmpWatermarkUpdate::Head(1), &limits)
+			.expect("update is within the limits");
 
 		assert_eq!(segment.used_bandwidth.hrmp_outgoing.len(), 2);
 
diff --git a/pallets/parachain-system/src/validate_block/tests.rs b/pallets/parachain-system/src/validate_block/tests.rs
index 2c39188a085..8e61481193a 100644
--- a/pallets/parachain-system/src/validate_block/tests.rs
+++ b/pallets/parachain-system/src/validate_block/tests.rs
@@ -18,7 +18,9 @@ use codec::{Decode, DecodeAll, Encode};
 use cumulus_primitives_core::{ParachainBlockData, PersistedValidationData};
 use cumulus_test_client::{
 	generate_extrinsic,
-	runtime::{Block, Hash, Header, TestPalletCall, UncheckedExtrinsic, WASM_BINARY},
+	runtime::{
+		self as test_runtime, Block, Hash, Header, TestPalletCall, UncheckedExtrinsic, WASM_BINARY,
+	},
 	transfer, BlockData, BuildParachainBlockData, Client, DefaultTestClientBuilderExt, HeadData,
 	InitBlockBuilder, TestClientBuilder, TestClientBuilderExt, ValidationParams,
 };
@@ -79,8 +81,10 @@ fn build_block_with_witness(
 	client: &Client,
 	extra_extrinsics: Vec<UncheckedExtrinsic>,
 	parent_head: Header,
-	sproof_builder: RelayStateSproofBuilder,
+	mut sproof_builder: RelayStateSproofBuilder,
 ) -> TestBlockData {
+	sproof_builder.para_id = test_runtime::PARACHAIN_ID.into();
+	sproof_builder.included_para_head = Some(HeadData(parent_head.encode()));
 	let (relay_parent_storage_root, _) = sproof_builder.clone().into_state_root_and_proof();
 	let mut validation_data = PersistedValidationData {
 		relay_parent_number: 1,
diff --git a/pallets/xcmp-queue/src/mock.rs b/pallets/xcmp-queue/src/mock.rs
index 0d7d6eda00b..084ed98f372 100644
--- a/pallets/xcmp-queue/src/mock.rs
+++ b/pallets/xcmp-queue/src/mock.rs
@@ -116,6 +116,7 @@ impl cumulus_pallet_parachain_system::Config for Test {
 	type XcmpMessageHandler = XcmpQueue;
 	type ReservedXcmpWeight = ();
 	type CheckAssociatedRelayNumber = AnyRelayNumber;
+	type ConsensusHook = cumulus_pallet_parachain_system::consensus_hook::ExpectParentIncluded;
 }
 
 parameter_types! {
diff --git a/parachain-template/runtime/src/lib.rs b/parachain-template/runtime/src/lib.rs
index 61839e0a621..c45401f2a05 100644
--- a/parachain-template/runtime/src/lib.rs
+++ b/parachain-template/runtime/src/lib.rs
@@ -375,6 +375,7 @@ impl cumulus_pallet_parachain_system::Config for Runtime {
 	type XcmpMessageHandler = XcmpQueue;
 	type ReservedXcmpWeight = ReservedXcmpWeight;
 	type CheckAssociatedRelayNumber = RelayNumberStrictlyIncreases;
+	type ConsensusHook = cumulus_pallet_parachain_system::consensus_hook::ExpectParentIncluded;
 }
 
 impl parachain_info::Config for Runtime {}
diff --git a/parachains/runtimes/assets/statemine/src/lib.rs b/parachains/runtimes/assets/statemine/src/lib.rs
index 5f987851a18..06aed9d59b1 100644
--- a/parachains/runtimes/assets/statemine/src/lib.rs
+++ b/parachains/runtimes/assets/statemine/src/lib.rs
@@ -521,6 +521,7 @@ impl cumulus_pallet_parachain_system::Config for Runtime {
 	type XcmpMessageHandler = XcmpQueue;
 	type ReservedXcmpWeight = ReservedXcmpWeight;
 	type CheckAssociatedRelayNumber = RelayNumberStrictlyIncreases;
+	type ConsensusHook = cumulus_pallet_parachain_system::consensus_hook::ExpectParentIncluded;
 }
 
 impl parachain_info::Config for Runtime {}
diff --git a/parachains/runtimes/assets/statemint/src/lib.rs b/parachains/runtimes/assets/statemint/src/lib.rs
index aa90ca7a157..ac4a369b3ca 100644
--- a/parachains/runtimes/assets/statemint/src/lib.rs
+++ b/parachains/runtimes/assets/statemint/src/lib.rs
@@ -469,6 +469,7 @@ impl cumulus_pallet_parachain_system::Config for Runtime {
 	type XcmpMessageHandler = XcmpQueue;
 	type ReservedXcmpWeight = ReservedXcmpWeight;
 	type CheckAssociatedRelayNumber = RelayNumberStrictlyIncreases;
+	type ConsensusHook = cumulus_pallet_parachain_system::consensus_hook::ExpectParentIncluded;
 }
 
 impl parachain_info::Config for Runtime {}
diff --git a/parachains/runtimes/assets/westmint/src/lib.rs b/parachains/runtimes/assets/westmint/src/lib.rs
index c237c8dc5cf..6e9c0e83f29 100644
--- a/parachains/runtimes/assets/westmint/src/lib.rs
+++ b/parachains/runtimes/assets/westmint/src/lib.rs
@@ -506,6 +506,7 @@ impl cumulus_pallet_parachain_system::Config for Runtime {
 	type XcmpMessageHandler = XcmpQueue;
 	type ReservedXcmpWeight = ReservedXcmpWeight;
 	type CheckAssociatedRelayNumber = RelayNumberStrictlyIncreases;
+	type ConsensusHook = cumulus_pallet_parachain_system::consensus_hook::ExpectParentIncluded;
 }
 
 impl parachain_info::Config for Runtime {}
diff --git a/parachains/runtimes/bridge-hubs/bridge-hub-kusama/src/lib.rs b/parachains/runtimes/bridge-hubs/bridge-hub-kusama/src/lib.rs
index 4ebf760e849..a7ece2d4045 100644
--- a/parachains/runtimes/bridge-hubs/bridge-hub-kusama/src/lib.rs
+++ b/parachains/runtimes/bridge-hubs/bridge-hub-kusama/src/lib.rs
@@ -286,6 +286,7 @@ impl cumulus_pallet_parachain_system::Config for Runtime {
 	type XcmpMessageHandler = XcmpQueue;
 	type ReservedXcmpWeight = ReservedXcmpWeight;
 	type CheckAssociatedRelayNumber = RelayNumberStrictlyIncreases;
+	type ConsensusHook = cumulus_pallet_parachain_system::consensus_hook::ExpectParentIncluded;
 }
 
 impl parachain_info::Config for Runtime {}
diff --git a/parachains/runtimes/bridge-hubs/bridge-hub-polkadot/src/lib.rs b/parachains/runtimes/bridge-hubs/bridge-hub-polkadot/src/lib.rs
index 109cd2434b3..50d987ae1ec 100644
--- a/parachains/runtimes/bridge-hubs/bridge-hub-polkadot/src/lib.rs
+++ b/parachains/runtimes/bridge-hubs/bridge-hub-polkadot/src/lib.rs
@@ -286,6 +286,7 @@ impl cumulus_pallet_parachain_system::Config for Runtime {
 	type XcmpMessageHandler = XcmpQueue;
 	type ReservedXcmpWeight = ReservedXcmpWeight;
 	type CheckAssociatedRelayNumber = RelayNumberStrictlyIncreases;
+	type ConsensusHook = cumulus_pallet_parachain_system::consensus_hook::ExpectParentIncluded;
 }
 
 impl parachain_info::Config for Runtime {}
diff --git a/parachains/runtimes/bridge-hubs/bridge-hub-rococo/src/lib.rs b/parachains/runtimes/bridge-hubs/bridge-hub-rococo/src/lib.rs
index 04eddf59e3a..ac6b0117334 100644
--- a/parachains/runtimes/bridge-hubs/bridge-hub-rococo/src/lib.rs
+++ b/parachains/runtimes/bridge-hubs/bridge-hub-rococo/src/lib.rs
@@ -353,6 +353,7 @@ impl cumulus_pallet_parachain_system::Config for Runtime {
 	type XcmpMessageHandler = XcmpQueue;
 	type ReservedXcmpWeight = ReservedXcmpWeight;
 	type CheckAssociatedRelayNumber = RelayNumberStrictlyIncreases;
+	type ConsensusHook = cumulus_pallet_parachain_system::consensus_hook::ExpectParentIncluded;
 }
 
 impl parachain_info::Config for Runtime {}
diff --git a/parachains/runtimes/collectives/collectives-polkadot/src/lib.rs b/parachains/runtimes/collectives/collectives-polkadot/src/lib.rs
index e0f95cb052b..6ab122c2bc3 100644
--- a/parachains/runtimes/collectives/collectives-polkadot/src/lib.rs
+++ b/parachains/runtimes/collectives/collectives-polkadot/src/lib.rs
@@ -370,6 +370,7 @@ impl cumulus_pallet_parachain_system::Config for Runtime {
 	type XcmpMessageHandler = XcmpQueue;
 	type ReservedXcmpWeight = ReservedXcmpWeight;
 	type CheckAssociatedRelayNumber = RelayNumberStrictlyIncreases;
+	type ConsensusHook = cumulus_pallet_parachain_system::consensus_hook::ExpectParentIncluded;
 }
 
 impl parachain_info::Config for Runtime {}
diff --git a/parachains/runtimes/contracts/contracts-rococo/src/lib.rs b/parachains/runtimes/contracts/contracts-rococo/src/lib.rs
index 819bb7ba537..57a85f20b64 100644
--- a/parachains/runtimes/contracts/contracts-rococo/src/lib.rs
+++ b/parachains/runtimes/contracts/contracts-rococo/src/lib.rs
@@ -266,6 +266,7 @@ impl cumulus_pallet_parachain_system::Config for Runtime {
 	type XcmpMessageHandler = XcmpQueue;
 	type ReservedXcmpWeight = ReservedXcmpWeight;
 	type CheckAssociatedRelayNumber = RelayNumberStrictlyIncreases;
+	type ConsensusHook = cumulus_pallet_parachain_system::consensus_hook::ExpectParentIncluded;
 }
 
 impl pallet_insecure_randomness_collective_flip::Config for Runtime {}
diff --git a/parachains/runtimes/starters/seedling/src/lib.rs b/parachains/runtimes/starters/seedling/src/lib.rs
index 2861cac1fe4..97b18cd6f3f 100644
--- a/parachains/runtimes/starters/seedling/src/lib.rs
+++ b/parachains/runtimes/starters/seedling/src/lib.rs
@@ -176,6 +176,7 @@ impl cumulus_pallet_parachain_system::Config for Runtime {
 	type XcmpMessageHandler = ();
 	type ReservedXcmpWeight = ();
 	type CheckAssociatedRelayNumber = RelayNumberStrictlyIncreases;
+	type ConsensusHook = cumulus_pallet_parachain_system::consensus_hook::ExpectParentIncluded;
 }
 
 impl parachain_info::Config for Runtime {}
diff --git a/parachains/runtimes/starters/shell/src/lib.rs b/parachains/runtimes/starters/shell/src/lib.rs
index a05a78863c4..a896611bf46 100644
--- a/parachains/runtimes/starters/shell/src/lib.rs
+++ b/parachains/runtimes/starters/shell/src/lib.rs
@@ -180,6 +180,7 @@ impl cumulus_pallet_parachain_system::Config for Runtime {
 	type XcmpMessageHandler = ();
 	type ReservedXcmpWeight = ();
 	type CheckAssociatedRelayNumber = RelayNumberStrictlyIncreases;
+	type ConsensusHook = cumulus_pallet_parachain_system::consensus_hook::ExpectParentIncluded;
 }
 
 impl parachain_info::Config for Runtime {}
diff --git a/parachains/runtimes/testing/penpal/src/lib.rs b/parachains/runtimes/testing/penpal/src/lib.rs
index 7ade3bd2f63..0ecb0c6794d 100644
--- a/parachains/runtimes/testing/penpal/src/lib.rs
+++ b/parachains/runtimes/testing/penpal/src/lib.rs
@@ -458,6 +458,7 @@ impl cumulus_pallet_parachain_system::Config for Runtime {
 	type XcmpMessageHandler = XcmpQueue;
 	type ReservedXcmpWeight = ReservedXcmpWeight;
 	type CheckAssociatedRelayNumber = RelayNumberStrictlyIncreases;
+	type ConsensusHook = cumulus_pallet_parachain_system::consensus_hook::ExpectParentIncluded;
 }
 
 impl parachain_info::Config for Runtime {}
diff --git a/parachains/runtimes/testing/rococo-parachain/src/lib.rs b/parachains/runtimes/testing/rococo-parachain/src/lib.rs
index 1a26290f2a3..4feb14bb4fd 100644
--- a/parachains/runtimes/testing/rococo-parachain/src/lib.rs
+++ b/parachains/runtimes/testing/rococo-parachain/src/lib.rs
@@ -272,6 +272,7 @@ impl cumulus_pallet_parachain_system::Config for Runtime {
 	type XcmpMessageHandler = XcmpQueue;
 	type ReservedXcmpWeight = ReservedXcmpWeight;
 	type CheckAssociatedRelayNumber = RelayNumberStrictlyIncreases;
+	type ConsensusHook = cumulus_pallet_parachain_system::consensus_hook::ExpectParentIncluded;
 }
 
 impl parachain_info::Config for Runtime {}
diff --git a/primitives/timestamp/src/lib.rs b/primitives/timestamp/src/lib.rs
index ddc2fe340dd..97fd3128ca2 100644
--- a/primitives/timestamp/src/lib.rs
+++ b/primitives/timestamp/src/lib.rs
@@ -110,11 +110,15 @@ mod tests {
 		timestamp: u64,
 		relay_chain_slot: Slot,
 	) -> (ParachainBlockData, PHash) {
-		let sproof_builder =
-			RelayStateSproofBuilder { current_slot: relay_chain_slot, ..Default::default() };
-
 		let parent_header = client.header(hash).ok().flatten().expect("Genesis header exists");
 
+		let sproof_builder = RelayStateSproofBuilder {
+			para_id: cumulus_test_client::runtime::PARACHAIN_ID.into(),
+			included_para_head: Some(HeadData(parent_header.encode())),
+			current_slot: relay_chain_slot,
+			..Default::default()
+		};
+
 		let relay_parent_storage_root = sproof_builder.clone().into_state_root_and_proof().0;
 
 		let validation_data = PersistedValidationData {
diff --git a/test/runtime/src/lib.rs b/test/runtime/src/lib.rs
index 9d7e9f078da..0886f528905 100644
--- a/test/runtime/src/lib.rs
+++ b/test/runtime/src/lib.rs
@@ -76,6 +76,9 @@ impl_opaque_keys! {
 /// [`OnRuntimeUpgrade`] works as expected.
 pub const TEST_RUNTIME_UPGRADE_KEY: &[u8] = b"+test_runtime_upgrade_key+";
 
+/// The para-id used in this runtime.
+pub const PARACHAIN_ID: u32 = 100;
+
 // The only difference between the two declarations below is the `spec_version`. With the
 // `increment-spec-version` feature enabled `spec_version` should be greater than the one of without the
 // `increment-spec-version` feature.
@@ -274,10 +277,11 @@ impl cumulus_pallet_parachain_system::Config for Runtime {
 	type XcmpMessageHandler = ();
 	type ReservedXcmpWeight = ();
 	type CheckAssociatedRelayNumber = cumulus_pallet_parachain_system::AnyRelayNumber;
+	type ConsensusHook = cumulus_pallet_parachain_system::consensus_hook::RequireParentIncluded;
 }
 
 parameter_types! {
-	pub storage ParachainId: cumulus_primitives_core::ParaId = 100.into();
+	pub storage ParachainId: cumulus_primitives_core::ParaId = PARACHAIN_ID.into();
 }
 
 impl test_pallet::Config for Runtime {}