diff --git a/Cargo.lock b/Cargo.lock
index 14ce58be7faa..59b6d92bde5d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -16046,6 +16046,7 @@ dependencies = [
  "polkadot-primitives-test-helpers",
  "polkadot-subsystem-bench",
  "rand_chacha",
+ "rstest",
  "sc-keystore",
  "sc-network",
  "sp-application-crypto 30.0.0",
diff --git a/polkadot/node/core/prospective-parachains/src/fragment_chain/tests.rs b/polkadot/node/core/prospective-parachains/src/fragment_chain/tests.rs
index 9708b0871c2a..2f8a5525570c 100644
--- a/polkadot/node/core/prospective-parachains/src/fragment_chain/tests.rs
+++ b/polkadot/node/core/prospective-parachains/src/fragment_chain/tests.rs
@@ -71,7 +71,7 @@ fn make_committed_candidate(
 			persisted_validation_data_hash: persisted_validation_data.hash(),
 			pov_hash: Hash::repeat_byte(1),
 			erasure_root: Hash::repeat_byte(1),
-			signature: test_helpers::dummy_collator_signature(),
+			signature: test_helpers::zero_collator_signature(),
 			para_head: para_head.hash(),
 			validation_code_hash: Hash::repeat_byte(42).into(),
 		}
diff --git a/polkadot/node/network/statement-distribution/Cargo.toml b/polkadot/node/network/statement-distribution/Cargo.toml
index 08059353033d..de07937ffb0a 100644
--- a/polkadot/node/network/statement-distribution/Cargo.toml
+++ b/polkadot/node/network/statement-distribution/Cargo.toml
@@ -44,6 +44,7 @@ polkadot-primitives = { workspace = true, features = ["test"] }
 polkadot-primitives-test-helpers = { workspace = true }
 rand_chacha = { workspace = true, default-features = true }
 polkadot-subsystem-bench = { workspace = true }
+rstest = { workspace = true }
 
 [[bench]]
 name = "statement-distribution-regression-bench"
diff --git a/polkadot/node/network/statement-distribution/src/error.rs b/polkadot/node/network/statement-distribution/src/error.rs
index d7f52162fe23..cff9afbf8667 100644
--- a/polkadot/node/network/statement-distribution/src/error.rs
+++ b/polkadot/node/network/statement-distribution/src/error.rs
@@ -72,9 +72,6 @@ pub enum Error {
 	#[error("Fetching session info failed {0:?}")]
 	FetchSessionInfo(RuntimeApiError),
 
-	#[error("Fetching availability cores failed {0:?}")]
-	FetchAvailabilityCores(RuntimeApiError),
-
 	#[error("Fetching disabled validators failed {0:?}")]
 	FetchDisabledValidators(runtime::Error),
 
@@ -82,7 +79,7 @@ pub enum Error {
 	FetchValidatorGroups(RuntimeApiError),
 
 	#[error("Fetching claim queue failed {0:?}")]
-	FetchClaimQueue(runtime::Error),
+	FetchClaimQueue(RuntimeApiError),
 
 	#[error("Attempted to share statement when not a validator or not assigned")]
 	InvalidShare,
diff --git a/polkadot/node/network/statement-distribution/src/v2/mod.rs b/polkadot/node/network/statement-distribution/src/v2/mod.rs
index c79ae3953ad9..6bb49e5de13d 100644
--- a/polkadot/node/network/statement-distribution/src/v2/mod.rs
+++ b/polkadot/node/network/statement-distribution/src/v2/mod.rs
@@ -46,12 +46,15 @@ use polkadot_node_subsystem_util::{
 	backing_implicit_view::View as ImplicitView,
 	reputation::ReputationAggregator,
 	runtime::{
-		fetch_claim_queue, request_min_backing_votes, ClaimQueueSnapshot, ProspectiveParachainsMode,
+		request_min_backing_votes, request_node_features, ClaimQueueSnapshot,
+		ProspectiveParachainsMode,
 	},
 };
 use polkadot_primitives::{
-	vstaging::CoreState, AuthorityDiscoveryId, CandidateHash, CompactStatement, CoreIndex,
-	GroupIndex, GroupRotationInfo, Hash, Id as ParaId, IndexedVec, SessionIndex, SessionInfo,
+	node_features::FeatureIndex,
+	vstaging::{transpose_claim_queue, CandidateDescriptorVersion, TransposedClaimQueue},
+	AuthorityDiscoveryId, CandidateHash, CompactStatement, CoreIndex, GroupIndex,
+	GroupRotationInfo, Hash, Id as ParaId, IndexedVec, NodeFeatures, SessionIndex, SessionInfo,
 	SignedStatement, SigningContext, UncheckedSignedStatement, ValidatorId, ValidatorIndex,
 };
 
@@ -137,6 +140,12 @@ const COST_UNREQUESTED_RESPONSE_STATEMENT: Rep =
 	Rep::CostMajor("Un-requested Statement In Response");
 const COST_INACCURATE_ADVERTISEMENT: Rep =
 	Rep::CostMajor("Peer advertised a candidate inaccurately");
+const COST_UNSUPPORTED_DESCRIPTOR_VERSION: Rep =
+	Rep::CostMajor("Candidate Descriptor version is not supported");
+const COST_INVALID_CORE_INDEX: Rep =
+	Rep::CostMajor("Candidate Descriptor contains an invalid core index");
+const COST_INVALID_SESSION_INDEX: Rep =
+	Rep::CostMajor("Candidate Descriptor contains an invalid session index");
 
 const COST_INVALID_REQUEST: Rep = Rep::CostMajor("Peer sent unparsable request");
 const COST_INVALID_REQUEST_BITFIELD_SIZE: Rep =
@@ -156,6 +165,7 @@ struct PerRelayParentState {
 	statement_store: StatementStore,
 	seconding_limit: usize,
 	session: SessionIndex,
+	transposed_cq: TransposedClaimQueue,
 	groups_per_para: HashMap<ParaId, Vec<GroupIndex>>,
 	disabled_validators: HashSet<ValidatorIndex>,
 }
@@ -219,10 +229,17 @@ struct PerSessionState {
 	// getting the topology from the gossip-support subsystem
 	grid_view: Option<grid::SessionTopologyView>,
 	local_validator: Option<LocalValidatorIndex>,
+	// `true` if v2 candidate receipts are allowed by the runtime
+	allow_v2_descriptors: bool,
 }
 
 impl PerSessionState {
-	fn new(session_info: SessionInfo, keystore: &KeystorePtr, backing_threshold: u32) -> Self {
+	fn new(
+		session_info: SessionInfo,
+		keystore: &KeystorePtr,
+		backing_threshold: u32,
+		allow_v2_descriptors: bool,
+	) -> Self {
 		let groups = Groups::new(session_info.validator_groups.clone(), backing_threshold);
 		let mut authority_lookup = HashMap::new();
 		for (i, ad) in session_info.discovery_keys.iter().cloned().enumerate() {
@@ -235,7 +252,14 @@ impl PerSessionState {
 		)
 		.map(|(_, index)| LocalValidatorIndex::Active(index));
 
-		PerSessionState { session_info, groups, authority_lookup, grid_view: None, local_validator }
+		PerSessionState {
+			session_info,
+			groups,
+			authority_lookup,
+			grid_view: None,
+			local_validator,
+			allow_v2_descriptors,
+		}
 	}
 
 	fn supply_topology(
@@ -271,6 +295,11 @@ impl PerSessionState {
 	fn is_not_validator(&self) -> bool {
 		self.grid_view.is_some() && self.local_validator.is_none()
 	}
+
+	/// Returns `true` if v2 candidate receipts are enabled
+	fn candidate_receipt_v2_enabled(&self) -> bool {
+		self.allow_v2_descriptors
+	}
 }
 
 pub(crate) struct State {
@@ -615,8 +644,18 @@ pub(crate) async fn handle_active_leaves_update<Context>(
 
 			let minimum_backing_votes =
 				request_min_backing_votes(new_relay_parent, session_index, ctx.sender()).await?;
-			let mut per_session_state =
-				PerSessionState::new(session_info, &state.keystore, minimum_backing_votes);
+			let node_features =
+				request_node_features(new_relay_parent, session_index, ctx.sender()).await?;
+			let mut per_session_state = PerSessionState::new(
+				session_info,
+				&state.keystore,
+				minimum_backing_votes,
+				node_features
+					.unwrap_or(NodeFeatures::EMPTY)
+					.get(FeatureIndex::CandidateReceiptV2 as usize)
+					.map(|b| *b)
+					.unwrap_or(false),
+			);
 			if let Some(topology) = state.unused_topologies.remove(&session_index) {
 				per_session_state.supply_topology(&topology.topology, topology.local_index);
 			}
@@ -642,18 +681,6 @@ pub(crate) async fn handle_active_leaves_update<Context>(
 			continue
 		}
 
-		// New leaf: fetch info from runtime API and initialize
-		// `per_relay_parent`.
-
-		let availability_cores = polkadot_node_subsystem_util::request_availability_cores(
-			new_relay_parent,
-			ctx.sender(),
-		)
-		.await
-		.await
-		.map_err(JfyiError::RuntimeApiUnavailable)?
-		.map_err(JfyiError::FetchAvailabilityCores)?;
-
 		let group_rotation_info =
 			polkadot_node_subsystem_util::request_validator_groups(new_relay_parent, ctx.sender())
 				.await
@@ -662,23 +689,22 @@ pub(crate) async fn handle_active_leaves_update<Context>(
 				.map_err(JfyiError::FetchValidatorGroups)?
 				.1;
 
-		let maybe_claim_queue = fetch_claim_queue(ctx.sender(), new_relay_parent)
-		.await
-		.unwrap_or_else(|err| {
-				gum::debug!(target: LOG_TARGET, ?new_relay_parent, ?err, "handle_active_leaves_update: `claim_queue` API not available");
-				None
-			});
+		let claim_queue = ClaimQueueSnapshot(
+			polkadot_node_subsystem_util::request_claim_queue(new_relay_parent, ctx.sender())
+				.await
+				.await
+				.map_err(JfyiError::RuntimeApiUnavailable)?
+				.map_err(JfyiError::FetchClaimQueue)?,
+		);
 
 		let local_validator = per_session.local_validator.and_then(|v| {
 			if let LocalValidatorIndex::Active(idx) = v {
 				find_active_validator_state(
 					idx,
 					&per_session.groups,
-					&availability_cores,
 					&group_rotation_info,
-					&maybe_claim_queue,
+					&claim_queue,
 					seconding_limit,
-					max_candidate_depth,
 				)
 			} else {
 				Some(LocalValidatorState { grid_tracker: GridTracker::default(), active: None })
@@ -686,13 +712,14 @@ pub(crate) async fn handle_active_leaves_update<Context>(
 		});
 
 		let groups_per_para = determine_groups_per_para(
-			availability_cores,
+			per_session.groups.all().len(),
 			group_rotation_info,
-			&maybe_claim_queue,
-			max_candidate_depth,
+			&claim_queue,
 		)
 		.await;
 
+		let transposed_cq = transpose_claim_queue(claim_queue.0);
+
 		state.per_relay_parent.insert(
 			new_relay_parent,
 			PerRelayParentState {
@@ -702,6 +729,7 @@ pub(crate) async fn handle_active_leaves_update<Context>(
 				session: session_index,
 				groups_per_para,
 				disabled_validators,
+				transposed_cq,
 			},
 		);
 	}
@@ -741,11 +769,9 @@ pub(crate) async fn handle_active_leaves_update<Context>(
 fn find_active_validator_state(
 	validator_index: ValidatorIndex,
 	groups: &Groups,
-	availability_cores: &[CoreState],
 	group_rotation_info: &GroupRotationInfo,
-	maybe_claim_queue: &Option<ClaimQueueSnapshot>,
+	claim_queue: &ClaimQueueSnapshot,
 	seconding_limit: usize,
-	max_candidate_depth: usize,
 ) -> Option<LocalValidatorState> {
 	if groups.all().is_empty() {
 		return None
@@ -753,23 +779,8 @@ fn find_active_validator_state(
 
 	let our_group = groups.by_validator_index(validator_index)?;
 
-	let core_index = group_rotation_info.core_for_group(our_group, availability_cores.len());
-	let paras_assigned_to_core = if let Some(claim_queue) = maybe_claim_queue {
-		claim_queue.iter_claims_for_core(&core_index).copied().collect()
-	} else {
-		availability_cores
-			.get(core_index.0 as usize)
-			.and_then(|core_state| match core_state {
-				CoreState::Scheduled(scheduled_core) => Some(scheduled_core.para_id),
-				CoreState::Occupied(occupied_core) if max_candidate_depth >= 1 => occupied_core
-					.next_up_on_available
-					.as_ref()
-					.map(|scheduled_core| scheduled_core.para_id),
-				CoreState::Free | CoreState::Occupied(_) => None,
-			})
-			.into_iter()
-			.collect()
-	};
+	let core_index = group_rotation_info.core_for_group(our_group, groups.all().len());
+	let paras_assigned_to_core = claim_queue.iter_claims_for_core(&core_index).copied().collect();
 	let group_validators = groups.get(our_group)?.to_owned();
 
 	Some(LocalValidatorState {
@@ -2174,39 +2185,16 @@ async fn provide_candidate_to_grid<Context>(
 
 // Utility function to populate per relay parent `ParaId` to `GroupIndex` mappings.
 async fn determine_groups_per_para(
-	availability_cores: Vec<CoreState>,
+	n_cores: usize,
 	group_rotation_info: GroupRotationInfo,
-	maybe_claim_queue: &Option<ClaimQueueSnapshot>,
-	max_candidate_depth: usize,
+	claim_queue: &ClaimQueueSnapshot,
 ) -> HashMap<ParaId, Vec<GroupIndex>> {
-	let n_cores = availability_cores.len();
-
 	// Determine the core indices occupied by each para at the current relay parent. To support
 	// on-demand parachains we also consider the core indices at next blocks.
-	let schedule: HashMap<CoreIndex, Vec<ParaId>> = if let Some(claim_queue) = maybe_claim_queue {
-		claim_queue
-			.iter_all_claims()
-			.map(|(core_index, paras)| (*core_index, paras.iter().copied().collect()))
-			.collect()
-	} else {
-		availability_cores
-			.into_iter()
-			.enumerate()
-			.filter_map(|(index, core)| match core {
-				CoreState::Scheduled(scheduled_core) =>
-					Some((CoreIndex(index as u32), vec![scheduled_core.para_id])),
-				CoreState::Occupied(occupied_core) =>
-					if max_candidate_depth >= 1 {
-						occupied_core.next_up_on_available.map(|scheduled_core| {
-							(CoreIndex(index as u32), vec![scheduled_core.para_id])
-						})
-					} else {
-						None
-					},
-				CoreState::Free => None,
-			})
-			.collect()
-	};
+	let schedule: HashMap<CoreIndex, Vec<ParaId>> = claim_queue
+		.iter_all_claims()
+		.map(|(core_index, paras)| (*core_index, paras.iter().copied().collect()))
+		.collect();
 
 	let mut groups_per_para = HashMap::new();
 	// Map from `CoreIndex` to `GroupIndex` and collect as `HashMap`.
@@ -3106,11 +3094,12 @@ pub(crate) async fn handle_response<Context>(
 ) {
 	let &requests::CandidateIdentifier { relay_parent, candidate_hash, group_index } =
 		response.candidate_identifier();
+	let peer = *response.requested_peer();
 
 	gum::trace!(
 		target: LOG_TARGET,
 		?candidate_hash,
-		peer = ?response.requested_peer(),
+		?peer,
 		"Received response",
 	);
 
@@ -3145,6 +3134,8 @@ pub(crate) async fn handle_response<Context>(
 				expected_groups.iter().any(|g| g == &g_index)
 			},
 			disabled_mask,
+			&relay_parent_state.transposed_cq,
+			per_session.candidate_receipt_v2_enabled(),
 		);
 
 		for (peer, rep) in res.reputation_changes {
diff --git a/polkadot/node/network/statement-distribution/src/v2/requests.rs b/polkadot/node/network/statement-distribution/src/v2/requests.rs
index 74f29710956f..3b46922c2297 100644
--- a/polkadot/node/network/statement-distribution/src/v2/requests.rs
+++ b/polkadot/node/network/statement-distribution/src/v2/requests.rs
@@ -30,9 +30,11 @@
 //! (which requires state not owned by the request manager).
 
 use super::{
-	seconded_and_sufficient, BENEFIT_VALID_RESPONSE, BENEFIT_VALID_STATEMENT,
-	COST_IMPROPERLY_DECODED_RESPONSE, COST_INVALID_RESPONSE, COST_INVALID_SIGNATURE,
-	COST_UNREQUESTED_RESPONSE_STATEMENT, REQUEST_RETRY_DELAY,
+	seconded_and_sufficient, CandidateDescriptorVersion, TransposedClaimQueue,
+	BENEFIT_VALID_RESPONSE, BENEFIT_VALID_STATEMENT, COST_IMPROPERLY_DECODED_RESPONSE,
+	COST_INVALID_CORE_INDEX, COST_INVALID_RESPONSE, COST_INVALID_SESSION_INDEX,
+	COST_INVALID_SIGNATURE, COST_UNREQUESTED_RESPONSE_STATEMENT,
+	COST_UNSUPPORTED_DESCRIPTOR_VERSION, REQUEST_RETRY_DELAY,
 };
 use crate::LOG_TARGET;
 
@@ -566,6 +568,8 @@ impl UnhandledResponse {
 		validator_key_lookup: impl Fn(ValidatorIndex) -> Option<ValidatorId>,
 		allowed_para_lookup: impl Fn(ParaId, GroupIndex) -> bool,
 		disabled_mask: BitVec<u8, Lsb0>,
+		transposed_cq: &TransposedClaimQueue,
+		allow_v2_descriptors: bool,
 	) -> ResponseValidationOutput {
 		let UnhandledResponse {
 			response: TaggedResponse { identifier, requested_peer, props, response },
@@ -650,6 +654,8 @@ impl UnhandledResponse {
 			validator_key_lookup,
 			allowed_para_lookup,
 			disabled_mask,
+			transposed_cq,
+			allow_v2_descriptors,
 		);
 
 		if let CandidateRequestStatus::Complete { .. } = output.request_status {
@@ -670,6 +676,8 @@ fn validate_complete_response(
 	validator_key_lookup: impl Fn(ValidatorIndex) -> Option<ValidatorId>,
 	allowed_para_lookup: impl Fn(ParaId, GroupIndex) -> bool,
 	disabled_mask: BitVec<u8, Lsb0>,
+	transposed_cq: &TransposedClaimQueue,
+	allow_v2_descriptors: bool,
 ) -> ResponseValidationOutput {
 	let RequestProperties { backing_threshold, mut unwanted_mask } = props;
 
@@ -687,39 +695,83 @@ fn validate_complete_response(
 		unwanted_mask.validated_in_group.resize(group.len(), true);
 	}
 
-	let invalid_candidate_output = || ResponseValidationOutput {
+	let invalid_candidate_output = |cost: Rep| ResponseValidationOutput {
 		request_status: CandidateRequestStatus::Incomplete,
-		reputation_changes: vec![(requested_peer, COST_INVALID_RESPONSE)],
+		reputation_changes: vec![(requested_peer, cost)],
 		requested_peer,
 	};
 
+	let mut rep_changes = Vec::new();
+
 	// sanity-check candidate response.
 	// note: roughly ascending cost of operations
 	{
 		if response.candidate_receipt.descriptor.relay_parent() != identifier.relay_parent {
-			return invalid_candidate_output()
+			return invalid_candidate_output(COST_INVALID_RESPONSE)
 		}
 
 		if response.candidate_receipt.descriptor.persisted_validation_data_hash() !=
 			response.persisted_validation_data.hash()
 		{
-			return invalid_candidate_output()
+			return invalid_candidate_output(COST_INVALID_RESPONSE)
 		}
 
 		if !allowed_para_lookup(
 			response.candidate_receipt.descriptor.para_id(),
 			identifier.group_index,
 		) {
-			return invalid_candidate_output()
+			return invalid_candidate_output(COST_INVALID_RESPONSE)
 		}
 
 		if response.candidate_receipt.hash() != identifier.candidate_hash {
-			return invalid_candidate_output()
+			return invalid_candidate_output(COST_INVALID_RESPONSE)
+		}
+
+		let candidate_hash = response.candidate_receipt.hash();
+
+		// V2 descriptors are invalid if not enabled by runtime.
+		if !allow_v2_descriptors &&
+			response.candidate_receipt.descriptor.version() == CandidateDescriptorVersion::V2
+		{
+			gum::debug!(
+				target: LOG_TARGET,
+				?candidate_hash,
+				peer = ?requested_peer,
+				"Version 2 candidate receipts are not enabled by the runtime"
+			);
+			return invalid_candidate_output(COST_UNSUPPORTED_DESCRIPTOR_VERSION)
+		}
+		// Validate the core index.
+		if let Err(err) = response.candidate_receipt.check_core_index(transposed_cq) {
+			gum::debug!(
+				target: LOG_TARGET,
+				?candidate_hash,
+				?err,
+				peer = ?requested_peer,
+				"Received candidate has invalid core index"
+			);
+			return invalid_candidate_output(COST_INVALID_CORE_INDEX)
+		}
+
+		// Check if `session_index` of relay parent matches candidate descriptor
+		// `session_index`.
+		if let Some(candidate_session_index) = response.candidate_receipt.descriptor.session_index()
+		{
+			if candidate_session_index != session {
+				gum::debug!(
+					target: LOG_TARGET,
+					?candidate_hash,
+					peer = ?requested_peer,
+					session_index = session,
+					candidate_session_index,
+					"Received candidate has invalid session index"
+				);
+				return invalid_candidate_output(COST_INVALID_SESSION_INDEX)
+			}
 		}
 	}
 
 	// statement checks.
-	let mut rep_changes = Vec::new();
 	let statements = {
 		let mut statements =
 			Vec::with_capacity(std::cmp::min(response.statements.len(), group.len() * 2));
@@ -815,7 +867,7 @@ fn validate_complete_response(
 		// Only accept responses which are sufficient, according to our
 		// required backing threshold.
 		if !seconded_and_sufficient(&received_filter, backing_threshold) {
-			return invalid_candidate_output()
+			return invalid_candidate_output(COST_INVALID_RESPONSE)
 		}
 
 		statements
@@ -1091,6 +1143,8 @@ mod tests {
 				validator_key_lookup,
 				allowed_para_lookup,
 				disabled_mask.clone(),
+				&Default::default(),
+				false,
 			);
 			assert_eq!(
 				output,
@@ -1130,6 +1184,8 @@ mod tests {
 				validator_key_lookup,
 				allowed_para_lookup,
 				disabled_mask,
+				&Default::default(),
+				false,
 			);
 			assert_eq!(
 				output,
@@ -1214,6 +1270,8 @@ mod tests {
 				validator_key_lookup,
 				allowed_para_lookup,
 				disabled_mask,
+				&Default::default(),
+				false,
 			);
 			assert_eq!(
 				output,
@@ -1296,6 +1354,8 @@ mod tests {
 				validator_key_lookup,
 				allowed_para_lookup,
 				disabled_mask,
+				&Default::default(),
+				false,
 			);
 			assert_eq!(
 				output,
@@ -1434,6 +1494,8 @@ mod tests {
 				validator_key_lookup,
 				allowed_para_lookup,
 				disabled_mask.clone(),
+				&Default::default(),
+				false,
 			);
 
 			// First request served successfully
diff --git a/polkadot/node/network/statement-distribution/src/v2/tests/cluster.rs b/polkadot/node/network/statement-distribution/src/v2/tests/cluster.rs
index fe51f953e244..040123f1774c 100644
--- a/polkadot/node/network/statement-distribution/src/v2/tests/cluster.rs
+++ b/polkadot/node/network/statement-distribution/src/v2/tests/cluster.rs
@@ -25,6 +25,7 @@ fn share_seconded_circulated_to_cluster() {
 		group_size: 3,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -125,6 +126,7 @@ fn cluster_valid_statement_before_seconded_ignored() {
 		group_size: 3,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -185,6 +187,7 @@ fn cluster_statement_bad_signature() {
 		group_size: 3,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -258,6 +261,7 @@ fn useful_cluster_statement_from_non_cluster_peer_rejected() {
 		group_size: 3,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -320,6 +324,7 @@ fn elastic_scaling_useful_cluster_statement_from_non_cluster_peer_rejected() {
 		group_size: 3,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -379,6 +384,7 @@ fn statement_from_non_cluster_originator_unexpected() {
 		group_size: 3,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -434,6 +440,7 @@ fn seconded_statement_leads_to_request() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -522,6 +529,7 @@ fn cluster_statements_shared_seconded_first() {
 		group_size: 3,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -636,6 +644,7 @@ fn cluster_accounts_for_implicit_view() {
 		group_size: 3,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -772,6 +781,7 @@ fn cluster_messages_imported_after_confirmed_candidate_importable_check() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -895,6 +905,7 @@ fn cluster_messages_imported_after_new_leaf_importable_check() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -1031,6 +1042,7 @@ fn ensure_seconding_limit_is_respected() {
 			max_candidate_depth: 1,
 			allowed_ancestry_len: 3,
 		}),
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
diff --git a/polkadot/node/network/statement-distribution/src/v2/tests/grid.rs b/polkadot/node/network/statement-distribution/src/v2/tests/grid.rs
index d2bf031368c1..0133d9e219f6 100644
--- a/polkadot/node/network/statement-distribution/src/v2/tests/grid.rs
+++ b/polkadot/node/network/statement-distribution/src/v2/tests/grid.rs
@@ -31,6 +31,7 @@ fn backed_candidate_leads_to_advertisement() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -240,6 +241,7 @@ fn received_advertisement_before_confirmation_leads_to_request() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -412,6 +414,7 @@ fn received_advertisement_after_backing_leads_to_acknowledgement() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	test_harness(config, |state, mut overseer| async move {
@@ -593,6 +596,7 @@ fn receive_ack_for_unconfirmed_candidate() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	test_harness(config, |state, mut overseer| async move {
@@ -654,6 +658,7 @@ fn received_acknowledgements_for_locally_confirmed() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	test_harness(config, |state, mut overseer| async move {
@@ -816,6 +821,7 @@ fn received_acknowledgements_for_externally_confirmed() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	test_harness(config, |state, mut overseer| async move {
@@ -951,6 +957,7 @@ fn received_advertisement_after_confirmation_before_backing() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -1129,6 +1136,7 @@ fn additional_statements_are_shared_after_manifest_exchange() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -1416,6 +1424,7 @@ fn advertisement_sent_when_peer_enters_relay_parent_view() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -1629,6 +1638,7 @@ fn advertisement_not_re_sent_when_peer_re_enters_view() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -1840,6 +1850,7 @@ fn inner_grid_statements_imported_to_backing(groups_for_first_para: usize) {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -2048,6 +2059,7 @@ fn advertisements_rejected_from_incorrect_peers() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -2184,6 +2196,7 @@ fn manifest_rejected_with_unknown_relay_parent() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -2281,6 +2294,7 @@ fn manifest_rejected_when_not_a_validator() {
 		group_size,
 		local_validator: LocalRole::None,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -2374,6 +2388,7 @@ fn manifest_rejected_when_group_does_not_match_para() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -2472,6 +2487,7 @@ fn peer_reported_for_advertisement_conflicting_with_confirmed_candidate() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -2662,6 +2678,7 @@ fn inactive_local_participates_in_grid() {
 		group_size,
 		local_validator: LocalRole::InactiveValidator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
diff --git a/polkadot/node/network/statement-distribution/src/v2/tests/mod.rs b/polkadot/node/network/statement-distribution/src/v2/tests/mod.rs
index 9f2c36ad1018..46b72f5adac9 100644
--- a/polkadot/node/network/statement-distribution/src/v2/tests/mod.rs
+++ b/polkadot/node/network/statement-distribution/src/v2/tests/mod.rs
@@ -33,9 +33,9 @@ use polkadot_node_subsystem::messages::{
 use polkadot_node_subsystem_test_helpers as test_helpers;
 use polkadot_node_subsystem_util::TimeoutExt;
 use polkadot_primitives::{
-	vstaging::{CommittedCandidateReceiptV2 as CommittedCandidateReceipt, CoreState},
-	AssignmentPair, AsyncBackingParams, Block, BlockNumber, GroupRotationInfo, HeadData, Header,
-	IndexedVec, PersistedValidationData, ScheduledCore, SessionIndex, SessionInfo, ValidatorPair,
+	vstaging::CommittedCandidateReceiptV2 as CommittedCandidateReceipt, AssignmentPair,
+	AsyncBackingParams, Block, BlockNumber, GroupRotationInfo, HeadData, Header, IndexedVec,
+	PersistedValidationData, SessionIndex, SessionInfo, ValidatorPair,
 };
 use sc_keystore::LocalKeystore;
 use sc_network::ProtocolName;
@@ -82,6 +82,8 @@ struct TestConfig {
 	// whether the local node should be a validator
 	local_validator: LocalRole,
 	async_backing_params: Option<AsyncBackingParams>,
+	// allow v2 descriptors (feature bit)
+	allow_v2_descriptors: bool,
 }
 
 #[derive(Debug, Clone)]
@@ -96,6 +98,7 @@ struct TestState {
 	validators: Vec<ValidatorPair>,
 	session_info: SessionInfo,
 	req_sender: async_channel::Sender<sc_network::config::IncomingRequest>,
+	node_features: NodeFeatures,
 }
 
 impl TestState {
@@ -174,7 +177,13 @@ impl TestState {
 			random_seed: [0u8; 32],
 		};
 
-		TestState { config, local, validators, session_info, req_sender }
+		let mut node_features = NodeFeatures::new();
+		if config.allow_v2_descriptors {
+			node_features.resize(FeatureIndex::FirstUnassigned as usize, false);
+			node_features.set(FeatureIndex::CandidateReceiptV2 as usize, true);
+		}
+
+		TestState { config, local, validators, session_info, req_sender, node_features }
 	}
 
 	fn make_dummy_leaf(&self, relay_parent: Hash) -> TestLeaf {
@@ -186,20 +195,23 @@ impl TestState {
 		relay_parent: Hash,
 		groups_for_first_para: usize,
 	) -> TestLeaf {
+		let mut cq = std::collections::BTreeMap::new();
+
+		for i in 0..self.session_info.validator_groups.len() {
+			if i < groups_for_first_para {
+				cq.entry(CoreIndex(i as u32))
+					.or_insert_with(|| vec![ParaId::from(0u32), ParaId::from(0u32)].into());
+			} else {
+				cq.entry(CoreIndex(i as u32))
+					.or_insert_with(|| vec![ParaId::from(i), ParaId::from(i)].into());
+			};
+		}
+
 		TestLeaf {
 			number: 1,
 			hash: relay_parent,
 			parent_hash: Hash::repeat_byte(0),
 			session: 1,
-			availability_cores: self.make_availability_cores(|i| {
-				let para_id = if i < groups_for_first_para {
-					ParaId::from(0u32)
-				} else {
-					ParaId::from(i as u32)
-				};
-
-				CoreState::Scheduled(ScheduledCore { para_id, collator: None })
-			}),
 			disabled_validators: Default::default(),
 			para_data: (0..self.session_info.validator_groups.len())
 				.map(|i| {
@@ -213,6 +225,7 @@ impl TestState {
 				})
 				.collect(),
 			minimum_backing_votes: 2,
+			claim_queue: ClaimQueueSnapshot(cq),
 		}
 	}
 
@@ -232,10 +245,6 @@ impl TestState {
 		TestLeaf { minimum_backing_votes, ..self.make_dummy_leaf(relay_parent) }
 	}
 
-	fn make_availability_cores(&self, f: impl Fn(usize) -> CoreState) -> Vec<CoreState> {
-		(0..self.session_info.validator_groups.len()).map(f).collect()
-	}
-
 	fn make_dummy_topology(&self) -> NewGossipTopology {
 		let validator_count = self.config.validator_count;
 		let is_local_inactive = matches!(self.config.local_validator, LocalRole::InactiveValidator);
@@ -423,10 +432,10 @@ struct TestLeaf {
 	hash: Hash,
 	parent_hash: Hash,
 	session: SessionIndex,
-	availability_cores: Vec<CoreState>,
 	pub disabled_validators: Vec<ValidatorIndex>,
 	para_data: Vec<(ParaId, PerParaData)>,
 	minimum_backing_votes: u32,
+	claim_queue: ClaimQueueSnapshot,
 }
 
 impl TestLeaf {
@@ -574,9 +583,9 @@ async fn handle_leaf_activation(
 		parent_hash,
 		para_data,
 		session,
-		availability_cores,
 		disabled_validators,
 		minimum_backing_votes,
+		claim_queue,
 	} = leaf;
 
 	assert_matches!(
@@ -623,7 +632,7 @@ async fn handle_leaf_activation(
 				_parent,
 				RuntimeApiRequest::Version(tx),
 			)) => {
-				tx.send(Ok(RuntimeApiRequest::DISABLED_VALIDATORS_RUNTIME_REQUIREMENT)).unwrap();
+				tx.send(Ok(RuntimeApiRequest::CLAIM_QUEUE_RUNTIME_REQUIREMENT)).unwrap();
 			},
 			AllMessages::RuntimeApi(RuntimeApiMessage::Request(
 				parent,
@@ -657,12 +666,6 @@ async fn handle_leaf_activation(
 				assert!(is_new_session, "only expecting this call in a new session");
 				tx.send(Ok(*minimum_backing_votes)).unwrap();
 			},
-			AllMessages::RuntimeApi(RuntimeApiMessage::Request(
-				parent,
-				RuntimeApiRequest::AvailabilityCores(tx),
-			)) if parent == *hash => {
-				tx.send(Ok(availability_cores.clone())).unwrap();
-			},
 			AllMessages::RuntimeApi(RuntimeApiMessage::Request(
 				parent,
 				RuntimeApiRequest::ValidatorGroups(tx),
@@ -675,6 +678,18 @@ async fn handle_leaf_activation(
 				};
 				tx.send(Ok((validator_groups, group_rotation_info))).unwrap();
 			},
+			AllMessages::RuntimeApi(RuntimeApiMessage::Request(
+				parent,
+				RuntimeApiRequest::NodeFeatures(_session_index, tx),
+			)) if parent == *hash => {
+				tx.send(Ok(test_state.node_features.clone())).unwrap();
+			},
+			AllMessages::RuntimeApi(RuntimeApiMessage::Request(
+				parent,
+				RuntimeApiRequest::ClaimQueue(tx),
+			)) if parent == *hash => {
+				tx.send(Ok(claim_queue.0.clone())).unwrap();
+			},
 			AllMessages::ProspectiveParachains(
 				ProspectiveParachainsMessage::GetHypotheticalMembership(req, tx),
 			) => {
diff --git a/polkadot/node/network/statement-distribution/src/v2/tests/requests.rs b/polkadot/node/network/statement-distribution/src/v2/tests/requests.rs
index dcb90bacdcde..fc880c1d9a83 100644
--- a/polkadot/node/network/statement-distribution/src/v2/tests/requests.rs
+++ b/polkadot/node/network/statement-distribution/src/v2/tests/requests.rs
@@ -21,19 +21,25 @@ use codec::{Decode, Encode};
 use polkadot_node_network_protocol::{
 	request_response::v2 as request_v2, v2::BackedCandidateManifest,
 };
-use polkadot_primitives_test_helpers::make_candidate;
+use polkadot_primitives_test_helpers::{make_candidate, make_candidate_v2};
 use sc_network::config::{
 	IncomingRequest as RawIncomingRequest, OutgoingResponse as RawOutgoingResponse,
 };
 
-#[test]
-fn cluster_peer_allowed_to_send_incomplete_statements() {
+use polkadot_primitives::vstaging::MutateDescriptorV2;
+use rstest::rstest;
+
+#[rstest]
+#[case(false)]
+#[case(true)]
+fn cluster_peer_allowed_to_send_incomplete_statements(#[case] allow_v2_descriptors: bool) {
 	let group_size = 3;
 	let config = TestConfig {
 		validator_count: 20,
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -48,14 +54,28 @@ fn cluster_peer_allowed_to_send_incomplete_statements() {
 
 		let test_leaf = state.make_dummy_leaf(relay_parent);
 
-		let (candidate, pvd) = make_candidate(
-			relay_parent,
-			1,
-			local_para,
-			test_leaf.para_data(local_para).head_data.clone(),
-			vec![4, 5, 6].into(),
-			Hash::repeat_byte(42).into(),
-		);
+		let (candidate, pvd) = if allow_v2_descriptors {
+			let (mut candidate, pvd) = make_candidate_v2(
+				relay_parent,
+				1,
+				local_para,
+				test_leaf.para_data(local_para).head_data.clone(),
+				vec![4, 5, 6].into(),
+				Hash::repeat_byte(42).into(),
+			);
+			candidate.descriptor.set_core_index(CoreIndex(local_group_index.0));
+			(candidate, pvd)
+		} else {
+			make_candidate(
+				relay_parent,
+				1,
+				local_para,
+				test_leaf.para_data(local_para).head_data.clone(),
+				vec![4, 5, 6].into(),
+				Hash::repeat_byte(42).into(),
+			)
+		};
+
 		let candidate_hash = candidate.hash();
 
 		let other_group_validators = state.group_validators(local_group_index, true);
@@ -187,6 +207,7 @@ fn peer_reported_for_providing_statements_meant_to_be_masked_out() {
 			max_candidate_depth: 1,
 			allowed_ancestry_len: 3,
 		}),
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -462,6 +483,7 @@ fn peer_reported_for_not_enough_statements() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -649,6 +671,7 @@ fn peer_reported_for_duplicate_statements() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -802,6 +825,7 @@ fn peer_reported_for_providing_statements_with_invalid_signatures() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -925,6 +949,415 @@ fn peer_reported_for_providing_statements_with_invalid_signatures() {
 	});
 }
 
+#[test]
+fn peer_reported_for_invalid_v2_descriptor() {
+	let group_size = 3;
+	let config = TestConfig {
+		validator_count: 20,
+		group_size,
+		local_validator: LocalRole::Validator,
+		async_backing_params: None,
+		allow_v2_descriptors: true,
+	};
+
+	let relay_parent = Hash::repeat_byte(1);
+	let peer_a = PeerId::random();
+	let peer_b = PeerId::random();
+	let peer_c = PeerId::random();
+
+	test_harness(config, |state, mut overseer| async move {
+		let local_validator = state.local.clone().unwrap();
+		let local_group_index = local_validator.group_index.unwrap();
+		let local_para = ParaId::from(local_group_index.0);
+
+		let test_leaf = state.make_dummy_leaf(relay_parent);
+
+		let (mut candidate, pvd) = make_candidate_v2(
+			relay_parent,
+			1,
+			local_para,
+			test_leaf.para_data(local_para).head_data.clone(),
+			vec![4, 5, 6].into(),
+			Hash::repeat_byte(42).into(),
+		);
+
+		candidate.descriptor.set_core_index(CoreIndex(100));
+
+		let candidate_hash = candidate.hash();
+
+		let other_group_validators = state.group_validators(local_group_index, true);
+		let v_a = other_group_validators[0];
+		let v_b = other_group_validators[1];
+		let v_c = other_group_validators[1];
+
+		// peer A is in group, has relay parent in view.
+		// peer B is in group, has no relay parent in view.
+		// peer C is not in group, has relay parent in view.
+		{
+			connect_peer(
+				&mut overseer,
+				peer_a.clone(),
+				Some(vec![state.discovery_id(other_group_validators[0])].into_iter().collect()),
+			)
+			.await;
+
+			connect_peer(
+				&mut overseer,
+				peer_b.clone(),
+				Some(vec![state.discovery_id(other_group_validators[1])].into_iter().collect()),
+			)
+			.await;
+
+			connect_peer(&mut overseer, peer_c.clone(), None).await;
+
+			send_peer_view_change(&mut overseer, peer_a.clone(), view![relay_parent]).await;
+			send_peer_view_change(&mut overseer, peer_c.clone(), view![relay_parent]).await;
+		}
+
+		activate_leaf(&mut overseer, &test_leaf, &state, true, vec![]).await;
+
+		// Peer in cluster sends a statement, triggering a request.
+		{
+			let a_seconded = state
+				.sign_statement(
+					v_a,
+					CompactStatement::Seconded(candidate_hash),
+					&SigningContext { parent_hash: relay_parent, session_index: 1 },
+				)
+				.as_unchecked()
+				.clone();
+
+			send_peer_message(
+				&mut overseer,
+				peer_a.clone(),
+				protocol_v2::StatementDistributionMessage::Statement(relay_parent, a_seconded),
+			)
+			.await;
+
+			assert_matches!(
+				overseer.recv().await,
+				AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::ReportPeer(ReportPeerMessage::Single(p, r)))
+					if p == peer_a && r == BENEFIT_VALID_STATEMENT_FIRST.into() => { }
+			);
+		}
+
+		// Send a request to peer and mock its response to include a candidate with invalid core
+		// index.
+		{
+			let b_seconded_invalid = state
+				.sign_statement(
+					v_b,
+					CompactStatement::Seconded(candidate_hash),
+					&SigningContext { parent_hash: relay_parent, session_index: 1 },
+				)
+				.as_unchecked()
+				.clone();
+			let statements = vec![b_seconded_invalid.clone()];
+
+			handle_sent_request(
+				&mut overseer,
+				peer_a,
+				candidate_hash,
+				StatementFilter::blank(group_size),
+				candidate.clone(),
+				pvd.clone(),
+				statements,
+			)
+			.await;
+
+			assert_matches!(
+				overseer.recv().await,
+				AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::ReportPeer(ReportPeerMessage::Single(p, r)))
+					if p == peer_a && r == COST_INVALID_CORE_INDEX.into() => { }
+			);
+		}
+
+		// Test invalid session index
+		candidate.descriptor.set_session_index(100);
+		// Set good core index
+		candidate.descriptor.set_core_index(CoreIndex(local_group_index.0));
+
+		let candidate_hash = candidate.hash();
+
+		// Peer in cluster sends a statement, triggering a request.
+		{
+			let a_seconded = state
+				.sign_statement(
+					v_a,
+					CompactStatement::Seconded(candidate_hash),
+					&SigningContext { parent_hash: relay_parent, session_index: 1 },
+				)
+				.as_unchecked()
+				.clone();
+
+			send_peer_message(
+				&mut overseer,
+				peer_a.clone(),
+				protocol_v2::StatementDistributionMessage::Statement(relay_parent, a_seconded),
+			)
+			.await;
+
+			assert_matches!(
+				overseer.recv().await,
+				AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::ReportPeer(ReportPeerMessage::Single(p, r)))
+					if p == peer_a && r == BENEFIT_VALID_STATEMENT_FIRST.into() => { }
+			);
+		}
+
+		// Send a request to peer and mock its response to include a candidate with invalid session
+		// index.
+		{
+			let b_seconded_invalid = state
+				.sign_statement(
+					v_b,
+					CompactStatement::Seconded(candidate_hash),
+					&SigningContext { parent_hash: relay_parent, session_index: 1 },
+				)
+				.as_unchecked()
+				.clone();
+			let statements = vec![b_seconded_invalid.clone()];
+
+			handle_sent_request(
+				&mut overseer,
+				peer_a,
+				candidate_hash,
+				StatementFilter::blank(group_size),
+				candidate.clone(),
+				pvd.clone(),
+				statements,
+			)
+			.await;
+
+			assert_matches!(
+				overseer.recv().await,
+				AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::ReportPeer(ReportPeerMessage::Single(p, r)))
+					if p == peer_a && r == COST_INVALID_SESSION_INDEX.into() => { }
+			);
+		}
+
+		// Test valid candidate does not lead to punishment
+		candidate.descriptor.set_session_index(1);
+
+		let candidate_hash = candidate.hash();
+
+		// Peer in cluster sends a statement, triggering a request.
+		{
+			let a_seconded = state
+				.sign_statement(
+					v_a,
+					CompactStatement::Seconded(candidate_hash),
+					&SigningContext { parent_hash: relay_parent, session_index: 1 },
+				)
+				.as_unchecked()
+				.clone();
+
+			send_peer_message(
+				&mut overseer,
+				peer_a.clone(),
+				protocol_v2::StatementDistributionMessage::Statement(relay_parent, a_seconded),
+			)
+			.await;
+
+			assert_matches!(
+				overseer.recv().await,
+				AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::ReportPeer(ReportPeerMessage::Single(p, r)))
+					if p == peer_a && r == BENEFIT_VALID_STATEMENT_FIRST.into() => { }
+			);
+		}
+
+		// Send a request to peer and mock its response to include a valid candidate.
+		{
+			let b_seconded_invalid = state
+				.sign_statement(
+					v_b,
+					CompactStatement::Seconded(candidate_hash),
+					&SigningContext { parent_hash: relay_parent, session_index: 1 },
+				)
+				.as_unchecked()
+				.clone();
+			let statements = vec![b_seconded_invalid.clone()];
+
+			handle_sent_request(
+				&mut overseer,
+				peer_a,
+				candidate_hash,
+				StatementFilter::blank(group_size),
+				candidate.clone(),
+				pvd.clone(),
+				statements,
+			)
+			.await;
+
+			assert_matches!(
+				overseer.recv().await,
+				AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::ReportPeer(ReportPeerMessage::Single(p, r)))
+					if p == peer_a && r == BENEFIT_VALID_STATEMENT.into() => { }
+			);
+			assert_matches!(
+				overseer.recv().await,
+				AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::ReportPeer(ReportPeerMessage::Single(p, r)))
+					if p == peer_a && r == BENEFIT_VALID_RESPONSE.into() => { }
+			);
+
+			assert_matches!(
+				overseer.recv().await,
+				AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
+					peers,
+					Versioned::V2(protocol_v2::ValidationProtocol::StatementDistribution(
+						protocol_v2::StatementDistributionMessage::Statement(
+							r,
+							s,
+						)
+					))
+				)) => {
+					assert_eq!(peers, vec![peer_a.clone()]);
+					assert_eq!(r, relay_parent);
+					assert_eq!(s.unchecked_payload(), &CompactStatement::Seconded(candidate_hash));
+					assert_eq!(s.unchecked_validator_index(), v_c);
+				}
+			);
+
+			answer_expected_hypothetical_membership_request(&mut overseer, vec![]).await;
+		}
+		overseer
+	});
+}
+
+#[rstest]
+#[case(false)]
+#[case(true)]
+// Test if v2 descriptors are filtered and peers punished if the node feature is disabled.
+// Also test if the peer is rewarded for providing v2 descriptor if the node feature is enabled.
+fn v2_descriptors_filtered(#[case] allow_v2_descriptors: bool) {
+	let group_size = 3;
+	let config = TestConfig {
+		validator_count: 20,
+		group_size,
+		local_validator: LocalRole::Validator,
+		async_backing_params: None,
+		allow_v2_descriptors,
+	};
+
+	let relay_parent = Hash::repeat_byte(1);
+	let peer_a = PeerId::random();
+	let peer_b = PeerId::random();
+	let peer_c = PeerId::random();
+
+	test_harness(config, |state, mut overseer| async move {
+		let local_validator = state.local.clone().unwrap();
+		let local_group_index = local_validator.group_index.unwrap();
+		let local_para = ParaId::from(local_group_index.0);
+
+		let test_leaf = state.make_dummy_leaf(relay_parent);
+
+		let (mut candidate, pvd) = make_candidate_v2(
+			relay_parent,
+			1,
+			local_para,
+			test_leaf.para_data(local_para).head_data.clone(),
+			vec![4, 5, 6].into(),
+			Hash::repeat_byte(42).into(),
+		);
+
+		// Makes the candidate invalid.
+		candidate.descriptor.set_core_index(CoreIndex(100));
+
+		let candidate_hash = candidate.hash();
+
+		let other_group_validators = state.group_validators(local_group_index, true);
+		let v_a = other_group_validators[0];
+		let v_b = other_group_validators[1];
+
+		// peer A is in group, has relay parent in view.
+		// peer B is in group, has no relay parent in view.
+		// peer C is not in group, has relay parent in view.
+		{
+			connect_peer(
+				&mut overseer,
+				peer_a.clone(),
+				Some(vec![state.discovery_id(other_group_validators[0])].into_iter().collect()),
+			)
+			.await;
+
+			connect_peer(
+				&mut overseer,
+				peer_b.clone(),
+				Some(vec![state.discovery_id(other_group_validators[1])].into_iter().collect()),
+			)
+			.await;
+
+			connect_peer(&mut overseer, peer_c.clone(), None).await;
+
+			send_peer_view_change(&mut overseer, peer_a.clone(), view![relay_parent]).await;
+			send_peer_view_change(&mut overseer, peer_c.clone(), view![relay_parent]).await;
+		}
+
+		activate_leaf(&mut overseer, &test_leaf, &state, true, vec![]).await;
+
+		// Peer in cluster sends a statement, triggering a request.
+		{
+			let a_seconded = state
+				.sign_statement(
+					v_a,
+					CompactStatement::Seconded(candidate_hash),
+					&SigningContext { parent_hash: relay_parent, session_index: 1 },
+				)
+				.as_unchecked()
+				.clone();
+
+			send_peer_message(
+				&mut overseer,
+				peer_a.clone(),
+				protocol_v2::StatementDistributionMessage::Statement(relay_parent, a_seconded),
+			)
+			.await;
+
+			assert_matches!(
+				overseer.recv().await,
+				AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::ReportPeer(ReportPeerMessage::Single(p, r)))
+					if p == peer_a && r == BENEFIT_VALID_STATEMENT_FIRST.into() => { }
+			);
+		}
+
+		// Send a request to peer and mock its response to include a candidate with invalid core
+		// index.
+		{
+			let b_seconded_invalid = state
+				.sign_statement(
+					v_b,
+					CompactStatement::Seconded(candidate_hash),
+					&SigningContext { parent_hash: relay_parent, session_index: 1 },
+				)
+				.as_unchecked()
+				.clone();
+			let statements = vec![b_seconded_invalid.clone()];
+
+			handle_sent_request(
+				&mut overseer,
+				peer_a,
+				candidate_hash,
+				StatementFilter::blank(group_size),
+				candidate.clone(),
+				pvd.clone(),
+				statements,
+			)
+			.await;
+
+			let expected_rep_change = if allow_v2_descriptors {
+				COST_INVALID_CORE_INDEX.into()
+			} else {
+				COST_UNSUPPORTED_DESCRIPTOR_VERSION.into()
+			};
+			assert_matches!(
+				overseer.recv().await,
+				AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::ReportPeer(ReportPeerMessage::Single(p, r)))
+					if p == peer_a && r == expected_rep_change => { }
+			);
+		}
+
+		overseer
+	});
+}
 #[test]
 fn peer_reported_for_providing_statements_with_wrong_validator_id() {
 	let group_size = 3;
@@ -933,6 +1366,7 @@ fn peer_reported_for_providing_statements_with_wrong_validator_id() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -1063,6 +1497,7 @@ fn disabled_validators_added_to_unwanted_mask() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -1229,6 +1664,7 @@ fn disabling_works_from_relay_parent_not_the_latest_state() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_1 = Hash::repeat_byte(1);
@@ -1428,6 +1864,7 @@ fn local_node_sanity_checks_incoming_requests() {
 		group_size: 3,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -1629,6 +2066,7 @@ fn local_node_checks_that_peer_can_request_before_responding() {
 		group_size: 3,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -1828,6 +2266,7 @@ fn local_node_respects_statement_mask() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
@@ -2070,6 +2509,7 @@ fn should_delay_before_retrying_dropped_requests() {
 		group_size,
 		local_validator: LocalRole::Validator,
 		async_backing_params: None,
+		allow_v2_descriptors: false,
 	};
 
 	let relay_parent = Hash::repeat_byte(1);
diff --git a/polkadot/primitives/src/vstaging/mod.rs b/polkadot/primitives/src/vstaging/mod.rs
index 94b7b200e68f..ca9c3e1bebad 100644
--- a/polkadot/primitives/src/vstaging/mod.rs
+++ b/polkadot/primitives/src/vstaging/mod.rs
@@ -204,6 +204,8 @@ pub trait MutateDescriptorV2<H> {
 	fn set_version(&mut self, version: InternalVersion);
 	/// Set the PVD of the descriptor.
 	fn set_persisted_validation_data_hash(&mut self, persisted_validation_data_hash: Hash);
+	/// Set the validation code hash of the descriptor.
+	fn set_validation_code_hash(&mut self, validation_code_hash: ValidationCodeHash);
 	/// Set the erasure root of the descriptor.
 	fn set_erasure_root(&mut self, erasure_root: Hash);
 	/// Set the para head of the descriptor.
@@ -244,6 +246,10 @@ impl<H> MutateDescriptorV2<H> for CandidateDescriptorV2<H> {
 		self.persisted_validation_data_hash = persisted_validation_data_hash;
 	}
 
+	fn set_validation_code_hash(&mut self, validation_code_hash: ValidationCodeHash) {
+		self.validation_code_hash = validation_code_hash;
+	}
+
 	fn set_erasure_root(&mut self, erasure_root: Hash) {
 		self.erasure_root = erasure_root;
 	}
diff --git a/polkadot/primitives/test-helpers/src/lib.rs b/polkadot/primitives/test-helpers/src/lib.rs
index c2eccafef788..1717dd5b0eda 100644
--- a/polkadot/primitives/test-helpers/src/lib.rs
+++ b/polkadot/primitives/test-helpers/src/lib.rs
@@ -23,13 +23,15 @@
 //! Note that `dummy_` prefixed values are meant to be fillers, that should not matter, and will
 //! contain randomness based data.
 use polkadot_primitives::{
-	vstaging::{CandidateDescriptorV2, CandidateReceiptV2, CommittedCandidateReceiptV2},
+	vstaging::{
+		CandidateDescriptorV2, CandidateReceiptV2, CommittedCandidateReceiptV2, MutateDescriptorV2,
+	},
 	CandidateCommitments, CandidateDescriptor, CandidateReceipt, CollatorId, CollatorSignature,
 	CommittedCandidateReceipt, CoreIndex, Hash, HeadData, Id as ParaId, PersistedValidationData,
 	SessionIndex, ValidationCode, ValidationCodeHash, ValidatorId,
 };
 pub use rand;
-use sp_application_crypto::sr25519;
+use sp_application_crypto::{sr25519, ByteArray};
 use sp_keyring::Sr25519Keyring;
 use sp_runtime::generic::Digest;
 
@@ -199,8 +201,15 @@ pub fn dummy_collator() -> CollatorId {
 	CollatorId::from(sr25519::Public::default())
 }
 
-/// Create a meaningless collator signature.
+/// Create a meaningless collator signature. It is important to not be 0, as we'd confuse
+/// v1 and v2 descriptors.
 pub fn dummy_collator_signature() -> CollatorSignature {
+	CollatorSignature::from_slice(&mut (0..64).into_iter().collect::<Vec<_>>().as_slice())
+		.expect("64 bytes; qed")
+}
+
+/// Create a zeroed collator signature.
+pub fn zero_collator_signature() -> CollatorSignature {
 	CollatorSignature::from(sr25519::Signature::default())
 }
 
@@ -250,6 +259,34 @@ pub fn make_candidate(
 	(candidate, pvd)
 }
 
+/// Create a meaningless v2 candidate, returning its receipt and PVD.
+pub fn make_candidate_v2(
+	relay_parent_hash: Hash,
+	relay_parent_number: u32,
+	para_id: ParaId,
+	parent_head: HeadData,
+	head_data: HeadData,
+	validation_code_hash: ValidationCodeHash,
+) -> (CommittedCandidateReceiptV2, PersistedValidationData) {
+	let pvd = dummy_pvd(parent_head, relay_parent_number);
+	let commitments = CandidateCommitments {
+		head_data,
+		horizontal_messages: Default::default(),
+		upward_messages: Default::default(),
+		new_validation_code: None,
+		processed_downward_messages: 0,
+		hrmp_watermark: relay_parent_number,
+	};
+
+	let mut descriptor = dummy_candidate_descriptor_v2(relay_parent_hash);
+	descriptor.set_para_id(para_id);
+	descriptor.set_persisted_validation_data_hash(pvd.hash());
+	descriptor.set_validation_code_hash(validation_code_hash);
+	let candidate = CommittedCandidateReceiptV2 { descriptor, commitments };
+
+	(candidate, pvd)
+}
+
 /// Create a new candidate descriptor, and apply a valid signature
 /// using the provided `collator` key.
 pub fn make_valid_candidate_descriptor<H: AsRef<[u8]>>(
diff --git a/prdoc/pr_5883.prdoc b/prdoc/pr_5883.prdoc
new file mode 100644
index 000000000000..96225a89bc99
--- /dev/null
+++ b/prdoc/pr_5883.prdoc
@@ -0,0 +1,15 @@
+title: 'statement-distribution RFC103 implementation'
+
+doc:
+  - audience: Node Dev
+    description: |
+       Introduces checks for the new candidate descriptor fields: `core_index` and `session_index`.
+
+crates:
+  - name: polkadot-statement-distribution
+    bump: minor
+  - name: polkadot-primitives
+    bump: major
+  - name: polkadot-primitives-test-helpers
+    bump: major
+