diff --git a/cumulus/client/consensus/aura/src/collators/slot_based/block_builder_task.rs b/cumulus/client/consensus/aura/src/collators/slot_based/block_builder_task.rs
index 88d68611a4de5..8c3479d32df2f 100644
--- a/cumulus/client/consensus/aura/src/collators/slot_based/block_builder_task.rs
+++ b/cumulus/client/consensus/aura/src/collators/slot_based/block_builder_task.rs
@@ -15,7 +15,7 @@
 // You should have received a copy of the GNU General Public License
 // along with Cumulus. If not, see <https://www.gnu.org/licenses/>.
 
-use super::CollatorMessage;
+use super::{CollatorMessage, CollatorResubmitSegment, SegmentKind};
 use crate::{
 	collator::{self as collator_util, BuildBlockAndImportParams, Collator, SlotClaim},
 	collators::{
@@ -108,8 +108,10 @@ pub struct BuilderTaskParams<
 	pub proposer: Proposer,
 	/// The generic collator service used to plug into this consensus engine.
 	pub collator_service: CS,
-	/// Channel to send built blocks to the collation task.
+	/// Channel for V2 single-bundle collations.
 	pub collator_sender: sc_utils::mpsc::TracingUnboundedSender<CollatorMessage<Block>>,
+	/// Channel for V3 segment collations.
+	pub resubmit_sender: sc_utils::mpsc::TracingUnboundedSender<CollatorResubmitSegment<Block>>,
 	/// Slot duration of the relay chain.
 	pub relay_chain_slot_duration: Duration,
 	/// Offset all time operations by this duration.
@@ -185,6 +187,7 @@ where
 			proposer,
 			collator_service,
 			collator_sender,
+			resubmit_sender,
 			code_hash_provider,
 			relay_chain_slot_duration,
 			para_backend,
@@ -274,7 +277,7 @@ where
 					.await
 					.unwrap_or(0);
 			}
-			let Ok(Some(relay_parent_data)) = offset_relay_parent_find_descendants(
+			let Ok(Some(mut relay_parent_data)) = offset_relay_parent_find_descendants(
 				&mut relay_chain_data_cache,
 				scheduling_parent_header.clone(),
 				relay_parent_offset,
@@ -484,12 +487,43 @@ where
 				"Core configuration",
 			);
 
+			// Build the V3 scheduling proof once per slot — it's shared by every candidate in the
+			// segment. Consumes `relay_parent_data.descendants`, so the inherent created later
+			// downstream sees an empty list (which V3 verification expects).
+			let scheduling_proof = if v3_enabled {
+				let descendants = relay_parent_data.take_descendants();
+				let header_chain: Vec<_> = descendants.into_iter().rev().collect();
+				let scheduling_parent =
+					header_chain.first().map(|header| header.hash()).unwrap_or(relay_parent_hash);
+
+				tracing::debug!(
+					target: LOG_TARGET,
+					relay_parent = ?relay_parent_hash,
+					?scheduling_parent,
+					header_chain_len = header_chain.len(),
+					"Building V3 collation with scheduling proof",
+				);
+
+				Some(SchedulingProof {
+					header_chain,
+					internal_scheduling_parent_header: relay_parent_header.clone(),
+					// TODO: sign and populate when resubmission lands. The relay-chain verifier
+					// rejects `ResubmitOnly` / mixed segments whose proof lacks
+					// `signed_scheduling_info`.
+					signed_scheduling_info: None,
+				})
+			} else {
+				None
+			};
+
 			let mut pov_parent_header = initial_parent_header;
 			let mut pov_parent_hash = initial_parent_hash;
 			let block_time = relay_chain_slot_duration / number_of_blocks;
 
-			for blocks_per_core in blocks_per_cores {
+			for (core_iter_index, blocks_per_core) in blocks_per_cores.into_iter().enumerate() {
 				let time_for_core = slot_time.time_left() / cores.cores_left();
+				let is_first_core = core_iter_index == 0;
+				let this_core_index = cores.core_index();
 
 				match build_collation_for_core(BuildCollationParams {
 					pov_parent_header,
@@ -502,10 +536,11 @@ where
 					code_hash_provider: &code_hash_provider,
 					slot_claim: &slot_claim,
 					collator_sender: &collator_sender,
+					resubmit_sender: &resubmit_sender,
 					collator: &mut collator,
 					allowed_pov_size,
 					core_info: cores.core_info(),
-					core_index: cores.core_index(),
+					core_index: this_core_index,
 					block_time,
 					blocks_per_core,
 					time_for_core,
@@ -518,7 +553,7 @@ where
 					relay_slot,
 					para_slot: para_slot.slot,
 					para_client: &*para_client,
-					v3_enabled,
+					scheduling_proof: scheduling_proof.clone(),
 				})
 				.await
 				{
@@ -526,8 +561,25 @@ where
 						pov_parent_header = header;
 						pov_parent_hash = pov_parent_header.hash();
 					},
-					// Let's wait for the next slot
-					Ok(None) => break,
+					Ok(None) => {
+						// First-core failed to build a fresh bundle: still ship the held prior
+						// unincluded segment via `ResubmitOnly`.
+						//
+						// TODO: seding to a certain core can apply in a custom way too, and
+						// other strategies for resubmitting a segment should be available - e.g.
+						// split segment between cores equally, or attempt grouping unincluded
+						// blocks based on the core index resulting from their `core_selector`
+						// and `claim_queue_offset` combination, at the `scheduling_parent`).
+						if is_first_core {
+							if let Some(proof) = scheduling_proof.clone() {
+								let _ = resubmit_sender.unbounded_send(CollatorResubmitSegment {
+									scheduling_proof: proof,
+									kind: SegmentKind::ResubmitOnly { core_index: this_core_index },
+								});
+							}
+						}
+						break;
+					},
 					Err(()) => return,
 				}
 
@@ -562,6 +614,7 @@ struct BuildCollationParams<
 	code_hash_provider: &'a CHP,
 	slot_claim: &'a SlotClaim<P::Public>,
 	collator_sender: &'a sc_utils::mpsc::TracingUnboundedSender<CollatorMessage<Block>>,
+	resubmit_sender: &'a sc_utils::mpsc::TracingUnboundedSender<CollatorResubmitSegment<Block>>,
 	collator: &'a mut Collator<Block, P, BI, CIDP, RelayClient, Proposer, CS>,
 	allowed_pov_size: usize,
 	core_info: CoreInfo,
@@ -578,7 +631,8 @@ struct BuildCollationParams<
 	relay_slot: cumulus_primitives_aura::Slot,
 	para_slot: cumulus_primitives_aura::Slot,
 	para_client: &'a Client,
-	v3_enabled: bool,
+	/// V3 scheduling proof, built once per slot. `Some` iff V3 is enabled.
+	scheduling_proof: Option<SchedulingProof>,
 }
 
 /// Build a collation for one core.
@@ -606,6 +660,7 @@ async fn build_collation_for_core<
 		code_hash_provider,
 		slot_claim,
 		collator_sender,
+		resubmit_sender,
 		collator,
 		allowed_pov_size,
 		core_info,
@@ -615,13 +670,13 @@ async fn build_collation_for_core<
 		time_for_core: slot_time_for_core,
 		is_last_core_in_parachain_slot,
 		collator_peer_id,
-		mut relay_parent_data,
+		relay_parent_data,
 		total_number_of_blocks,
 		included_header_hash,
 		relay_slot,
 		para_slot,
 		para_client,
-		v3_enabled,
+		scheduling_proof,
 	}: BuildCollationParams<'_, Block, P, RelayClient, BI, CIDP, Proposer, CS, CHP, Client>,
 ) -> Result<Option<Block::Header>, ()>
 where
