diff --git a/crates/protocol/src/batch/bits.rs b/crates/protocol/src/batch/bits.rs
index a82b937f0..9fa938c36 100644
--- a/crates/protocol/src/batch/bits.rs
+++ b/crates/protocol/src/batch/bits.rs
@@ -16,6 +16,11 @@ impl AsRef<[u8]> for SpanBatchBits {
 }
 
 impl SpanBatchBits {
+    /// Creates a new span batch bits.
+    pub const fn new(inner: Vec<u8>) -> Self {
+        Self(inner)
+    }
+
     /// Decodes a standard span-batch bitlist from a reader.
     /// The bitlist is encoded as big-endian integer, left-padded with zeroes to a multiple of 8
     /// bits. The encoded bitlist cannot be longer than `bit_length`.
diff --git a/crates/protocol/src/batch/core.rs b/crates/protocol/src/batch/core.rs
new file mode 100644
index 000000000..8637abece
--- /dev/null
+++ b/crates/protocol/src/batch/core.rs
@@ -0,0 +1,51 @@
+//! Module containing the core [Batch] enum.
+
+use crate::{BatchDecodingError, BatchType, RawSpanBatch, SingleBatch, SpanBatch};
+use alloy_rlp::{Buf, Decodable};
+use op_alloy_genesis::RollupConfig;
+
+/// A Batch.
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[allow(clippy::large_enum_variant)]
+pub enum Batch {
+    /// A single batch
+    Single(SingleBatch),
+    /// Span Batches
+    Span(SpanBatch),
+}
+
+impl Batch {
+    /// Returns the timestamp for the batch.
+    pub fn timestamp(&self) -> u64 {
+        match self {
+            Self::Single(sb) => sb.timestamp,
+            Self::Span(sb) => sb.starting_timestamp(),
+        }
+    }
+
+    /// Attempts to decode a batch from a reader.
+    pub fn decode(r: &mut &[u8], cfg: &RollupConfig) -> Result<Self, BatchDecodingError> {
+        if r.is_empty() {
+            return Err(BatchDecodingError::EmptyBuffer);
+        }
+
+        // Read the batch type
+        let batch_type = BatchType::from(r[0]);
+        r.advance(1);
+
+        match batch_type {
+            BatchType::Single => {
+                let single_batch =
+                    SingleBatch::decode(r).map_err(BatchDecodingError::AlloyRlpError)?;
+                Ok(Self::Single(single_batch))
+            }
+            BatchType::Span => {
+                let mut raw_span_batch = RawSpanBatch::decode(r)?;
+                let span_batch = raw_span_batch
+                    .derive(cfg.block_time, cfg.genesis.l2_time, cfg.l2_chain_id)
+                    .map_err(BatchDecodingError::SpanBatchError)?;
+                Ok(Self::Span(span_batch))
+            }
+        }
+    }
+}
diff --git a/crates/protocol/src/batch/errors.rs b/crates/protocol/src/batch/errors.rs
index 8422f4aba..8ab4b5eec 100644
--- a/crates/protocol/src/batch/errors.rs
+++ b/crates/protocol/src/batch/errors.rs
@@ -35,6 +35,41 @@ impl core::error::Error for SpanBatchError {
     }
 }
 