@@ -649,34 +704,6 @@ where
 		max_pov_size,
 	};
 
-	// Check if V3 scheduling is enabled and build scheduling proof if so.
-	let mut scheduling_proof = None;
-	if v3_enabled {
-		// The relay parent descendants are only needed for v2.
-		let descendants = relay_parent_data.take_descendants();
-		// The descendants are ordered from oldest to newest, so we need to reverse them.
-		let header_chain: Vec<_> = descendants.into_iter().rev().collect();
-		let scheduling_parent =
-			header_chain.first().map(|header| header.hash()).unwrap_or(relay_parent_hash);
-
-		tracing::debug!(
-			target: LOG_TARGET,
-			relay_parent = ?relay_parent_hash,
-			?scheduling_parent,
-			header_chain_len = header_chain.len(),
-			"Building V3 collation with scheduling proof",
-		);
-
-		scheduling_proof = Some(SchedulingProof {
-			header_chain,
-			// Initial submission: internal_scheduling_parent == relay_parent, so the
-			// internal scheduling parent header is the relay parent's header itself.
-			internal_scheduling_parent_header: relay_parent_header.clone(),
-			// Initial submission: no signature needed, core selection from UMP signals
-			signed_scheduling_info: None,
-		});
-	}
-
 	let Some(validation_code_hash) = code_hash_provider.code_hash_at(pov_parent_hash) else {
 		tracing::error!(
 			target: LOG_TARGET,
@@ -902,17 +929,41 @@ where
 		"Sending out PoV"
 	);
 
-	if let Err(err) = collator_sender.unbounded_send(CollatorMessage {
-		relay_parent: relay_parent_hash,
-		scheduling_proof,
-		parent_header: pov_parent_header.clone(),
-		blocks,
-		proof,
-		validation_code_hash,
-		core_index,
-		validation_data,
-	}) {
-		tracing::error!(target: LOG_TARGET, ?err, "Unable to send block to collation task.");
+	// `scheduling_proof` is `Some` iff `v3_enabled`; routes V3 bundles to `resubmit_sender`,
+	// V2 bundles (no proof) to `collator_sender`.
+	let send_ok = if let Some(scheduling_proof) = scheduling_proof {
+		resubmit_sender
+			.unbounded_send(CollatorResubmitSegment {
+				scheduling_proof,
+				kind: SegmentKind::WithBundle {
+					bundle: CollatorMessage {
+						relay_parent: relay_parent_hash,
+						parent_header: pov_parent_header.clone(),
+						blocks,
+						proof,
+						validation_code_hash,
+						validation_data,
+						core_index,
+					},
+				},
+			})
+			.is_ok()
+	} else {
+		collator_sender
+			.unbounded_send(CollatorMessage {
+				core_index,
+				relay_parent: relay_parent_hash,
+				parent_header: pov_parent_header.clone(),
+				blocks,
+				proof,
+				validation_code_hash,
+				validation_data,
+			})
+			.is_ok()
+	};
+
+	if !send_ok {
+		tracing::error!(target: LOG_TARGET, "Unable to send block to collation task.");
 		Err(())
 	} else {
 		// Now let's sleep for the rest of the core.
diff --git a/cumulus/client/consensus/aura/src/collators/slot_based/collation_task.rs b/cumulus/client/consensus/aura/src/collators/slot_based/collation_task.rs
index 2ddff00c40421..4b572b568664b 100644
--- a/cumulus/client/consensus/aura/src/collators/slot_based/collation_task.rs
+++ b/cumulus/client/consensus/aura/src/collators/slot_based/collation_task.rs
@@ -20,7 +20,9 @@ use std::path::PathBuf;
 use cumulus_client_collator::service::ServiceInterface as CollatorServiceInterface;
 use cumulus_relay_chain_interface::RelayChainInterface;
 
-use polkadot_node_primitives::{MaybeCompressedPoV, SubmitCollationParams};
+use polkadot_node_primitives::{
+	MaybeCompressedPoV, SegmentCollation, SubmitCollationParams, SubmitSegmentParams,
+};
 use polkadot_node_subsystem::messages::CollationGenerationMessage;
 use polkadot_overseer::Handle as OverseerHandle;
 use polkadot_primitives::{CollatorPair, Id as ParaId};
@@ -32,7 +34,7 @@ use crate::export_pov_to_path;
 use sc_utils::mpsc::TracingUnboundedReceiver;
 use sp_runtime::traits::{Block as BlockT, Header};
 
-use super::CollatorMessage;
+use super::{CollatorMessage, CollatorResubmitSegment, SegmentKind};
 
 const LOG_TARGET: &str = "aura::cumulus::collation_task";
 
@@ -48,8 +50,10 @@ pub struct Params<Block: BlockT, RClient, CS> {
 	pub reinitialize: bool,
 	/// Collator service interface
 	pub collator_service: CS,
-	/// Receiver channel for communication with the block builder task.
+	/// Receiver channel for V2 single-bundle collations from the block builder task.
 	pub collator_receiver: TracingUnboundedReceiver<CollatorMessage<Block>>,
+	/// Receiver channel for V3 segment collations from the block builder task.
+	pub resubmit_receiver: TracingUnboundedReceiver<CollatorResubmitSegment<Block>>,
 	/// The handle from the special slot based block import.
 	pub block_import_handle: super::SlotBasedBlockImportHandle<Block>,
 	/// When set, the collator will export every produced `POV` to this folder.
@@ -70,6 +74,7 @@ pub async fn run_collation_task<Block, RClient, CS>(
 		reinitialize,
 		collator_service,
 		mut collator_receiver,
+		mut resubmit_receiver,
 		mut block_import_handle,
 		export_pov,
 	}: Params<Block, RClient, CS>,
@@ -98,7 +103,18 @@ pub async fn run_collation_task<Block, RClient, CS>(
 					return;
 				};
 
-				handle_collation_message(message, &collator_service, &mut overseer_handle,relay_client.clone(),export_pov.clone()).await;
+				handle_collation_message(message, &collator_service, &mut overseer_handle, relay_client.clone(), export_pov.clone()).await;
+			},
+			resubmit_message = resubmit_receiver.next() => {
+				let Some(message) = resubmit_message else {
+					return;
+				};
+
+				// Seed the segment with the current unincluded-segment collations. Empty for now;
+				// future resubmission machinery — the layer that computes and keeps the unincluded
+				// segment up to date across parachain slots — will populate this before dispatch.
+				let collations = Vec::new();
+				handle_resubmit_segment(message, collations, &collator_service, &mut overseer_handle, relay_client.clone(), export_pov.clone()).await;
 			},
 			block_import_msg = block_import_handle.next().fuse() => {
 				// TODO: Implement me.
@@ -109,9 +125,8 @@ pub async fn run_collation_task<Block, RClient, CS>(
 	}
 }
 
-/// Handle an incoming collation message from the block builder task.
-/// This builds the collation from the [`CollatorMessage`] and submits it to
-/// the collation-generation subsystem of the relay chain.
+/// Handle an incoming single-bundle V2 [`CollatorMessage`] and forward it to the collation
+/// generation subsystem as a [`CollationGenerationMessage::SubmitCollation`].
 async fn handle_collation_message<Block: BlockT, RClient: RelayChainInterface + Clone + 'static>(
 	message: CollatorMessage<Block>,
 	collator_service: &impl CollatorServiceInterface<Block>,
@@ -120,7 +135,6 @@ async fn handle_collation_message<Block: BlockT, RClient: RelayChainInterface +
 	export_pov: Option<PathBuf>,
 ) {
 	let CollatorMessage {
-		scheduling_proof,
 		parent_header,
 		blocks,
 		proof,
@@ -130,21 +144,14 @@ async fn handle_collation_message<Block: BlockT, RClient: RelayChainInterface +
 		validation_data,
 	} = message;
 
-	// Derive scheduling_parent from the proof (the ISP header's hash is used when the
-	// header chain is empty — that's the case with `relay_parent_offset = 0`).
-	let scheduling_parent = scheduling_proof.as_ref().map(|p| p.scheduling_parent());
-	let (collation, block_data) = match collator_service.build_multi_block_collation(
-		&parent_header,
-		blocks,
-		proof,
-		scheduling_proof,
-	) {
-		Some(collation) => collation,
-		None => {
-			tracing::warn!(target: LOG_TARGET, ?core_index, "Unable to build collation.");
-			return;
-		},
-	};
+	let (collation, block_data) =
+		match collator_service.build_multi_block_collation(&parent_header, blocks, proof, None) {
+			Some(collation) => collation,
+			None => {
+				tracing::warn!(target: LOG_TARGET, ?core_index, "Unable to build collation.");
+				return;
+			},
+		};
 
 	block_data.log_size_info();
 
@@ -206,7 +213,7 @@ async fn handle_collation_message<Block: BlockT, RClient: RelayChainInterface +
 				validation_code_hash,
 				core_index,
 				result_sender: None,
-				scheduling_parent,
+				scheduling_parent: None,
 				session_index,
 				validation_data,
 			}),
@@ -214,3 +221,127 @@ async fn handle_collation_message<Block: BlockT, RClient: RelayChainInterface +
 		)
 		.await;
 }