+/// An error decoding a batch.
+#[derive(Debug, derive_more::Display, Clone, PartialEq, Eq)]
+pub enum BatchDecodingError {
+    /// Empty buffer
+    #[display("Empty buffer")]
+    EmptyBuffer,
+    /// Error decoding an Alloy RLP
+    #[display("Error decoding an Alloy RLP: {_0}")]
+    AlloyRlpError(alloy_rlp::Error),
+    /// Error decoding a span batch
+    #[display("Error decoding a span batch: {_0}")]
+    SpanBatchError(SpanBatchError),
+}
+
+impl From<alloy_rlp::Error> for BatchDecodingError {
+    fn from(err: alloy_rlp::Error) -> Self {
+        Self::AlloyRlpError(err)
+    }
+}
+
+impl From<SpanBatchError> for BatchDecodingError {
+    fn from(err: SpanBatchError) -> Self {
+        Self::SpanBatchError(err)
+    }
+}
+
+impl core::error::Error for BatchDecodingError {
+    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
+        match self {
+            Self::SpanBatchError(err) => Some(err),
+            _ => None,
+        }
+    }
+}
+
 /// Decoding Error
 #[derive(Debug, derive_more::Display, Clone, PartialEq, Eq)]
 pub enum SpanDecodingError {
diff --git a/crates/protocol/src/batch/inclusion.rs b/crates/protocol/src/batch/inclusion.rs
new file mode 100644
index 000000000..1a630f91b
--- /dev/null
+++ b/crates/protocol/src/batch/inclusion.rs
@@ -0,0 +1,44 @@
+//! Module containing the [BatchWithInclusionBlock] struct.
+
+use crate::{Batch, BatchValidationProvider, BatchValidity, BlockInfo, L2BlockInfo};
+use op_alloy_genesis::RollupConfig;
+
+/// A batch with its inclusion block.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct BatchWithInclusionBlock {
+    /// The inclusion block
+    pub inclusion_block: BlockInfo,
+    /// The batch
+    pub batch: Batch,
+}
+
+impl BatchWithInclusionBlock {
+    /// Creates a new batch with inclusion block.
+    pub const fn new(inclusion_block: BlockInfo, batch: Batch) -> Self {
+        Self { inclusion_block, batch }
+    }
+
+    /// Validates the batch can be applied on top of the specified L2 safe head.
+    /// The first entry of the l1_blocks should match the origin of the l2_safe_head.
+    /// One or more consecutive l1_blocks should be provided.
+    /// In case of only a single L1 block, the decision whether a batch is valid may have to stay
+    /// undecided.
+    pub async fn check_batch<BF: BatchValidationProvider>(
+        &self,
+        cfg: &RollupConfig,
+        l1_blocks: &[BlockInfo],
+        l2_safe_head: L2BlockInfo,
+        fetcher: &mut BF,
+    ) -> BatchValidity {
+        match &self.batch {
+            Batch::Single(single_batch) => {
+                single_batch.check_batch(cfg, l1_blocks, l2_safe_head, &self.inclusion_block)
+            }
+            Batch::Span(span_batch) => {
+                span_batch
+                    .check_batch(cfg, l1_blocks, l2_safe_head, &self.inclusion_block, fetcher)
+                    .await
+            }
+        }
+    }
+}
diff --git a/crates/protocol/src/batch/mod.rs b/crates/protocol/src/batch/mod.rs
index 9fecc001b..a350f2e0c 100644
--- a/crates/protocol/src/batch/mod.rs
+++ b/crates/protocol/src/batch/mod.rs
@@ -3,8 +3,23 @@
 mod r#type;
 pub use r#type::*;
 
+mod core;
+pub use core::Batch;
+
+mod raw;
+pub use raw::RawSpanBatch;
+
+mod payload;
+pub use payload::SpanBatchPayload;
+
+mod prefix;
+pub use prefix::SpanBatchPrefix;
+
+mod inclusion;
+pub use inclusion::BatchWithInclusionBlock;
+
 mod errors;
-pub use errors::{SpanBatchError, SpanDecodingError};
+pub use errors::{BatchDecodingError, SpanBatchError, SpanDecodingError};
 
 mod bits;
 pub use bits::SpanBatchBits;
diff --git a/crates/protocol/src/batch/payload.rs b/crates/protocol/src/batch/payload.rs
new file mode 100644
index 000000000..6b713a54e
--- /dev/null
+++ b/crates/protocol/src/batch/payload.rs
@@ -0,0 +1,196 @@
+//! Raw Span Batch Payload
+
+use super::MAX_SPAN_BATCH_ELEMENTS;
+use crate::{SpanBatchBits, SpanBatchError, SpanBatchTransactions, SpanDecodingError};
+use alloc::vec::Vec;
+
+/// Span Batch Payload
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct SpanBatchPayload {
+    /// Number of L2 block in the span
+    pub block_count: u64,
+    /// Standard span-batch bitlist of blockCount bits. Each bit indicates if the L1 origin is
+    /// changed at the L2 block.
+    pub origin_bits: SpanBatchBits,
+    /// List of transaction counts for each L2 block
+    pub block_tx_counts: Vec<u64>,
+    /// Transactions encoded in SpanBatch specs
+    pub txs: SpanBatchTransactions,
+}
+
+impl SpanBatchPayload {
+    /// Decodes a [SpanBatchPayload] from a reader.
+    pub fn decode_payload(r: &mut &[u8]) -> Result<Self, SpanBatchError> {
+        let mut payload = Self::default();
+        payload.decode_block_count(r)?;
+        payload.decode_origin_bits(r)?;
+        payload.decode_block_tx_counts(r)?;
+        payload.decode_txs(r)?;
+        Ok(payload)
+    }
+
+    /// Encodes a [SpanBatchPayload] into a writer.
+    pub fn encode_payload(&self, w: &mut Vec<u8>) -> Result<(), SpanBatchError> {
+        self.encode_block_count(w);
+        self.encode_origin_bits(w)?;
+        self.encode_block_tx_counts(w);
+        self.encode_txs(w)
+    }
+
+    /// Decodes the origin bits from a reader.
+    pub fn decode_origin_bits(&mut self, r: &mut &[u8]) -> Result<(), SpanBatchError> {
+        if self.block_count > MAX_SPAN_BATCH_ELEMENTS {
+            return Err(SpanBatchError::TooBigSpanBatchSize);
+        }
+
+        self.origin_bits = SpanBatchBits::decode(r, self.block_count as usize)?;
+        Ok(())
+    }
+
+    /// Decode a block count from a reader.
+    pub fn decode_block_count(&mut self, r: &mut &[u8]) -> Result<(), SpanBatchError> {
+        let (block_count, remaining) = unsigned_varint::decode::u64(r)
+            .map_err(|_| SpanBatchError::Decoding(SpanDecodingError::BlockCount))?;
+        // The number of transactions in a single L2 block cannot be greater than
+        // [MAX_SPAN_BATCH_ELEMENTS].
+        if block_count > MAX_SPAN_BATCH_ELEMENTS {
+            return Err(SpanBatchError::TooBigSpanBatchSize);
+        }
+        if block_count == 0 {
+            return Err(SpanBatchError::EmptySpanBatch);
+        }
+        self.block_count = block_count;
+        *r = remaining;
+        Ok(())
+    }
+
+    /// Decode block transaction counts from a reader.
+    pub fn decode_block_tx_counts(&mut self, r: &mut &[u8]) -> Result<(), SpanBatchError> {
+        // Initially allocate the vec with the block count, to reduce re-allocations in the first
+        // few blocks.
+        let mut block_tx_counts = Vec::with_capacity(self.block_count as usize);
+
+        for _ in 0..self.block_count {
+            let (block_tx_count, remaining) = unsigned_varint::decode::u64(r)
+                .map_err(|_| SpanBatchError::Decoding(SpanDecodingError::BlockTxCounts))?;
+
+            // The number of transactions in a single L2 block cannot be greater than
+            // [MAX_SPAN_BATCH_ELEMENTS].
+            if block_tx_count > MAX_SPAN_BATCH_ELEMENTS {
+                return Err(SpanBatchError::TooBigSpanBatchSize);
+            }
+            block_tx_counts.push(block_tx_count);
+            *r = remaining;
+        }
+        self.block_tx_counts = block_tx_counts;
+        Ok(())
+    }
+
+    /// Decode transactions from a reader.
+    pub fn decode_txs(&mut self, r: &mut &[u8]) -> Result<(), SpanBatchError> {
+        if self.block_tx_counts.is_empty() {
+            return Err(SpanBatchError::EmptySpanBatch);
+        }
+
+        let total_block_tx_count =
+            self.block_tx_counts.iter().try_fold(0u64, |acc, block_tx_count| {
+                acc.checked_add(*block_tx_count).ok_or(SpanBatchError::TooBigSpanBatchSize)
+            })?;
+
+        // The total number of transactions in a span batch cannot be greater than
+        // [MAX_SPAN_BATCH_ELEMENTS].
+        if total_block_tx_count > MAX_SPAN_BATCH_ELEMENTS {
+            return Err(SpanBatchError::TooBigSpanBatchSize);
+        }
+        self.txs.total_block_tx_count = total_block_tx_count;
+        self.txs.decode(r)?;
+        Ok(())
+    }
+
+    /// Encode the origin bits into a writer.
+    pub fn encode_origin_bits(&self, w: &mut Vec<u8>) -> Result<(), SpanBatchError> {
+        SpanBatchBits::encode(w, self.block_count as usize, &self.origin_bits)
+    }
+
+    /// Encode the block count into a writer.
+    pub fn encode_block_count(&self, w: &mut Vec<u8>) {
+        let mut u64_varint_buf = [0u8; 10];
+        w.extend_from_slice(unsigned_varint::encode::u64(self.block_count, &mut u64_varint_buf));
+    }
+
+    /// Encode the block transaction counts into a writer.
+    pub fn encode_block_tx_counts(&self, w: &mut Vec<u8>) {
+        let mut u64_varint_buf = [0u8; 10];
+        for block_tx_count in &self.block_tx_counts {
+            u64_varint_buf.fill(0);
+            w.extend_from_slice(unsigned_varint::encode::u64(*block_tx_count, &mut u64_varint_buf));
+        }
+    }
+
+    /// Encode the transactions into a writer.
+    pub fn encode_txs(&self, w: &mut Vec<u8>) -> Result<(), SpanBatchError> {
+        self.txs.encode(w)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alloc::vec;
+
+    #[test]
+    fn test_decode_origin_bits() {
+        let block_count = 10;
+        let encoded = vec![2; block_count / 8 + 1];
+        let mut payload =
+            SpanBatchPayload { block_count: block_count as u64, ..Default::default() };
+        payload.decode_origin_bits(&mut encoded.as_slice()).unwrap();
+        assert_eq!(payload.origin_bits, SpanBatchBits::new(vec![2; block_count / 8 + 1]));
+    }
+
+    #[test]
+    fn test_zero_block_count() {
+        let mut u64_varint_buf = [0; 10];
+        let mut encoded = unsigned_varint::encode::u64(0, &mut u64_varint_buf);
+        let mut payload = SpanBatchPayload::default();
+        let err = payload.decode_block_count(&mut encoded).unwrap_err();
+        assert_eq!(err, SpanBatchError::EmptySpanBatch);
+    }
+
+    #[test]
+    fn test_decode_block_count() {
+        let block_count = MAX_SPAN_BATCH_ELEMENTS;
+        let mut u64_varint_buf = [0; 10];
+        let mut encoded = unsigned_varint::encode::u64(block_count, &mut u64_varint_buf);
+        let mut payload = SpanBatchPayload::default();
+        payload.decode_block_count(&mut encoded).unwrap();
+        assert_eq!(payload.block_count, block_count);
+    }
+
+    #[test]
+    fn test_decode_block_count_errors() {
+        let block_count = MAX_SPAN_BATCH_ELEMENTS + 1;
+        let mut u64_varint_buf = [0; 10];
+        let mut encoded = unsigned_varint::encode::u64(block_count, &mut u64_varint_buf);
+        let mut payload = SpanBatchPayload::default();
+        let err = payload.decode_block_count(&mut encoded).unwrap_err();
+        assert_eq!(err, SpanBatchError::TooBigSpanBatchSize);
+    }
+
+    #[test]
+    fn test_decode_block_tx_counts() {
+        let block_count = 2;
+        let mut u64_varint_buf = [0; 10];
+        let mut encoded = unsigned_varint::encode::u64(block_count, &mut u64_varint_buf);
+        let mut payload = SpanBatchPayload::default();
+        payload.decode_block_count(&mut encoded).unwrap();
+        let mut r: Vec<u8> = Vec::new();
+        for _ in 0..2 {
+            let mut buf = [0u8; 10];
+            let encoded = unsigned_varint::encode::u64(2, &mut buf);
+            r.append(&mut encoded.to_vec());
+        }
+        payload.decode_block_tx_counts(&mut r.as_slice()).unwrap();
+        assert_eq!(payload.block_tx_counts, vec![2, 2]);
+    }
+}
diff --git a/crates/protocol/src/batch/prefix.rs b/crates/protocol/src/batch/prefix.rs
new file mode 100644
index 000000000..e3e5fc32b
--- /dev/null
+++ b/crates/protocol/src/batch/prefix.rs
@@ -0,0 +1,97 @@
+//! Raw Span Batch Prefix
+
+use crate::{SpanBatchError, SpanDecodingError};
+use alloc::vec::Vec;
+use alloy_primitives::FixedBytes;
+
+/// Span Batch Prefix
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct SpanBatchPrefix {
+    /// Relative timestamp of the first block
+    pub rel_timestamp: u64,
+    /// L1 origin number
+    pub l1_origin_num: u64,
+    /// First 20 bytes of the first block's parent hash
+    pub parent_check: FixedBytes<20>,
+    /// First 20 bytes of the last block's L1 origin hash
+    pub l1_origin_check: FixedBytes<20>,
+}
+
+impl SpanBatchPrefix {
+    /// Decodes a [SpanBatchPrefix] from a reader.
+    pub fn decode_prefix(r: &mut &[u8]) -> Result<Self, SpanBatchError> {
+        let mut prefix = Self::default();
+        prefix.decode_rel_timestamp(r)?;
+        prefix.decode_l1_origin_num(r)?;
+        prefix.decode_parent_check(r)?;
+        prefix.decode_l1_origin_check(r)?;
+        Ok(prefix)
+    }
+
+    /// Decodes the relative timestamp from a reader.
+    pub fn decode_rel_timestamp(&mut self, r: &mut &[u8]) -> Result<(), SpanBatchError> {
+        let (rel_timestamp, remaining) = unsigned_varint::decode::u64(r)
+            .map_err(|_| SpanBatchError::Decoding(SpanDecodingError::RelativeTimestamp))?;
+        *r = remaining;
+        self.rel_timestamp = rel_timestamp;
+        Ok(())
+    }
+
+    /// Decodes the L1 origin number from a reader.
+    pub fn decode_l1_origin_num(&mut self, r: &mut &[u8]) -> Result<(), SpanBatchError> {
+        let (l1_origin_num, remaining) = unsigned_varint::decode::u64(r)
+            .map_err(|_| SpanBatchError::Decoding(SpanDecodingError::L1OriginNumber))?;
+        *r = remaining;
+        self.l1_origin_num = l1_origin_num;
+        Ok(())
+    }
+
+    /// Decodes the parent check from a reader.
+    pub fn decode_parent_check(&mut self, r: &mut &[u8]) -> Result<(), SpanBatchError> {
+        let (parent_check, remaining) = r.split_at(20);
+        let parent_check = FixedBytes::<20>::from_slice(parent_check);
+        *r = remaining;
+        self.parent_check = parent_check;
+        Ok(())
+    }
+
+    /// Decodes the L1 origin check from a reader.
+    pub fn decode_l1_origin_check(&mut self, r: &mut &[u8]) -> Result<(), SpanBatchError> {
+        let (l1_origin_check, remaining) = r.split_at(20);
+        let l1_origin_check = FixedBytes::<20>::from_slice(l1_origin_check);
+        *r = remaining;
+        self.l1_origin_check = l1_origin_check;
+        Ok(())
+    }
+
+    /// Encodes the [SpanBatchPrefix] into a writer.
+    pub fn encode_prefix(&self, w: &mut Vec<u8>) {
+        let mut u64_buf = [0u8; 10];
+        w.extend_from_slice(unsigned_varint::encode::u64(self.rel_timestamp, &mut u64_buf));
+        w.extend_from_slice(unsigned_varint::encode::u64(self.l1_origin_num, &mut u64_buf));
+        w.extend_from_slice(self.parent_check.as_slice());
+        w.extend_from_slice(self.l1_origin_check.as_slice());
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use alloc::vec::Vec;
+    use alloy_primitives::address;
+
+    #[test]
+    fn test_span_batch_prefix_encoding_roundtrip() {
+        let expected = SpanBatchPrefix {
+            rel_timestamp: 0xFF,
+            l1_origin_num: 0xEE,
+            parent_check: address!("beef00000000000000000000000000000000beef").into(),
+            l1_origin_check: address!("babe00000000000000000000000000000000babe").into(),
+        };
+
+        let mut buf = Vec::new();
+        expected.encode_prefix(&mut buf);
+
+        assert_eq!(SpanBatchPrefix::decode_prefix(&mut buf.as_slice()).unwrap(), expected);
+    }
+}
diff --git a/crates/protocol/src/batch/raw.rs b/crates/protocol/src/batch/raw.rs
new file mode 100644
index 000000000..59d990f3c
--- /dev/null
+++ b/crates/protocol/src/batch/raw.rs
@@ -0,0 +1,175 @@
+//! Module containing the [RawSpanBatch] struct.
+
+use alloc::{vec, vec::Vec};
+
+use crate::{
+    BatchType, SpanBatch, SpanBatchElement, SpanBatchError, SpanBatchPayload, SpanBatchPrefix,
+    SpanDecodingError,
+};
+
+/// Raw Span Batch
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RawSpanBatch {
+    /// The span batch prefix
+    pub prefix: SpanBatchPrefix,
+    /// The span batch payload
+    pub payload: SpanBatchPayload,
+}
+
+impl TryFrom<SpanBatch> for RawSpanBatch {
+    type Error = SpanBatchError;
+
+    fn try_from(value: SpanBatch) -> Result<Self, Self::Error> {
+        if value.batches.is_empty() {
+            return Err(SpanBatchError::EmptySpanBatch);
+        }
+
+        // These should never error since we check for an empty batch above.
+        let span_start = value.batches.first().ok_or(SpanBatchError::EmptySpanBatch)?;
+        let span_end = value.batches.last().ok_or(SpanBatchError::EmptySpanBatch)?;
+
+        Ok(Self {
+            prefix: SpanBatchPrefix {
+                rel_timestamp: span_start.timestamp - value.genesis_timestamp,
+                l1_origin_num: span_end.epoch_num,
+                parent_check: value.parent_check,
+                l1_origin_check: value.l1_origin_check,
+            },
+            payload: SpanBatchPayload {
+                block_count: value.batches.len() as u64,
+                origin_bits: value.origin_bits.clone(),
+                block_tx_counts: value.block_tx_counts.clone(),
+                txs: value.txs.clone(),
+            },
+        })
+    }
+}
+
+impl RawSpanBatch {
+    /// Returns the batch type
+    pub const fn get_batch_type(&self) -> BatchType {
+        BatchType::Span
+    }
+
+    /// Encodes the [RawSpanBatch] into a writer.
+    pub fn encode(&self, w: &mut Vec<u8>) -> Result<(), SpanBatchError> {
+        self.prefix.encode_prefix(w);
+        self.payload.encode_payload(w)
+    }
+
+    /// Decodes the [RawSpanBatch] from a reader.]
+    pub fn decode(r: &mut &[u8]) -> Result<Self, SpanBatchError> {
+        let prefix = SpanBatchPrefix::decode_prefix(r)?;
+        let payload = SpanBatchPayload::decode_payload(r)?;
+        Ok(Self { prefix, payload })
+    }
+
+    /// Converts a [RawSpanBatch] into a [SpanBatch], which has a list of [SpanBatchElement]s. Thos
+    /// function does not populate the [SpanBatch] with chain configuration data, which is
+    /// required for making payload attributes.
+    pub fn derive(
+        &mut self,
+        block_time: u64,
+        genesis_time: u64,
+        chain_id: u64,
+    ) -> Result<SpanBatch, SpanBatchError> {
+        if self.payload.block_count == 0 {
+            return Err(SpanBatchError::EmptySpanBatch);
+        }
+
+        let mut block_origin_nums = vec![0u64; self.payload.block_count as usize];
+        let mut l1_origin_number = self.prefix.l1_origin_num;
+        for i in (0..self.payload.block_count).rev() {
+            block_origin_nums[i as usize] = l1_origin_number;
+            if self
+                .payload
+                .origin_bits
+                .get_bit(i as usize)
+                .ok_or(SpanBatchError::Decoding(SpanDecodingError::L1OriginCheck))?
+                == 1
+                && i > 0
+            {
+                l1_origin_number -= 1;
+            }
+        }
+
+        // Recover `v` values in transaction signatures within the batch.
+        self.payload.txs.recover_v(chain_id)?;
+
+        // Get all transactions in the batch.
+        let enveloped_txs = self.payload.txs.full_txs(chain_id)?;
+
+        let mut tx_idx = 0;
+        let batches = (0..self.payload.block_count).fold(Vec::new(), |mut acc, i| {
+            let transactions =
+                (0..self.payload.block_tx_counts[i as usize]).fold(Vec::new(), |mut acc, _| {
+                    acc.push(enveloped_txs[tx_idx].clone());
+                    tx_idx += 1;
+                    acc
+                });
+            acc.push(SpanBatchElement {
+                epoch_num: block_origin_nums[i as usize],
+                timestamp: genesis_time + self.prefix.rel_timestamp + block_time * i,
+                transactions: transactions.into_iter().map(|v| v.into()).collect(),
+            });
+            acc
+        });
+
+        Ok(SpanBatch {
+            parent_check: self.prefix.parent_check,
+            l1_origin_check: self.prefix.l1_origin_check,
+            batches,
+            ..Default::default()
+        })
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use alloy_primitives::FixedBytes;
+
+    #[test]
+    fn test_try_from_span_batch_empty_batches_errors() {
+        let span_batch = SpanBatch::default();
+        let raw_span_batch = RawSpanBatch::try_from(span_batch).unwrap_err();
+        assert_eq!(raw_span_batch, SpanBatchError::EmptySpanBatch);
+    }
+
+    #[test]
+    fn test_try_from_span_batch_succeeds() {
+        let parent_check = FixedBytes::from([2u8; 20]);
+        let l1_origin_check = FixedBytes::from([3u8; 20]);
+        let first = SpanBatchElement { epoch_num: 100, timestamp: 400, transactions: Vec::new() };
+        let last = SpanBatchElement { epoch_num: 200, timestamp: 500, transactions: Vec::new() };
+        let span_batch = SpanBatch {
+            batches: vec![first, last],
+            genesis_timestamp: 300,
+            parent_check,
+            l1_origin_check,
+            ..Default::default()
+        };
+        let expected_prefix = SpanBatchPrefix {
+            rel_timestamp: 100,
+            l1_origin_num: 200,
+            parent_check,
+            l1_origin_check,
+        };
+        let expected_payload = SpanBatchPayload { block_count: 2, ..Default::default() };
+        let raw_span_batch = RawSpanBatch::try_from(span_batch).unwrap();
+        assert_eq!(raw_span_batch.prefix, expected_prefix);
+        assert_eq!(raw_span_batch.payload, expected_payload);
+    }
+
+    #[test]
+    fn test_decode_encode_raw_span_batch() {
+        // Load in the raw span batch from the `op-node` derivation pipeline implementation.
+        let raw_span_batch_hex = include_bytes!("./testdata/raw_batch.hex");
+        let mut raw_span_batch = RawSpanBatch::decode(&mut raw_span_batch_hex.as_slice()).unwrap();
+        raw_span_batch.payload.txs.recover_v(981).unwrap();
+
+        let mut encoding_buf = Vec::new();
+        raw_span_batch.encode(&mut encoding_buf).unwrap();
+        assert_eq!(encoding_buf, raw_span_batch_hex);
+    }
+}
diff --git a/crates/protocol/src/batch/testdata/raw_batch.hex b/crates/protocol/src/batch/testdata/raw_batch.hex
new file mode 100644
index 000000000..311a7b999
Binary files /dev/null and b/crates/protocol/src/batch/testdata/raw_batch.hex differ
diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs
index b6bc3a8e4..f9f46a226 100644
--- a/crates/protocol/src/lib.rs
+++ b/crates/protocol/src/lib.rs
@@ -11,11 +11,12 @@ extern crate alloc;
 
 mod batch;
 pub use batch::{
-    BatchType, BatchValidationProvider, BatchValidity, SingleBatch, SpanBatch, SpanBatchBits,
+    Batch, BatchDecodingError, BatchType, BatchValidationProvider, BatchValidity,
+    BatchWithInclusionBlock, RawSpanBatch, SingleBatch, SpanBatch, SpanBatchBits,
     SpanBatchEip1559TransactionData, SpanBatchEip2930TransactionData, SpanBatchElement,
-    SpanBatchError, SpanBatchLegacyTransactionData, SpanBatchTransactionData,
-    SpanBatchTransactions, SpanDecodingError, MAX_SPAN_BATCH_ELEMENTS, SINGLE_BATCH_TYPE,
-    SPAN_BATCH_TYPE,
+    SpanBatchError, SpanBatchLegacyTransactionData, SpanBatchPayload, SpanBatchPrefix,
+    SpanBatchTransactionData, SpanBatchTransactions, SpanDecodingError, MAX_SPAN_BATCH_ELEMENTS,
+    SINGLE_BATCH_TYPE, SPAN_BATCH_TYPE,
 };
 
 mod block;