+
+/// Handle an incoming segment message from the block builder task. Each entry is built into a
+/// [`SegmentCollation`] (one PoV / one candidate on the relay chain); the whole segment is
+/// forwarded to the collation-generation subsystem as a single
+/// [`CollationGenerationMessage::SubmitSegment`]. Entries that fail to build or whose session
+/// lookup fails are skipped — they do not abort the whole segment.
+async fn handle_resubmit_segment<Block: BlockT, RClient: RelayChainInterface + Clone + 'static>(
+	message: CollatorResubmitSegment<Block>,
+	mut collations: Vec<SegmentCollation>,
+	collator_service: &impl CollatorServiceInterface<Block>,
+	overseer_handle: &mut OverseerHandle,
+	relay_client: RClient,
+	export_pov: Option<PathBuf>,
+) {
+	let CollatorResubmitSegment { scheduling_proof, kind } = message;
+	let scheduling_parent = scheduling_proof.scheduling_parent();
+	let core_index = match kind {
+		SegmentKind::WithBundle { bundle } => {
+			let CollatorMessage {
+				relay_parent,
+				parent_header,
+				blocks,
+				proof,
+				validation_code_hash,
+				core_index,
+				validation_data,
+			} = bundle;
+
+			let (collation, block_data) = match collator_service.build_multi_block_collation(
+				&parent_header,
+				blocks,
+				proof,
+				Some(scheduling_proof.clone()),
+			) {
+				Some(collation) => collation,
+				None => {
+					tracing::warn!(target: LOG_TARGET, ?core_index, ?relay_parent, "Unable to build collation for segment entry.");
+					return;
+				},
+			};
+
+			block_data.log_size_info();
+
+			if let MaybeCompressedPoV::Compressed(ref pov) = collation.proof_of_validity {
+				if let Some(pov_path) = export_pov.clone() {
+					if let Ok(Some(relay_parent_header)) =
+						relay_client.header(BlockId::Hash(relay_parent)).await
+					{
+						if let Some(header) = block_data.blocks().first().map(|b| b.header()) {
+							export_pov_to_path::<Block>(
+								pov_path,
+								pov.clone(),
+								header.hash(),
+								*header.number(),
+								parent_header.clone(),
+								relay_parent_header.state_root,
+								relay_parent_header.number,
+								validation_data.max_pov_size,
+							);
+						}
+					} else {
+						tracing::error!(target: LOG_TARGET, "Failed to get relay parent header from hash: {relay_parent:?}");
+					}
+				}
+
+				tracing::info!(
+					target: LOG_TARGET,
+					block_numbers = ?block_data.blocks().iter().map(|b| *b.header().number()).collect::<Vec<_>>(),
+					"Compressed PoV size: {}kb",
+					pov.block_data.0.len() as f64 / 1024f64,
+				);
+			}
+
+			let session_index = match relay_client.session_index_for_child(relay_parent).await {
+				Ok(session_index) => session_index,
+				Err(err) => {
+					tracing::error!(
+						target: LOG_TARGET,
+						?err,
+						?relay_parent,
+						"Failed to fetch session index for segment entry.",
+					);
+					return;
+				},
+			};
+
+			tracing::debug!(
+				target: LOG_TARGET,
+				?core_index,
+				?relay_parent,
+				block_numbers = ?block_data.blocks().iter().map(|b| *b.header().number()).collect::<Vec<_>>(),
+				"Adding entry to segment for core.",
+			);
+
+			collations.push(SegmentCollation {
+				relay_parent,
+				collation,
+				validation_code_hash,
+				result_sender: None,
+				session_index,
+				validation_data,
+			});
+
+			core_index
+		},
+		SegmentKind::ResubmitOnly { core_index } => {
+			if collations.is_empty() {
+				return;
+			}
+			core_index
+		},
+	};
+
+	overseer_handle
+		.send_msg(
+			CollationGenerationMessage::SubmitSegment(SubmitSegmentParams {
+				scheduling_parent,
+				core_index,
+				collations,
+			}),
+			"SubmitSegment",
+		)
+		.await;
+}
diff --git a/cumulus/client/consensus/aura/src/collators/slot_based/mod.rs b/cumulus/client/consensus/aura/src/collators/slot_based/mod.rs
index f309b6751eba4..d5cc88381d152 100644
--- a/cumulus/client/consensus/aura/src/collators/slot_based/mod.rs
+++ b/cumulus/client/consensus/aura/src/collators/slot_based/mod.rs
@@ -219,14 +219,16 @@ pub fn run<Block, P, BI, CIDP, Client, Backend, RClient, CHP, Proposer, CS, Spaw
 	// Initialize proof size recording cleanup
 	register_proof_size_recording_cleanup(para_client.clone());
 
-	let (tx, rx) = tracing_unbounded("mpsc_builder_to_collator", 100);
+	let (collation_tx, collation_rx) = tracing_unbounded("mpsc_builder_to_collator", 100);
+	let (resubmit_tx, resubmit_rx) = tracing_unbounded("mpsc_builder_to_collator_resubmit", 100);
 	let collator_task_params = collation_task::Params {
 		relay_client: relay_client.clone(),
 		collator_key,
 		para_id,
 		reinitialize,
 		collator_service: collator_service.clone(),
-		collator_receiver: rx,
+		collator_receiver: collation_rx,
+		resubmit_receiver: resubmit_rx,
 		block_import_handle,
 		export_pov,
 	};
@@ -245,7 +247,8 @@ pub fn run<Block, P, BI, CIDP, Client, Backend, RClient, CHP, Proposer, CS, Spaw
 		para_id,
 		proposer,
 		collator_service,
-		collator_sender: tx,
+		collator_sender: collation_tx,
+		resubmit_sender: resubmit_tx,
 		relay_chain_slot_duration,
 		slot_offset,
 		max_pov_percentage,
@@ -266,14 +269,11 @@ pub fn run<Block, P, BI, CIDP, Client, Backend, RClient, CHP, Proposer, CS, Spaw
 	);
 }
 
-/// Message to be sent from the block builder to the collation task.
-///
-/// Contains all data necessary to submit a collation to the relay chain.
+/// Single-bundle message sent from the block builder for V2 candidates. Routed to
+/// [`CollationGenerationMessage::SubmitCollation`] by the collation task.
 struct CollatorMessage<Block: BlockT> {
 	/// The hash of the relay chain block that provides the context for the parachain block.
 	pub relay_parent: RelayHash,
-	/// V3 scheduling proof. None for V1/V2 candidates.
-	pub scheduling_proof: Option<SchedulingProof>,
 	/// The header of the parent block.
 	pub parent_header: Block::Header,
 	/// The built blocks.
@@ -282,8 +282,26 @@ struct CollatorMessage<Block: BlockT> {
 	pub proof: StorageProof,
 	/// The validation code hash at the parent block.
 	pub validation_code_hash: ValidationCodeHash,
-	/// Core index that this block should be submitted on
+	/// Core index that this block should be submitted on.
 	pub core_index: CoreIndex,
 	/// The persisted validation data for this collation.
 	pub validation_data: PersistedValidationData,
 }
+
+/// V3/V4 resubmit-segment message sent by the block builder. Routed to
+/// [`CollationGenerationMessage::SubmitSegment`] by the collation task.
+struct CollatorResubmitSegment<Block: BlockT> {
+	/// Scheduling proof shared by every entry in the submitted segment.
+	pub scheduling_proof: SchedulingProof,
+	/// Whether this message carries a freshly-built block or signals resubmit-only.
+	pub kind: SegmentKind<Block>,
+}
+
+/// Kind of resubmit-segment message. The target `CoreIndex` is read from `bundle.core_index` in
+/// the `WithBundle` case and carried explicitly in the `ResubmitOnly` case.
+enum SegmentKind<Block: BlockT> {
+	/// A freshly-built block to submit.
+	WithBundle { bundle: CollatorMessage<Block> },
+	/// Resubmit only — no freshly-built block this slot.
+	ResubmitOnly { core_index: CoreIndex },
+}
diff --git a/cumulus/primitives/core/src/scheduling.rs b/cumulus/primitives/core/src/scheduling.rs
index 74859a6ad8625..80fe0ecbe6305 100644
--- a/cumulus/primitives/core/src/scheduling.rs
+++ b/cumulus/primitives/core/src/scheduling.rs
@@ -10,9 +10,9 @@
 //!
 //! # Resubmission
 //!
-//! When a candidate fails to get backed in time, a different collator can resubmit
-//! it with a new `scheduling_parent` (fresh relay tip) without re-executing the blocks.
-//! The `relay_parent` stays the same since the execution context hasn't changed.
+//! When a candidate fails to get backed in time, a collator can resubmit it with a new
+//! `scheduling_parent` (fresh relay tip) without re-executing the blocks. The `relay_parent`
+//! stays the same since the execution context hasn't changed.
 //!
 //! For resubmission, `signed_scheduling_info` must be provided. The resubmitting
 //! collator signs the core selection, proving they are the eligible author for the
@@ -23,32 +23,36 @@ use codec::{Decode, Encode};
 use polkadot_primitives::{ApprovedPeerId, CoreSelector, Header as RelayChainHeader};
 use sp_runtime::traits::{BlakeTwo256, Hash as HashT};
 
-/// Payload signed by a collator for resubmission.
+/// Payload signed by a collator to authorize core selection — required on resubmission,
+/// optional on initial submission.
 ///
 /// This binds the core selection and reputation-credit peer to a specific internal
 /// scheduling parent, preventing replay attacks across different scheduling contexts.
 #[derive(Clone, Encode, Decode, Debug, PartialEq, Eq)]
 pub struct SchedulingInfoPayload {
-	/// Which core to use (indexes into the parachain's assigned cores).
+	/// Index into the parachain's cores scheduled at depth `claim_queue_offset` of the relay
+	/// chain's claim queue. The relay chain resolves it to a `CoreIndex` at verification time.
 	pub core_selector: CoreSelector,
-	/// The claim queue offset.
+	/// Depth in the relay chain's claim queue (at `scheduling_parent`) used together with
+	/// `core_selector` to resolve the target `CoreIndex`.
 	pub claim_queue_offset: u8,
 	/// Peer ID to receive reputation credit for successful collation delivery.
 	pub peer_id: ApprovedPeerId,
-	/// The internal scheduling parent whom's slot decides the
-	/// eligible block author that must sign the payload.
+	/// The internal scheduling parent whose slot decides the eligible block author that must
+	/// sign the payload.
 	pub internal_scheduling_parent: polkadot_primitives::Hash,
 }
 
-/// Signed scheduling information for candidate resubmission.
+/// Signed scheduling information attached to a [`SchedulingProof`].
 ///
-/// When a collator resubmits a candidate (with a newer `scheduling_parent` but same
-/// `relay_parent`), they must sign the core selection to prove eligibility for the
-/// slot at `internal_scheduling_parent`.
+/// On **resubmission** (when the candidate is submitted once more under a newer
+/// `scheduling_parent` than the one used for its initial submission), this signature is
+/// required: the resubmitting collator signs the core selection to prove eligibility for the
+/// slot at the `internal_scheduling_parent`.
 ///
-/// The `claim_queue_offset` is derived from the runtime's `relay_parent_offset`
-/// configuration and is not part of this struct - it cannot be overridden by the
-/// collator.
+/// On **initial submission**, this is optional. When present, it provides an explicit signed
+/// core selection that overrides the parachain block's UMP signals; when absent, the relay
+/// chain falls back to those UMP signals.
 #[derive(Clone, Encode, Decode, Debug, PartialEq, Eq)]
 pub struct SignedSchedulingInfo {
 	/// The scheduling information.
@@ -75,36 +79,35 @@ impl SchedulingInfoPayload {
 
 /// V3 scheduling proof included in the POV.
 ///
-/// Provides the ancestry from scheduling_parent back to the internal scheduling
-/// parent. The PVF validates this against the relay_parent and scheduling_parent
-/// from the candidate descriptor extension.
+/// Carries the relay-chain ancestry starting at `scheduling_parent` and ending one step
+/// before `internal_scheduling_parent` (whose header is supplied separately). The PVF
+/// validates this against the `relay_parent` and `scheduling_parent` from the candidate
+/// descriptor extension.
 #[derive(Clone, Encode, Decode, Debug, PartialEq, Eq)]
 pub struct SchedulingProof {
-	/// Relay chain headers proving ancestry from scheduling_parent backward.
-	///
 	/// Forms a chain where each header's parent_hash equals the next header's hash.
 	/// The first header's hash must equal the candidate's scheduling_parent.
 	/// The last header's parent_hash is the internal scheduling parent.
-	/// Length is defined by the parachain runtime config (RelayParentOffset).
+	/// Length is defined by the parachain runtime config (`RelayParentOffset`); when that is
+	/// `0` the chain is empty and `scheduling_parent == internal_scheduling_parent`.
 	pub header_chain: Vec<RelayChainHeader>,
-	/// The relay chain header at `internal_scheduling_parent`. Its hash must equal the
-	/// `internal_scheduling_parent` derived from `header_chain` (the parent of the chain's
-	/// last header, or `scheduling_parent` if the chain is empty).
+	/// The relay chain header at `internal_scheduling_parent` (the parent of the chain's last
+	/// header, or `scheduling_parent` if the chain is empty). Carries the BABE pre-digest used
+	/// to look up the eligible block author when verifying `signed_scheduling_info`.
 	pub internal_scheduling_parent_header: RelayChainHeader,
-	/// Signed scheduling info for core selection override.
+	/// Signed scheduling info. Required on resubmission, optional on initial submission.
 	///
 	/// - `None` with `relay_parent == internal_scheduling_parent`: Initial submission. Core
 	///   selection comes from the parachain block's UMP signals.
 	///
 	/// - `Some` with `relay_parent == internal_scheduling_parent`: Initial submission with
-	///   explicit core selection. This is optional but legal. Collators should refuse to
-	///   acknowledge blocks with invalid scheduling info, so providing a signature is not required
-	///   for initial submissions.
+	///   explicit signed core selection. Legal but not required.
 	///
 	/// - `Some` with `relay_parent != internal_scheduling_parent`: Resubmission (required). The
 	///   resubmitting collator signs the core selection, overriding the block's UMP signals.
-	///   Signature is verified against the eligible author for the slot at
-	///   `internal_scheduling_parent`.
+	///
+	/// - `None` with `relay_parent != internal_scheduling_parent`: invalid; rejected by the
+	///   verifier.
 	pub signed_scheduling_info: Option<SignedSchedulingInfo>,
 }
 
@@ -131,7 +134,9 @@ pub trait VerifySchedulingSignature {
 	/// Whether V3 scheduling validation is enabled.
 	const V3_SCHEDULING_ENABLED: bool;
 
-	/// Verifies `signed_info` against `internal_scheduling_parent_header`.
+	/// Verifies `signed_info.signature` over the encoded payload against the public key of the
+	/// eligible block author for the parachain slot derived from
+	/// `internal_scheduling_parent_header`'s BABE pre-digest.
 	fn verify(
 		signed_info: &SignedSchedulingInfo,
 		internal_scheduling_parent_header: &RelayChainHeader,
diff --git a/polkadot/node/collation-generation/src/lib.rs b/polkadot/node/collation-generation/src/lib.rs
index b523babed1848..dfffa2ba12613 100644
--- a/polkadot/node/collation-generation/src/lib.rs
+++ b/polkadot/node/collation-generation/src/lib.rs
@@ -90,7 +90,7 @@ use error::{Error, Result};
 use futures::{channel::oneshot, future::FutureExt, select};
 use polkadot_node_primitives::{
 	AvailableData, Collation, CollationGenerationConfig, CollationSecondedSignal, PoV,
-	SubmitCollationParams,
+	SegmentCollation, SubmitCollationParams, SubmitSegmentParams,
 };
 use polkadot_node_subsystem::{
 	messages::{CollationGenerationMessage, CollatorProtocolMessage, RuntimeApiMessage},
@@ -203,6 +203,15 @@ impl CollationGenerationSubsystem {
 
 				false
 			},
+			Ok(FromOrchestra::Communication {
+				msg: CollationGenerationMessage::SubmitSegment(params),
+			}) => {
+				if let Err(err) = self.handle_submit_segment(params, ctx).await {
+					gum::error!(target: LOG_TARGET, ?err, "Failed to submit segment");
+				}
+
+				false
+			},
 			Ok(FromOrchestra::Signal(OverseerSignal::BlockFinalized(..))) => false,
 			Err(err) => {
 				gum::error!(
@@ -276,6 +285,54 @@ impl CollationGenerationSubsystem {
 		Ok(())
 	}
 
+	// TODO: once the V4 collator-protocol lands, distribute the receipts for the whole segment as
+	// a single unit instead of one collation at a time, and drop the length-1 restriction.
+	async fn handle_submit_segment<Context>(
+		&mut self,
+		params: SubmitSegmentParams,
+		ctx: &mut Context,
+	) -> Result<()> {
+		let SubmitSegmentParams { scheduling_parent, core_index, mut collations } = params;
+
+		if collations.is_empty() {
+			gum::warn!(target: LOG_TARGET, "received empty segment submission");
+			return Ok(());
+		}
+
+		if collations.len() > 1 {
+			gum::warn!(
+				target: LOG_TARGET,
+				len = collations.len(),
+				"multi-collation segment submission not yet supported; ignoring",
+			);
+			return Ok(());
+		}
+
+		let SegmentCollation {
+			relay_parent,
+			collation,
+			validation_code_hash,
+			result_sender,
+			session_index,
+			validation_data,
+		} = collations.pop().expect("len == 1; qed");
+
+		self.handle_submit_collation(
+			SubmitCollationParams {
+				relay_parent,
+				collation,
+				validation_code_hash,
+				result_sender,
+				core_index,
+				scheduling_parent: Some(scheduling_parent),
+				session_index,
+				validation_data,
+			},
+			ctx,
+		)
+		.await
+	}
+
 	async fn handle_new_activation<Context>(
 		&mut self,
 		maybe_activated: Option<Hash>,
diff --git a/polkadot/node/primitives/src/lib.rs b/polkadot/node/primitives/src/lib.rs
index 075b9013021f3..69d3cdf85dd78 100644
--- a/polkadot/node/primitives/src/lib.rs
+++ b/polkadot/node/primitives/src/lib.rs
@@ -553,6 +553,50 @@ pub struct SubmitCollationParams {
 	pub validation_data: PersistedValidationData,
 }
 
+/// A single collation in a segment submitted via `CollationGenerationMessage::SubmitSegment`.
+#[derive(Debug)]
+pub struct SegmentCollation {
+	/// The relay-parent the collation is built against.
+	pub relay_parent: Hash,
+	/// The collation itself (PoV and commitments).
+	pub collation: Collation,
+	/// The hash of the validation code the collation was created against.
+	pub validation_code_hash: ValidationCodeHash,
+	/// An optional result sender that should be informed about a successfully seconded collation.
+	///
+	/// See [`SubmitCollationParams::result_sender`] for the same caveats.
+	pub result_sender: Option<futures::channel::oneshot::Sender<CollationSecondedSignal>>,
+	/// The session index of the relay parent. Goes into the candidate descriptor.
+	/// Must be provided by the caller because the relay parent's state may be pruned.
+	pub session_index: SessionIndex,
+	/// The persisted validation data for this collation. The `parent_head` field must be set
+	/// to the correct parent head-data for the parablock being submitted.
+	pub validation_data: PersistedValidationData,
+}
+
+/// Parameters for `CollationGenerationMessage::SubmitSegment`.
+///
+/// Submits a segment of collations that share a common scheduling parent and target core. Each
+/// [`SegmentCollation`] in `collations` carries the fields that may differ between blocks of the
+/// segment (relay parent, collation payload, validation data, etc.).
+///
+/// **Currently restricted to `collations.len() == 1`.** The collation-generation subsystem drops
+/// submissions with more than one entry (the V4 collator-protocol work will lift this and
+/// distribute the whole segment as a single unit).
+#[derive(Debug)]
+pub struct SubmitSegmentParams {
+	/// The scheduling parent shared by every collation in the segment. Segments are V3/V4-only,
+	/// so this is always present and produces V3 candidate descriptors.
+	///
+	/// WARNING: only valid when the `CandidateReceiptV3` node feature is set.
+	pub scheduling_parent: Hash,
+	/// The core index on which the resulting candidates should be backed.
+	pub core_index: CoreIndex,
+	/// The collations in this segment, in the order they should be submitted. Must currently
+	/// contain exactly one entry; see the type-level note above.
+	pub collations: Vec<SegmentCollation>,
+}
+
 /// This is the data we keep available for each candidate included in the relay chain.
 #[derive(Clone, Encode, Decode, PartialEq, Eq, Debug)]
 pub struct AvailableData {
diff --git a/polkadot/node/subsystem-types/src/messages.rs b/polkadot/node/subsystem-types/src/messages.rs
index 7117f16999dc3..74821703b0799 100644
--- a/polkadot/node/subsystem-types/src/messages.rs
+++ b/polkadot/node/subsystem-types/src/messages.rs
@@ -39,7 +39,7 @@ use polkadot_node_primitives::{
 	AvailableData, BabeEpoch, BlockWeight, CandidateVotes, CollationGenerationConfig,
 	CollationSecondedSignal, DisputeMessage, DisputeStatus, ErasureChunk, PoV,
 	SignedDisputeStatement, SignedFullStatement, SignedFullStatementWithPVD, SubmitCollationParams,
-	ValidationResult,
+	SubmitSegmentParams, ValidationResult,
 };
 use polkadot_primitives::{
 	self,
@@ -989,6 +989,12 @@ pub enum CollationGenerationMessage {
 	///
 	/// If sent before `Initialize`, this will be ignored.
 	SubmitCollation(SubmitCollationParams),
+	/// Submit a segment of collations that share a scheduling parent and target core. Each
+	/// collation is packaged into a signed [`CommittedCandidateReceipt`] and distributed to
+	/// validators, in the order they appear in [`SubmitSegmentParams::collations`].
+	///
+	/// If sent before `Initialize`, this will be ignored.
+	SubmitSegment(SubmitSegmentParams),
 }
 
 /// The result type of [`ApprovalVotingMessage::ImportAssignment`] request.