diff --git a/crates/consensus/src/error.rs b/crates/consensus/src/error.rs
index d8fe6df..280f7e0 100644
--- a/crates/consensus/src/error.rs
+++ b/crates/consensus/src/error.rs
@@ -37,7 +37,7 @@ pub enum MorphConsensusError {
     BaseFeeOverLimit(u64),
 
     /// Invalid next L1 message index in header.
-    #[error("Invalid next L1 message index: expected {expected}, got {actual}")]
+    #[error("invalid block.NextL1MsgIndex: expected {expected}, got {actual}")]
     InvalidNextL1MessageIndex {
         /// Expected next L1 message index.
         expected: u64,
@@ -71,3 +71,22 @@ impl From<alloy_rlp::Error> for MorphConsensusError {
         Self::TransactionDecodeError(err.to_string())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn invalid_next_l1_message_index_error_matches_node_retry_classifier() {
+        let error = MorphConsensusError::InvalidNextL1MessageIndex {
+            expected: 2,
+            actual: 3,
+        }
+        .to_string();
+
+        assert!(
+            error.contains("invalid block.NextL1MsgIndex"),
+            "node treats this substring as a non-retryable error: {error}"
+        );
+    }
+}
diff --git a/crates/consensus/src/validation.rs b/crates/consensus/src/validation.rs
index bea0997..eab4883 100644
--- a/crates/consensus/src/validation.rs
+++ b/crates/consensus/src/validation.rs
@@ -89,13 +89,15 @@ const GAS_LIMIT_BOUND_DIVISOR: u64 = 1024;
 ///    start, have sequential queue indices, and are consistent with `header.next_l1_msg_index`.
 /// 2. **Cross-block monotonicity** (`validate_header_against_parent`): `header.next_l1_msg_index`
 ///    is monotonically non-decreasing relative to the parent.
+/// 3. **Parent-aware exactness** (`MorphBasicEngineValidator`): once the engine tree has
+///    both parent header and block body, Jade blocks must exactly match the value derived
+///    from `parent.next_l1_msg_index` and the block's leading L1 messages.
 ///
-/// These two methods have no ordering dependency and share no mutable state. The strict
+/// The consensus trait methods have no ordering dependency and share no mutable state. The strict
 /// cross-block equality check (`header.next == parent.next + l1_count`) requires simultaneous
 /// access to both parent header and block body, which reth's trait API does not provide in
-/// any single method. In Morph's single-sequencer model, the remaining gap (queue index
-/// skipping) is prevented by the trusted sequencer and verified by the L1 message queue
-/// contract.
+/// any single method, so Morph performs that final check in the engine tree payload validator
+/// before a block is accepted.
 #[derive(Debug, Clone)]
 pub struct MorphConsensus {
     /// Chain specification containing hardfork information and chain config.
@@ -321,6 +323,7 @@ impl Consensus<Block> for MorphConsensus {
         validate_l1_messages_in_block(
             &block.body().transactions,
             block.header().next_l1_msg_index,
+            is_jade,
         )?;
 
         Ok(())
@@ -476,10 +479,13 @@ fn validate_against_parent_gas_limit<H: BlockHeader>(
 /// 2. **Sequential Queue Index**: L1 messages must have strictly sequential
 ///    `queue_index` values (each = previous + 1).
 ///
-/// 3. **Header Consistency**: If L1 messages are present,
-///    `header.next_l1_msg_index` must be >= `last_queue_index + 1`. It may be
-///    strictly greater because Morph allows L1 messages to be "skipped" — the
-///    sequencer can advance past queue indices not included in the block body.
+/// 3. **Header Consistency**: If L1 messages are present, `header.next_l1_msg_index`
+///    is checked against `last_queue_index + 1`. The strictness depends on the fork:
+///    - **Jade onward** (`is_jade == true`): must equal `last_queue_index + 1` exactly,
+///      matching go-ethereum's `writeBlockStateWithoutHead` hard-fail (PR #331).
+///    - **Pre-Jade**: must be `>= last_queue_index + 1`; it may be strictly greater
+///      because the sequencer was permitted to "skip" queue indices not included in
+///      the block body ("early L1 msg skip").
 ///
 /// # Cross-Block Validation
 ///
@@ -491,8 +497,9 @@ fn validate_against_parent_gas_limit<H: BlockHeader>(
 ///
 /// ```text
 /// [L1Msg(queue=5), L1Msg(queue=6), L1Msg(queue=7), RegularTx]
-/// // header.next_l1_msg_index = 8  ✓ (exact match)
-/// // header.next_l1_msg_index = 10 ✓ (skipped queue indices 8, 9)
+/// // header.next_l1_msg_index = 8  ✓ (exact match, required from Jade onward)
+/// // header.next_l1_msg_index = 10 ✓ pre-Jade only (skipped queue indices 8, 9);
+/// //                              ❌ rejected from Jade onward
 /// ```
 ///
 /// # Example (Invalid - L1 after L2)
@@ -504,6 +511,7 @@ fn validate_against_parent_gas_limit<H: BlockHeader>(
 fn validate_l1_messages_in_block(
     txs: &[MorphTxEnvelope],
     header_next_l1_msg_index: u64,
+    is_jade: bool,
 ) -> Result<(), ConsensusError> {
     let mut l1_msg_count = 0u64;
     let mut saw_l2_transaction = false;
@@ -548,29 +556,37 @@ fn validate_l1_messages_in_block(
         }
     }
 
-    // Validate header consistency: header.next_l1_msg_index must be at least
-    // last_queue_index + 1 (cannot go backwards relative to included messages).
-    // It may be strictly greater because Morph allows L1 messages to be
-    // "skipped" — the sequencer can advance past queue indices that are not
-    // included in the block body (e.g., messages that failed on L1 relay).
-    // go-eth's NumL1MessagesProcessed() comment: "This count includes both
-    // skipped and included messages."
-    // For blocks with no L1 messages, this check is skipped — the cross-block
-    // monotonicity check in validate_header_against_parent handles that case.
+    // Validate header consistency against the L1 messages included in this block.
+    //
+    // Jade onward: if this block contains L1 messages, `header.next_l1_msg_index`
+    // must EXACTLY equal `last_queue_index + 1`.
+    //
+    // Pre-Jade keeps the lenient lower bound (`>= last_queue_index + 1`): the sequencer
+    // was permitted to advance past queue indices not included in the block body
+    // ("early L1 msg skip"), so the value may be strictly greater.
+    //
+    // For blocks with no L1 messages this stateless check is skipped. The engine
+    // tree payload validator performs the parent-aware exact check (`parent.next + 0`)
+    // once both parent header and block body are available.
     if l1_msg_count > 0 {
         let last_queue_index = prev_queue_index.ok_or_else(|| {
             ConsensusError::msg("internal error: l1_msg_count > 0 but prev_queue_index is None")
         })?;
-        let min_expected = last_queue_index.checked_add(1).ok_or_else(|| {
+        let expected = last_queue_index.checked_add(1).ok_or_else(|| {
             ConsensusError::other(MorphConsensusError::InvalidNextL1MessageIndex {
                 expected: u64::MAX,
                 actual: header_next_l1_msg_index,
             })
         })?;
-        if header_next_l1_msg_index < min_expected {
+        let inconsistent = if is_jade {
+            header_next_l1_msg_index != expected
+        } else {
+            header_next_l1_msg_index < expected
+        };
+        if inconsistent {
             return Err(ConsensusError::other(
                 MorphConsensusError::InvalidNextL1MessageIndex {
-                    expected: min_expected,
+                    expected,
                     actual: header_next_l1_msg_index,
                 },
             ));
@@ -811,7 +827,7 @@ mod tests {
         ];
 
         // L1 msgs: 0, 1 → last+1=2==header_next
-        assert!(validate_l1_messages_in_block(&txs, 2).is_ok());
+        assert!(validate_l1_messages_in_block(&txs, 2, true).is_ok());
     }
 
     #[test]
@@ -822,7 +838,7 @@ mod tests {
             create_l1_msg_tx(1),
         ];
 
-        assert!(validate_l1_messages_in_block(&txs, 2).is_err());
+        assert!(validate_l1_messages_in_block(&txs, 2, true).is_err());
     }
 
     #[test]
@@ -948,9 +964,9 @@ mod tests {
         // Empty block: no L1 messages → internal check always passes.
         // Any header_next value is accepted because the cross-block
         // monotonicity check is in validate_header_against_parent.
-        assert!(validate_l1_messages_in_block(&txs, 0).is_ok());
-        assert!(validate_l1_messages_in_block(&txs, 5).is_ok());
-        assert!(validate_l1_messages_in_block(&txs, 100).is_ok());
+        assert!(validate_l1_messages_in_block(&txs, 0, true).is_ok());
+        assert!(validate_l1_messages_in_block(&txs, 5, true).is_ok());
+        assert!(validate_l1_messages_in_block(&txs, 100, true).is_ok());
     }
 
     #[test]
@@ -962,7 +978,7 @@ mod tests {
         ];
 
         // last=2, 2+1=3==header_next
-        assert!(validate_l1_messages_in_block(&txs, 3).is_ok());
+        assert!(validate_l1_messages_in_block(&txs, 3, true).is_ok());
     }
 
     #[test]
@@ -974,7 +990,7 @@ mod tests {
         ];
 
         // No L1 messages → internal check passes (header_next not checked)
-        assert!(validate_l1_messages_in_block(&txs, 0).is_ok());
+        assert!(validate_l1_messages_in_block(&txs, 0, true).is_ok());
     }
 
     #[test]
@@ -982,7 +998,7 @@ mod tests {
         // Block has 0 then 2 (skipping 1) — caught by sequential check
         let txs = [create_l1_msg_tx(0), create_l1_msg_tx(2)];
 
-        let result = validate_l1_messages_in_block(&txs, 3);
+        let result = validate_l1_messages_in_block(&txs, 3, true);
         assert!(result.is_err());
         let err_str = result.unwrap_err().to_string();
         assert!(err_str.contains("expected 1"));
@@ -999,7 +1015,38 @@ mod tests {
         ];
 
         // last=101, 101+1=102==header_next
-        assert!(validate_l1_messages_in_block(&txs, 102).is_ok());
+        assert!(validate_l1_messages_in_block(&txs, 102, true).is_ok());
+    }
+
+    #[test]
+    fn test_validate_l1_messages_jade_rejects_skipped_forward_index() {
+        // Jade onward: header.next_l1_msg_index must equal last_queue_index + 1 exactly.
+        // Here last=1, so the only valid value is 2; a "skipped forward" 5 is rejected,
+        // matching go-ethereum's writeBlockStateWithoutHead hard-fail (PR #331).
+        let txs = [create_l1_msg_tx(0), create_l1_msg_tx(1)];
+        let result = validate_l1_messages_in_block(&txs, 5, true);
+        assert!(
+            result.is_err(),
+            "Jade must reject next_l1_msg_index > last_queue_index + 1"
+        );
+    }
+
+    #[test]
+    fn test_validate_l1_messages_pre_jade_allows_skipped_forward_index() {
+        // Pre-Jade keeps the lenient lower bound: the sequencer was permitted to advance
+        // past queue indices not included in the block body ("early L1 msg skip").
+        let txs = [create_l1_msg_tx(0), create_l1_msg_tx(1)];
+        assert!(
+            validate_l1_messages_in_block(&txs, 5, false).is_ok(),
+            "pre-Jade must allow next_l1_msg_index > last_queue_index + 1"
+        );
+    }
+
+    #[test]
+    fn test_validate_l1_messages_jade_accepts_exact_index() {
+        // The exact value (last_queue_index + 1) is accepted under Jade.
+        let txs = [create_l1_msg_tx(0), create_l1_msg_tx(1)];
+        assert!(validate_l1_messages_in_block(&txs, 2, true).is_ok());
     }
 
     #[test]
@@ -1007,7 +1054,7 @@ mod tests {
         // Duplicate index: 0, 0 — caught by sequential check (prev=0, expected 1, got 0)
         let txs = [create_l1_msg_tx(0), create_l1_msg_tx(0)];
 
-        let result = validate_l1_messages_in_block(&txs, 1);
+        let result = validate_l1_messages_in_block(&txs, 1, true);
         assert!(result.is_err());
         let err_str = result.unwrap_err().to_string();
         assert!(err_str.contains("expected 1"));
@@ -1019,7 +1066,7 @@ mod tests {
         // Block has 1 then 0 — caught by sequential check (prev=1, expected 2, got 0)
         let txs = [create_l1_msg_tx(1), create_l1_msg_tx(0)];
 
-        let result = validate_l1_messages_in_block(&txs, 2);
+        let result = validate_l1_messages_in_block(&txs, 2, true);
         assert!(result.is_err());
     }
 
@@ -1034,7 +1081,7 @@ mod tests {
         ];
 
         // Header says 2 but should be 3 (last=2, 2+1=3). Value < min_expected triggers error.
-        let result = validate_l1_messages_in_block(&txs, 2);
+        let result = validate_l1_messages_in_block(&txs, 2, true);
         assert!(result.is_err());
         let err_str = result.unwrap_err().to_string();
         assert!(err_str.contains("expected 3"));
@@ -1051,7 +1098,7 @@ mod tests {
             create_l1_msg_tx(2),
         ];
 
-        assert!(validate_l1_messages_in_block(&txs, 3).is_err());
+        assert!(validate_l1_messages_in_block(&txs, 3, true).is_err());
     }
 
     // ========================================================================
@@ -1946,7 +1993,7 @@ mod tests {
         let txs = [create_l1_msg_tx(u64::MAX - 1), create_l1_msg_tx(u64::MAX)];
 
         // last=MAX, MAX+1 overflows
-        let result = validate_l1_messages_in_block(&txs, 0);
+        let result = validate_l1_messages_in_block(&txs, 0, true);
         assert!(result.is_err());
     }
 
@@ -1954,9 +2001,9 @@ mod tests {
     fn test_validate_l1_messages_in_block_single_l1() {
         let txs = [create_l1_msg_tx(42)];
         // last=42, 42+1=43==header_next
-        assert!(validate_l1_messages_in_block(&txs, 43).is_ok());
+        assert!(validate_l1_messages_in_block(&txs, 43, true).is_ok());
         // Wrong header_next
-        assert!(validate_l1_messages_in_block(&txs, 42).is_err());
+        assert!(validate_l1_messages_in_block(&txs, 42, true).is_err());
     }
 
     // ========================================================================
diff --git a/crates/engine-api/src/api.rs b/crates/engine-api/src/api.rs
index f5c9082..42cff59 100644
--- a/crates/engine-api/src/api.rs
+++ b/crates/engine-api/src/api.rs
@@ -6,7 +6,9 @@
 
 use crate::EngineApiResult;
 use alloy_primitives::B256;
-use morph_payload_types::{AssembleL2BlockParams, ExecutableL2Data, GenericResponse, SafeL2Data};
+use morph_payload_types::{
+    AssembleL2BlockParams, AssembleL2BlockV2Params, ExecutableL2Data, GenericResponse, SafeL2Data,
+};
 use morph_primitives::MorphHeader;
 
 /// Morph L2 Engine API trait.
@@ -45,6 +47,28 @@ pub trait MorphL2EngineApi: Send + Sync {
         params: AssembleL2BlockParams,
     ) -> EngineApiResult<ExecutableL2Data>;
 
+    /// Build a new L2 block on an explicitly given parent hash.
+    ///
+    /// Unlike [`assemble_l2_block`](Self::assemble_l2_block), which builds on the
+    /// current canonical head keyed by block number, this method keys assembly on
+    /// `parent_hash`, allowing the sequencer to build on any existing parent — including
+    /// one that is no longer the head. The block number is derived as `parent + 1`.
+    ///
+    /// # Arguments
+    ///
+    /// * `params` - The parameters for assembling the block, including:
+    ///   - `parent_hash`: Hash of the parent block to build on
+    ///   - `transactions`: RLP-encoded transactions to include in the block
+    ///   - `timestamp`: Optional block timestamp; defaults to a local clock value
+    ///
+    /// # Returns
+    ///
+    /// Returns the execution result including state root, receipts root, etc.
+    async fn assemble_l2_block_v2(
+        &self,
+        params: AssembleL2BlockV2Params,
+    ) -> EngineApiResult<ExecutableL2Data>;
+
     /// Validate an L2 block without importing it.
     ///
     /// This method validates a block by forwarding it to the reth engine tree
@@ -75,6 +99,24 @@ pub trait MorphL2EngineApi: Send + Sync {
     /// Returns `Ok(())` on success.
     async fn new_l2_block(&self, data: ExecutableL2Data) -> EngineApiResult<()>;
 
+    /// Import a new L2 block, selecting the parent by `data.parent_hash`.
+    ///
+    /// Unlike [`new_l2_block`](Self::new_l2_block), which requires the block to
+    /// extend the current canonical head, this method only requires that the
+    /// parent referenced by `data.parent_hash` exists. When that parent is not the
+    /// current head, the engine's forkchoice update reorganizes the canonical chain
+    /// onto the new block. This is the import path used by the centralized
+    /// sequencer, where recent blocks may be rebuilt and replaced.
+    ///
+    /// # Arguments
+    ///
+    /// * `data` - The block data to import
+    ///
+    /// # Returns
+    ///
+    /// Returns the header of the imported block.
+    async fn new_l2_block_v2(&self, data: ExecutableL2Data) -> EngineApiResult<MorphHeader>;
+
     /// Import a safe L2 block from derivation.
     ///
     /// This method is used by the derivation pipeline to import blocks that
diff --git a/crates/engine-api/src/builder.rs b/crates/engine-api/src/builder.rs
index 97fb6de..20b6221 100644
--- a/crates/engine-api/src/builder.rs
+++ b/crates/engine-api/src/builder.rs
@@ -13,8 +13,8 @@ use alloy_primitives::{Address, B64, B256, Sealable};
 use alloy_rpc_types_engine::{PayloadAttributes, PayloadStatus, PayloadStatusEnum};
 use morph_chainspec::MorphChainSpec;
 use morph_payload_types::{
-    AssembleL2BlockParams, ExecutableL2Data, GenericResponse, MorphBuiltPayload,
-    MorphExecutionData, MorphPayloadTypes, SafeL2Data,
+    AssembleL2BlockParams, AssembleL2BlockV2Params, ExecutableL2Data, GenericResponse,
+    MorphBuiltPayload, MorphExecutionData, MorphPayloadTypes, SafeL2Data,
 };
 use morph_primitives::{Block, BlockBody, MorphHeader, MorphTxEnvelope};
 use parking_lot::RwLock;
@@ -171,7 +171,7 @@ where
         params: AssembleL2BlockParams,
     ) -> EngineApiResult<ExecutableL2Data> {
         let started = Instant::now();
-        let result = self.build_l2_payload(params, None, None).await;
+        let result = self.build_l2_payload(params, None, None, None).await;
         self.metrics
             .assemble_l2_block_duration_seconds
             .record(started.elapsed());
@@ -192,6 +192,54 @@ where
         Ok(executable_data)
     }
 
+    async fn assemble_l2_block_v2(
+        &self,
+        params: AssembleL2BlockV2Params,
+    ) -> EngineApiResult<ExecutableL2Data> {
+        let started = Instant::now();
+        let parent_hash = params.parent_hash;
+
+        // Derive the block number from the pinned parent (parent + 1). The parent is
+        // looked up by hash and need not be the canonical head — that is the point of V2
+        // (the sequencer can build on a parent that diverges from its current head).
+        let parent = self
+            .provider
+            .sealed_header_by_hash(parent_hash)
+            .map_err(|e| MorphEngineApiError::Database(e.to_string()))?
+            .ok_or_else(|| {
+                MorphEngineApiError::Internal(format!("parent block not found: {parent_hash}"))
+            })?;
+
+        let assemble_params = AssembleL2BlockParams {
+            number: parent.number() + 1,
+            transactions: params.transactions,
+            timestamp: params.timestamp,
+        };
+
+        let result = self
+            .build_l2_payload(assemble_params, None, None, Some(parent_hash))
+            .await;
+        self.metrics
+            .assemble_l2_block_duration_seconds
+            .record(started.elapsed());
+
+        let built_payload = result.inspect_err(|_| {
+            self.metrics.assemble_l2_block_failures_total.increment(1);
+        })?;
+        let executable_data = built_payload.executable_data;
+
+        tracing::debug!(
+            target: "morph::engine",
+            block_hash = %executable_data.hash,
+            parent_hash = %parent_hash,
+            gas_used = executable_data.gas_used,
+            tx_count = executable_data.transactions.len(),
+            "L2 block assembled successfully (v2)"
+        );
+
+        Ok(executable_data)
+    }
+
     async fn validate_l2_block(&self, data: ExecutableL2Data) -> EngineApiResult<GenericResponse> {
         let validate_started = Instant::now();
         tracing::debug!(
@@ -400,31 +448,85 @@ where
         Ok(())
     }
 
-    async fn new_safe_l2_block(&self, mut data: SafeL2Data) -> EngineApiResult<MorphHeader> {
+    async fn new_l2_block_v2(&self, data: ExecutableL2Data) -> EngineApiResult<MorphHeader> {
         let started = Instant::now();
         tracing::debug!(
             target: "morph::engine",
             block_number = data.number,
-            "importing safe L2 block from L1 derivation"
+            block_hash = %data.hash,
+            parent_hash = %data.parent_hash,
+            "importing new L2 block (v2, reorg-capable)"
         );
 
-        // 1. Get latest block number
-        let latest_number = self.current_head()?.number;
+        // 1. Parent selection by hash. Relaxed from V1's "parent must be the current
+        //    head" to "parent must exist": when the parent is not the canonical head,
+        //    the forkchoice update inside import_l2_block_via_engine reorganizes the
+        //    chain onto this block. This is the centralized-sequencer import path,
+        //    where the sequencer may rebuild and replace recent blocks.
+        let parent = self
+            .provider
+            .sealed_header_by_hash(data.parent_hash)
+            .map_err(|e| MorphEngineApiError::Database(e.to_string()))?
+            .ok_or_else(|| {
+                MorphEngineApiError::Internal(format!(
+                    "parent block not found: {}",
+                    data.parent_hash
+                ))
+            })?;
 
-        if data.number != latest_number + 1 {
-            self.metrics.new_safe_l2_block_failures_total.increment(1);
+        // 2. Block number must be exactly parent + 1.
+        let expected_number = parent.number() + 1;
+        if data.number != expected_number {
+            self.metrics.new_l2_block_failures_total.increment(1);
             self.metrics
-                .new_safe_l2_block_duration_seconds
+                .new_l2_block_duration_seconds
                 .record(started.elapsed());
             return Err(MorphEngineApiError::DiscontinuousBlockNumber {
-                expected: latest_number + 1,
+                expected: expected_number,
                 actual: data.number,
             });
         }
 
+        // 3. Import via the engine tree (newPayload + forkchoiceUpdated). The hash check
+        //    against data.hash happens inside execution_payload_from_executable_data; the
+        //    FCU advances or reorgs the canonical head onto data.hash.
         let block_timestamp = data.timestamp;
+        let header = self
+            .import_l2_block_via_engine(data)
+            .await
+            .inspect_err(|_| {
+                self.metrics.new_l2_block_failures_total.increment(1);
+                self.metrics
+                    .new_l2_block_duration_seconds
+                    .record(started.elapsed());
+            })?;
+
+        self.metrics
+            .new_l2_block_duration_seconds
+            .record(started.elapsed());
+        self.record_head_metrics(block_timestamp);
+
+        Ok(header)
+    }
 
-        // 2. Assemble the block from SafeL2Data inputs.
+    async fn new_safe_l2_block(&self, mut data: SafeL2Data) -> EngineApiResult<MorphHeader> {
+        let started = Instant::now();
+        tracing::debug!(
+            target: "morph::engine",
+            block_number = data.number,
+            parent_hash = ?data.parent_hash,
+            "importing safe L2 block from L1 derivation"
+        );
+
+        let block_timestamp = data.timestamp;
+
+        // Parent selection: caller-pinned (derivation reorg path, deriveForce) or the
+        // current head (legacy sequential path). The block-number invariant
+        // (`number == parent + 1`) is validated inside build_l2_payload against the
+        // resolved parent, so callers that pin a non-head parent reorg correctly.
+        let parent_override = data.parent_hash;
+
+        // Assemble the block from SafeL2Data inputs.
         let assemble_params = AssembleL2BlockParams {
             number: data.number,
             // Move transactions out of data to avoid cloning the full Vec<Bytes>.
@@ -433,7 +535,12 @@ where
         };
 
         let built_payload = self
-            .build_l2_payload(assemble_params, Some(data.gas_limit), data.base_fee_per_gas)
+            .build_l2_payload(
+                assemble_params,
+                Some(data.gas_limit),
+                data.base_fee_per_gas,
+                parent_override,
+            )
             .await
             .inspect_err(|_| {
                 self.metrics.new_safe_l2_block_failures_total.increment(1);
@@ -596,6 +703,7 @@ impl<Provider> RealMorphL2EngineApi<Provider> {
         params: AssembleL2BlockParams,
         gas_limit_override: Option<u64>,
         base_fee_override: Option<u128>,
+        parent_override: Option<B256>,
     ) -> EngineApiResult<MorphBuiltPayload>
     where
         Provider:
@@ -608,20 +716,45 @@ impl<Provider> RealMorphL2EngineApi<Provider> {
             "assembling L2 block"
         );
 
-        // 1. Validate block number (must be current_head + 1).
-        let current_head = self.current_head()?;
-        if params.number != current_head.number + 1 {
+        // 1. Resolve the parent: caller-pinned (reorg path, e.g. derivation deriveForce
+        //    or assembleL2BlockV2) or the current canonical head (sequential path). When
+        //    pinned, the parent need not be the head — building on it lets the subsequent
+        //    forkchoice update reorganize the chain onto the new block.
+        let (parent_number, parent_hash, parent_timestamp) = match parent_override {
+            Some(parent_hash) => {
+                let parent = self
+                    .provider
+                    .sealed_header_by_hash(parent_hash)
+                    .map_err(|e| MorphEngineApiError::Database(e.to_string()))?
+                    .ok_or_else(|| {
+                        MorphEngineApiError::Internal(format!(
+                            "parent block not found: {parent_hash}"
+                        ))
+                    })?;
+                (parent.number(), parent_hash, parent.timestamp())
+            }
+            None => {
+                let current_head = self.current_head()?;
+                (
+                    current_head.number,
+                    current_head.hash,
+                    current_head.timestamp,
+                )
+            }
+        };
+
+        // 2. Validate block number (must be parent + 1).
+        if params.number != parent_number + 1 {
             return Err(MorphEngineApiError::DiscontinuousBlockNumber {
-                expected: current_head.number + 1,
+                expected: parent_number + 1,
                 actual: params.number,
             });
         }
 
-        // 2. Build payload attributes.
-        let parent_hash = current_head.hash;
+        // 3. Build payload attributes.
         let timestamp = params.timestamp.unwrap_or_else(|| {
             std::cmp::max(
-                current_head.timestamp + 1,
+                parent_timestamp + 1,
                 std::time::SystemTime::now()
                     .duration_since(std::time::UNIX_EPOCH)
                     .unwrap_or_default()
@@ -713,9 +846,6 @@ impl<Provider> RealMorphL2EngineApi<Provider> {
         let new_payload_elapsed = new_payload_started.elapsed();
         ensure_payload_status_valid(&payload_status, "newPayload")?;
 
-        // Morph uses Tendermint consensus with instant finality — every committed
-        // block is final and no reorgs are possible.
-        //
         // The safe/finalized hashes passed here serve two purposes in reth's engine
         // tree: (1) driving changeset-cache eviction and sidechain pruning (memory
         // management), and (2) setting the RPC-visible "safe"/"finalized" block tags.
diff --git a/crates/engine-api/src/rpc.rs b/crates/engine-api/src/rpc.rs
index 3282e17..9273ca9 100644
--- a/crates/engine-api/src/rpc.rs
+++ b/crates/engine-api/src/rpc.rs
@@ -6,7 +6,9 @@
 use crate::{EngineApiResult, api::MorphL2EngineApi};
 use alloy_primitives::B256;
 use jsonrpsee::{RpcModule, core::RpcResult, proc_macros::rpc};
-use morph_payload_types::{AssembleL2BlockParams, ExecutableL2Data, GenericResponse, SafeL2Data};
+use morph_payload_types::{
+    AssembleL2BlockParams, AssembleL2BlockV2Params, ExecutableL2Data, GenericResponse, SafeL2Data,
+};
 use morph_primitives::MorphHeader;
 use reth_rpc_api::IntoEngineApiRpcModule;
 use std::sync::Arc;
@@ -26,6 +28,17 @@ pub trait MorphL2EngineRpc {
     async fn assemble_l2_block(&self, params: AssembleL2BlockParams)
     -> RpcResult<ExecutableL2Data>;
 
+    /// Build a new L2 block on an explicitly given parent hash.
+    ///
+    /// # JSON-RPC Method
+    ///
+    /// `engine_assembleL2BlockV2`
+    #[method(name = "assembleL2BlockV2")]
+    async fn assemble_l2_block_v2(
+        &self,
+        params: AssembleL2BlockV2Params,
+    ) -> RpcResult<ExecutableL2Data>;
+
     /// Validate an L2 block without importing it.
     ///
     /// # JSON-RPC Method
@@ -42,6 +55,14 @@ pub trait MorphL2EngineRpc {
     #[method(name = "newL2Block")]
     async fn new_l2_block(&self, data: ExecutableL2Data) -> RpcResult<()>;
 
+    /// Import a new L2 block with reorg support (parent selected by hash).
+    ///
+    /// # JSON-RPC Method
+    ///
+    /// `engine_newL2BlockV2`
+    #[method(name = "newL2BlockV2")]
+    async fn new_l2_block_v2(&self, data: ExecutableL2Data) -> RpcResult<MorphHeader>;
+
     /// Import a safe L2 block from derivation.
     ///
     /// # JSON-RPC Method
@@ -103,6 +124,26 @@ where
         })
     }
 
+    async fn assemble_l2_block_v2(
+        &self,
+        params: AssembleL2BlockV2Params,
+    ) -> RpcResult<ExecutableL2Data> {
+        tracing::debug!(
+            target: "morph::engine",
+            parent_hash = %params.parent_hash,
+            ?params.timestamp,
+            "assembling L2 block (v2)"
+        );
+
+        self.inner
+            .assemble_l2_block_v2(params)
+            .await
+            .map_err(|e| {
+                tracing::error!(target: "morph::engine", error = %e, "failed to assemble L2 block (v2)");
+                e.into()
+            })
+    }
+
     async fn validate_l2_block(&self, data: ExecutableL2Data) -> RpcResult<GenericResponse> {
         tracing::debug!(
             target: "morph::engine",
@@ -131,6 +172,20 @@ where
         })
     }
 
+    async fn new_l2_block_v2(&self, data: ExecutableL2Data) -> RpcResult<MorphHeader> {
+        tracing::debug!(
+            target: "morph::engine",
+            block_number = data.number,
+            block_hash = %data.hash,
+            "RPC newL2BlockV2 called"
+        );
+
+        self.inner.new_l2_block_v2(data).await.map_err(|e| {
+            tracing::error!(target: "morph::engine", error = %e, "failed to import L2 block (v2)");
+            e.into()
+        })
+    }
+
     async fn new_safe_l2_block(&self, data: SafeL2Data) -> RpcResult<MorphHeader> {
         tracing::debug!(
             target: "morph::engine",
diff --git a/crates/engine-tree-ext/src/payload_validator.rs b/crates/engine-tree-ext/src/payload_validator.rs
index a0455e0..5d52e08 100644
--- a/crates/engine-tree-ext/src/payload_validator.rs
+++ b/crates/engine-tree-ext/src/payload_validator.rs
@@ -60,7 +60,8 @@ use reth_revm::database::StateProviderDatabase;
 use reth_trie_sparse::debug_recorder::TrieDebugRecorder;
 
 use crate::gate::state_root_enforced_at;
-use morph_chainspec::MorphChainSpec;
+use morph_chainspec::{MorphChainSpec, MorphHardforks};
+use morph_primitives::{MorphHeader, MorphTxEnvelope};
 use reth_chain_state::{DeferredTrieData, ExecutedBlock, ExecutionTimingStats, LazyOverlay};
 use reth_consensus::{ConsensusError, FullConsensus, ReceiptRootBloom};
 use reth_engine_primitives::{
@@ -164,7 +165,11 @@ where
 
 impl<N, P, Evm, V> MorphBasicEngineValidator<P, Evm, V>
 where
-    N: NodePrimitives,
+    N: NodePrimitives<
+            Block = morph_primitives::Block,
+            BlockHeader = MorphHeader,
+            SignedTx = MorphTxEnvelope,
+        >,
     P: DatabaseProviderFactory<
             Provider: BlockReader
                           + StageCheckpointReader
@@ -872,6 +877,54 @@ where
         Ok(())
     }
 
+    /// Validates Morph's parent-aware L1 message index invariant.
+    ///
+    /// go-ethereum derives `NextL1MsgIndex` from the parent queue index plus the
+    /// L1 messages processed by this block. Since `next_l1_msg_index` is not part
+    /// of the block hash, Jade and later must reject any mismatch before accepting
+    /// the payload.
+    fn validate_next_l1_msg_index_against_parent(
+        &self,
+        block: &RecoveredBlock<morph_primitives::Block>,
+        parent_block: &SealedHeader<MorphHeader>,
+    ) -> Result<(), ConsensusError> {
+        if !self
+            .chain_spec
+            .is_jade_active_at_timestamp(block.header().timestamp())
+        {
+            return Ok(());
+        }
+
+        let mut expected = parent_block.next_l1_msg_index;
+        for tx in block.body().transactions() {
+            if !tx.is_l1_msg() {
+                break;
+            }
+
+            let queue_index = tx.queue_index().ok_or_else(|| {
+                ConsensusError::msg("L1 message transaction is missing queue index")
+            })?;
+            // Match geth's `NumL1MessagesProcessed(parent_queue_index)`: forward skips
+            // are counted as processed, so the derived next index is last_queue + 1.
+            expected = queue_index.checked_add(1).ok_or_else(|| {
+                ConsensusError::msg(format!(
+                    "invalid block.NextL1MsgIndex: expected {}, got {}",
+                    u64::MAX,
+                    block.header().next_l1_msg_index
+                ))
+            })?;
+        }
+
+        let actual = block.header().next_l1_msg_index;
+        if actual != expected {
+            return Err(ConsensusError::msg(format!(
+                "invalid block.NextL1MsgIndex: expected {expected}, got {actual}"
+            )));
+        }
+
+        Ok(())
+    }
+
     /// Executes a block with the given state provider.
     ///
     /// This method orchestrates block execution:
@@ -1390,6 +1443,16 @@ where
         }
         drop(_enter);
 
+        if let Err(e) = self.validate_next_l1_msg_index_against_parent(block, parent_block) {
+            warn!(
+                target: "engine::tree::payload_validator",
+                ?block,
+                "Failed to validate NextL1MsgIndex against parent for block {}: {e}",
+                block.hash()
+            );
+            return Err(e.into());
+        }
+
         // Validate block post-execution rules
         let _enter =
             debug_span!(target: "engine::tree::payload_validator", "validate_block_post_execution")
@@ -1955,7 +2018,11 @@ where
         + HashedPostStateProvider
         + Clone
         + 'static,
-    N: NodePrimitives,
+    N: NodePrimitives<
+            Block = morph_primitives::Block,
+            BlockHeader = MorphHeader,
+            SignedTx = MorphTxEnvelope,
+        >,
     V: PayloadValidator<Types, Block = N::Block> + Clone,
     Evm: ConfigureEngineEvm<Types::ExecutionData, Primitives = N> + 'static,
     Types: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>,
diff --git a/crates/node/src/validator.rs b/crates/node/src/validator.rs
index 72c3299..5d288d3 100644
--- a/crates/node/src/validator.rs
+++ b/crates/node/src/validator.rs
@@ -74,10 +74,11 @@ where
 impl<Node, PVB> EngineValidatorBuilder<Node> for MorphTreeEngineValidatorBuilder<PVB>
 where
     Node: FullNodeComponents<
-        Evm: reth_node_api::ConfigureEngineEvm<
-            <<Node::Types as NodeTypes>::Payload as PayloadTypes>::ExecutionData,
+            Types = MorphNode,
+            Evm: reth_node_api::ConfigureEngineEvm<
+                <<Node::Types as NodeTypes>::Payload as PayloadTypes>::ExecutionData,
+            >,
         >,
-    >,
     Node::Provider: ChainSpecProvider<ChainSpec = MorphChainSpec>,
     PVB: PayloadValidatorBuilder<Node>,
     PVB::Validator: reth_node_api::PayloadValidator<
diff --git a/crates/node/tests/it/consensus.rs b/crates/node/tests/it/consensus.rs
index b365f4c..2e373d3 100644
--- a/crates/node/tests/it/consensus.rs
+++ b/crates/node/tests/it/consensus.rs
@@ -254,6 +254,38 @@ async fn next_l1_msg_index_decreases_rejected() -> eyre::Result<()> {
     Ok(())
 }
 
+/// From Jade onward, an empty block must not advance `next_l1_msg_index`.
+///
+/// geth PR #331 derives the expected value from the parent index plus the number of
+/// processed L1 messages in the block. For an empty block that count is zero, so the
+/// child header must keep the parent's value exactly.
+#[tokio::test(flavor = "multi_thread")]
+async fn next_l1_msg_index_empty_block_cannot_advance_post_jade() -> eyre::Result<()> {
+    reth_tracing::init_test_tracing();
+    let (mut nodes, _wallet) = TestNodeBuilder::new()
+        .with_schedule(HardforkSchedule::AllActive)
+        .build()
+        .await?;
+    let mut node = nodes.pop().unwrap();
+
+    // Block 1 includes queue indices 0 and 1, so the parent index becomes 2.
+    let l1_msgs = L1MessageBuilder::build_sequential(0, 2);
+    advance_block_with_l1_messages(&mut node, l1_msgs).await?;
+
+    // Block 2 has no L1 messages. Advancing the index would skip queue index 2.
+    let base = build_block_no_submit(&mut node, vec![]).await?;
+    let accepted = craft_and_try_import_block(&mut node, &base, |block| {
+        block.header.next_l1_msg_index = 3;
+    })
+    .await?;
+
+    assert!(
+        !accepted,
+        "post-Jade: empty block must not advance next_l1_msg_index"
+    );
+    Ok(())
+}
+
 /// A block with L1 messages but next_l1_msg_index too low is rejected.
 ///
 /// Block has L1 messages with queue indices 0, 1 -> next_l1_msg_index should be >= 2.
@@ -277,11 +309,49 @@ async fn next_l1_msg_index_insufficient_for_l1_msgs() -> eyre::Result<()> {
     Ok(())
 }
 
-/// A block may advance `next_l1_msg_index` past the included messages to account for skips.
+/// From Jade onward, `next_l1_msg_index` must equal `last_queue_index + 1` exactly:
+/// advancing past the included messages (a trailing skip) is rejected.
+///
+/// Matches go-ethereum's `writeBlockStateWithoutHead` hard-fail (PR #331). The field is
+/// not covered by the block hash, so an inflated value — which would make the consensus
+/// client skip unprocessed L1 messages — is rejected against the in-block L1 message
+/// stream. The pre-Jade counterpart below still permits the skip.
 #[tokio::test(flavor = "multi_thread")]
-async fn next_l1_msg_index_can_skip_past_included_messages() -> eyre::Result<()> {
+async fn next_l1_msg_index_skip_past_included_messages_rejected_post_jade() -> eyre::Result<()> {
     reth_tracing::init_test_tracing();
-    let (mut nodes, _wallet) = TestNodeBuilder::new().build().await?;
+    let (mut nodes, _wallet) = TestNodeBuilder::new()
+        .with_schedule(HardforkSchedule::AllActive)
+        .build()
+        .await?;
+    let mut node = nodes.pop().unwrap();
+
+    // Block includes queue indices 0,1 (last=1, so the only valid next is 2); advancing
+    // header.next_l1_msg_index to 4 (skipping 2,3) must be rejected under Jade.
+    let l1_msgs = L1MessageBuilder::build_sequential(0, 2);
+    let base = build_block_no_submit(&mut node, l1_msgs).await?;
+
+    let accepted = craft_and_try_import_block(&mut node, &base, |block| {
+        block.header.next_l1_msg_index = 4;
+    })
+    .await?;
+
+    assert!(
+        !accepted,
+        "post-Jade: next_l1_msg_index advanced past the included messages must be rejected"
+    );
+    Ok(())
+}
+
+/// Pre-Jade, a block may advance `next_l1_msg_index` past the included messages to
+/// represent skipped queue indices ("early L1 msg skip"). The Jade counterpart above
+/// rejects this; together they pin the Jade hardfork boundary for the exact-index rule.
+#[tokio::test(flavor = "multi_thread")]
+async fn next_l1_msg_index_can_skip_past_included_messages_pre_jade() -> eyre::Result<()> {
+    reth_tracing::init_test_tracing();
+    let (mut nodes, _wallet) = TestNodeBuilder::new()
+        .with_schedule(HardforkSchedule::PreJade)
+        .build()
+        .await?;
     let mut node = nodes.pop().unwrap();
 
     // Build block with queue indices 0,1 and then advance header.next_l1_msg_index to 4.
@@ -296,7 +366,94 @@ async fn next_l1_msg_index_can_skip_past_included_messages() -> eyre::Result<()>
 
     assert!(
         accepted,
-        "next_l1_msg_index may advance past included L1 messages to represent skipped queue indices"
+        "pre-Jade: next_l1_msg_index may advance past included L1 messages (skipped indices)"
+    );
+    Ok(())
+}
+
+/// From Jade onward, a block whose first L1 message skips past the parent's queue
+/// index is accepted when the header matches geth's derived processed count.
+///
+/// The parent index is 2, but the block's leading L1 message claims queue index 5,
+/// silently dropping queue indices 2, 3, 4. The in-block messages are internally
+/// contiguous (5, 6) and the header's `next_l1_msg_index` equals last + 1 (= 7).
+/// This matches go-ethereum PR #331: `NumL1MessagesProcessed(parent.next)` derives
+/// `last_queue_index - parent.next + 1`, so forward skips are counted as processed.
+/// `queue_index` does not affect execution (L1 message nonce is always 0), so the
+/// crafted block's state root stays valid while exercising only the index rule.
+#[tokio::test(flavor = "multi_thread")]
+async fn next_l1_msg_index_first_msg_skips_parent_allowed_post_jade() -> eyre::Result<()> {
+    reth_tracing::init_test_tracing();
+    let (mut nodes, _wallet) = TestNodeBuilder::new()
+        .with_schedule(HardforkSchedule::AllActive)
+        .build()
+        .await?;
+    let mut node = nodes.pop().unwrap();
+
+    // Block 1 includes queue indices 0, 1, so the parent index becomes 2.
+    advance_block_with_l1_messages(&mut node, L1MessageBuilder::build_sequential(0, 2)).await?;
+
+    // Block 2 is built with the valid continuation (queue 2, 3) so it executes and seals
+    // with a correct state root, then we rewrite the queue indices to 5, 6.
+    let base = build_block_no_submit(&mut node, L1MessageBuilder::build_sequential(2, 2)).await?;
+
+    let accepted = craft_and_try_import_block(&mut node, &base, |block| {
+        rewrite_l1_queue_indices(block, 3);
+        // Keep the header consistent with the rewritten in-block stream (last 6 + 1).
+        block.header.next_l1_msg_index = 7;
+    })
+    .await?;
+
+    assert!(
+        accepted,
+        "post-Jade: forward skips are accepted when header.next matches geth's derived index"
     );
     Ok(())
 }
+
+/// Pre-Jade, the parent-aware exact-index rule is not enforced, so a forward skip of
+/// queue indices is also tolerated. Pins the fork boundary against the post-Jade
+/// trailing-skip rejection tests above.
+#[tokio::test(flavor = "multi_thread")]
+async fn next_l1_msg_index_first_msg_skips_parent_allowed_pre_jade() -> eyre::Result<()> {
+    reth_tracing::init_test_tracing();
+    let (mut nodes, _wallet) = TestNodeBuilder::new()
+        .with_schedule(HardforkSchedule::PreJade)
+        .build()
+        .await?;
+    let mut node = nodes.pop().unwrap();
+
+    advance_block_with_l1_messages(&mut node, L1MessageBuilder::build_sequential(0, 2)).await?;
+    let base = build_block_no_submit(&mut node, L1MessageBuilder::build_sequential(2, 2)).await?;
+
+    let accepted = craft_and_try_import_block(&mut node, &base, |block| {
+        rewrite_l1_queue_indices(block, 3);
+        block.header.next_l1_msg_index = 7;
+    })
+    .await?;
+
+    assert!(
+        accepted,
+        "pre-Jade: a forward skip of L1 queue indices is tolerated"
+    );
+    Ok(())
+}
+
+/// Shift every leading L1 message's `queue_index` up by `delta`, re-sealing each tx.
+///
+/// Used to craft a block whose L1 stream no longer continues the parent's queue index
+/// while keeping execution identical (L1 message nonce is always 0, so `queue_index`
+/// does not influence the state root). Re-sealing recomputes the cached hash; the wire
+/// encoding and the recomputed transactions root both reflect the new indices.
+fn rewrite_l1_queue_indices(block: &mut morph_primitives::Block, delta: u64) {
+    use alloy_consensus::Sealable;
+    use morph_primitives::MorphTxEnvelope;
+
+    for tx in block.body.transactions.iter_mut() {
+        if let MorphTxEnvelope::L1Msg(sealed) = tx {
+            let mut inner = sealed.inner().clone();
+            inner.queue_index += delta;
+            *sealed = inner.seal_slow();
+        }
+    }
+}
diff --git a/crates/node/tests/it/engine.rs b/crates/node/tests/it/engine.rs
index 9952132..064452a 100644
--- a/crates/node/tests/it/engine.rs
+++ b/crates/node/tests/it/engine.rs
@@ -4,14 +4,15 @@
 //! enforcement — in particular the state-root validation gating introduced
 //! by the Jade hardfork.
 
-use alloy_consensus::BlockHeader;
+use alloy_consensus::{BlockHeader, Sealable};
 use alloy_primitives::{Address, B256};
 use alloy_rpc_types_engine::PayloadAttributes;
 use jsonrpsee::core::client::ClientT;
 use morph_node::test_utils::{HardforkSchedule, TestNodeBuilder};
 use morph_payload_types::{
-    AssembleL2BlockParams, ExecutableL2Data, GenericResponse, MorphPayloadAttributes,
+    AssembleL2BlockParams, ExecutableL2Data, GenericResponse, MorphPayloadAttributes, SafeL2Data,
 };
+use morph_primitives::MorphHeader;
 use reth_payload_builder::BuildNewPayload;
 use reth_payload_primitives::BuiltPayload;
 use reth_provider::BlockReaderIdExt;
@@ -117,6 +118,292 @@ async fn new_l2_block_imports_consecutive_assembled_blocks_over_rpc() -> eyre::R
     Ok(())
 }
 
+/// `engine_newL2BlockV2` imports a block built on the current head and returns its header.
+///
+/// V2 selects the parent via `data.parent_hash` (rather than requiring it to equal the
+/// current head as V1 does). For a block extending the head the two behave identically;
+/// this pins the additive happy path before the reorg behavior is exercised separately.
+#[tokio::test(flavor = "multi_thread")]
+async fn new_l2_block_v2_imports_block_on_current_head() -> eyre::Result<()> {
+    reth_tracing::init_test_tracing();
+
+    let (mut nodes, _wallet) = TestNodeBuilder::new().build().await?;
+    let node = nodes.pop().unwrap();
+
+    let auth = node.auth_server_handle();
+    let client = auth.http_client();
+    let mut params = AssembleL2BlockParams::empty(1);
+    params.timestamp = Some(1);
+
+    let data: ExecutableL2Data = client.request("engine_assembleL2Block", (params,)).await?;
+    let expected_hash = data.hash;
+
+    let header: MorphHeader = client.request("engine_newL2BlockV2", (data,)).await?;
+
+    assert_eq!(header.number(), 1, "returned header should be block 1");
+    assert_eq!(
+        header.hash_slow(),
+        expected_hash,
+        "returned header hash should match the assembled block"
+    );
+
+    let latest = node
+        .inner
+        .provider
+        .sealed_header_by_number_or_tag(alloy_rpc_types_eth::BlockNumberOrTag::Latest)?
+        .expect("latest header must exist after importing the block");
+    assert_eq!(
+        latest.number(),
+        1,
+        "engine_newL2BlockV2 should advance the head"
+    );
+    assert_eq!(
+        latest.hash(),
+        expected_hash,
+        "imported canonical head should match the assembled block hash"
+    );
+
+    Ok(())
+}
+
+/// `engine_newL2BlockV2` reorganizes the canonical chain onto a sibling block.
+///
+/// Two distinct blocks are assembled at height 2 on the same parent (block 1) while the
+/// head still points at block 1. Importing the first makes it canonical; importing the
+/// second — which builds on the same parent, not on the new head — must reorg the head
+/// onto it. This is the core capability the centralized sequencer relies on
+/// (`NewL2BlockV2` + `SetCanonical`); the V1 path would reject the sibling with a
+/// wrong-parent-hash error. Near-wall-clock timestamps keep the blocks out of the
+/// historical-finalization fallback so the engine permits the reorg.
+#[tokio::test(flavor = "multi_thread")]
+async fn new_l2_block_v2_reorgs_onto_sibling_block() -> eyre::Result<()> {
+    reth_tracing::init_test_tracing();
+
+    let (mut nodes, _wallet) = TestNodeBuilder::new().build().await?;
+    let node = nodes.pop().unwrap();
+    let auth = node.auth_server_handle();
+    let client = auth.http_client();
+
+    let now = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_secs();
+
+    // Block 1 on genesis. Timestamps sit a few seconds in the past: recent enough to
+    // stay out of the historical-finalization fallback (so the engine permits the
+    // reorg) but not in the future (which header validation would reject).
+    let mut params = AssembleL2BlockParams::empty(1);
+    params.timestamp = Some(now - 6);
+    let block1: ExecutableL2Data = client.request("engine_assembleL2Block", (params,)).await?;
+    let block1_hash = block1.hash;
+    let _: MorphHeader = client.request("engine_newL2BlockV2", (block1,)).await?;
+
+    // Assemble two distinct siblings at height 2 on block 1. The head stays at block 1
+    // until we import, so both build on it. They differ only by timestamp, hence by hash.
+    let mut params_a = AssembleL2BlockParams::empty(2);
+    params_a.timestamp = Some(now - 4);
+    let block2a: ExecutableL2Data = client
+        .request("engine_assembleL2Block", (params_a,))
+        .await?;
+
+    let mut params_b = AssembleL2BlockParams::empty(2);
+    params_b.timestamp = Some(now - 2);
+    let block2b: ExecutableL2Data = client
+        .request("engine_assembleL2Block", (params_b,))
+        .await?;
+
+    assert_eq!(block2a.parent_hash, block1_hash, "2a must build on block 1");
+    assert_eq!(block2b.parent_hash, block1_hash, "2b must build on block 1");
+    assert_ne!(
+        block2a.hash, block2b.hash,
+        "siblings must have distinct hashes"
+    );
+    let block2b_hash = block2b.hash;
+
+    // Import sibling A → canonical head = 2a.
+    let _: MorphHeader = client.request("engine_newL2BlockV2", (block2a,)).await?;
+    let head = node
+        .inner
+        .provider
+        .sealed_header_by_number_or_tag(alloy_rpc_types_eth::BlockNumberOrTag::Latest)?
+        .expect("head after importing sibling A");
+    assert_eq!(head.number(), 2, "sibling A should be at height 2");
+
+    // Import sibling B (builds on block 1, not on 2a) → must reorg the head onto it.
+    let _: MorphHeader = client.request("engine_newL2BlockV2", (block2b,)).await?;
+    let head = node
+        .inner
+        .provider
+        .sealed_header_by_number_or_tag(alloy_rpc_types_eth::BlockNumberOrTag::Latest)?
+        .expect("head after importing sibling B");
+    assert_eq!(head.number(), 2, "head stays at height 2 after the reorg");
+    assert_eq!(head.hash(), block2b_hash, "head must reorg onto sibling B");
+
+    Ok(())
+}
+
+/// `engine_newSafeL2Block` with `parentHash` reorganizes onto a non-head parent.
+///
+/// Mirrors derivation's `deriveForce`: after the local chain already has block 2A
+/// (imported live), the derivation pipeline re-derives block 2 from L1 batch data and
+/// pins its parent to block 1. The safe path executes the block on that pinned parent
+/// and the engine reorgs the head onto the L1-canonical block 2B. Without `parentHash`
+/// the safe path requires `number == head + 1` and would reject this.
+#[tokio::test(flavor = "multi_thread")]
+async fn new_safe_l2_block_with_parent_hash_reorgs_onto_non_head_parent() -> eyre::Result<()> {
+    reth_tracing::init_test_tracing();
+
+    let (mut nodes, _wallet) = TestNodeBuilder::new().build().await?;
+    let node = nodes.pop().unwrap();
+    let auth = node.auth_server_handle();
+    let client = auth.http_client();
+
+    let now = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_secs();
+
+    // Block 1 on genesis (live import). Past-but-recent timestamps keep the blocks out
+    // of the historical-finalization fallback (so the reorg is permitted) without
+    // tripping the future-timestamp header check.
+    let mut p1 = AssembleL2BlockParams::empty(1);
+    p1.timestamp = Some(now - 6);
+    let block1: ExecutableL2Data = client.request("engine_assembleL2Block", (p1,)).await?;
+    let block1_hash = block1.hash;
+    let _: MorphHeader = client.request("engine_newL2BlockV2", (block1,)).await?;
+
+    // Block 2A on block 1 (live import) → canonical head = 2A.
+    let mut p2 = AssembleL2BlockParams::empty(2);
+    p2.timestamp = Some(now - 4);
+    let block2a: ExecutableL2Data = client.request("engine_assembleL2Block", (p2,)).await?;
+    let block2a_hash = block2a.hash;
+    let gas_limit = block2a.gas_limit;
+    let base_fee = block2a.base_fee_per_gas;
+    let _: MorphHeader = client.request("engine_newL2BlockV2", (block2a,)).await?;
+
+    // Re-derive block 2 from L1 data, parent pinned to block 1. A different timestamp
+    // gives it a distinct hash from 2A, forcing a real reorg.
+    let safe = SafeL2Data {
+        number: 2,
+        gas_limit,
+        base_fee_per_gas: base_fee,
+        timestamp: now - 2,
+        transactions: vec![],
+        parent_hash: Some(block1_hash),
+    };
+
+    let header: MorphHeader = client.request("engine_newSafeL2Block", (safe,)).await?;
+    assert_eq!(header.number(), 2, "returned safe header is at height 2");
+
+    let head = node
+        .inner
+        .provider
+        .sealed_header_by_number_or_tag(alloy_rpc_types_eth::BlockNumberOrTag::Latest)?
+        .expect("head after safe reorg");
+    assert_eq!(
+        head.number(),
+        2,
+        "head stays at height 2 after the safe reorg"
+    );
+    assert_ne!(
+        head.hash(),
+        block2a_hash,
+        "head must have reorged off block 2A"
+    );
+    assert_eq!(
+        head.hash(),
+        header.hash_slow(),
+        "canonical head must match the returned safe header"
+    );
+
+    Ok(())
+}
+
+/// `engine_assembleL2BlockV2` builds on an explicitly given parent hash (not the head).
+///
+/// V2 keys assembly on a parent hash rather than a block number, so the sequencer can
+/// build on any parent — including one that is no longer the canonical head. Here a
+/// second block 1' is assembled on genesis after block 1 is already canonical, then
+/// imported as a reorg. V2 uses a single params object, matching the V1
+/// `engine_assembleL2Block` style.
+#[tokio::test(flavor = "multi_thread")]
+async fn assemble_l2_block_v2_builds_on_explicit_parent() -> eyre::Result<()> {
+    reth_tracing::init_test_tracing();
+
+    let (mut nodes, _wallet) = TestNodeBuilder::new().build().await?;
+    let node = nodes.pop().unwrap();
+    let auth = node.auth_server_handle();
+    let client = auth.http_client();
+
+    let now = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_secs();
+
+    let genesis = node
+        .inner
+        .provider
+        .sealed_header_by_number_or_tag(alloy_rpc_types_eth::BlockNumberOrTag::Latest)?
+        .expect("genesis header");
+    let genesis_hash = genesis.hash();
+
+    // Assemble + import block 1 on genesis.
+    let block1: ExecutableL2Data = client
+        .request(
+            "engine_assembleL2BlockV2",
+            (serde_json::json!({
+                "parentHash": genesis_hash,
+                "timestamp": format!("{:#x}", now - 6),
+                "transactions": [],
+            }),),
+        )
+        .await?;
+    assert_eq!(block1.number, 1, "assembled block is at height 1");
+    assert_eq!(
+        block1.parent_hash, genesis_hash,
+        "assembled block must build on the given parent"
+    );
+    let block1_hash = block1.hash;
+    let _: MorphHeader = client.request("engine_newL2BlockV2", (block1,)).await?;
+
+    // Assemble a sibling 1' on genesis (still a valid parent though no longer the head),
+    // distinguished by timestamp, and import it as a reorg.
+    let block1_prime: ExecutableL2Data = client
+        .request(
+            "engine_assembleL2BlockV2",
+            (serde_json::json!({
+                "parentHash": genesis_hash,
+                "timestamp": format!("{:#x}", now - 3),
+                "transactions": [],
+            }),),
+        )
+        .await?;
+    assert_eq!(block1_prime.parent_hash, genesis_hash);
+    assert_eq!(block1_prime.number, 1);
+    assert_ne!(
+        block1_prime.hash, block1_hash,
+        "sibling must differ from block 1"
+    );
+    let block1_prime_hash = block1_prime.hash;
+
+    let _: MorphHeader = client
+        .request("engine_newL2BlockV2", (block1_prime,))
+        .await?;
+    let head = node
+        .inner
+        .provider
+        .sealed_header_by_number_or_tag(alloy_rpc_types_eth::BlockNumberOrTag::Latest)?
+        .expect("head after sibling import");
+    assert_eq!(head.number(), 1, "head stays at height 1 after reorg");
+    assert_eq!(
+        head.hash(),
+        block1_prime_hash,
+        "head must reorg onto the sibling assembled via V2"
+    );
+
+    Ok(())
+}
+
 /// `engine_validateL2Block` rejects a tampered block hash over authenticated RPC.
 #[tokio::test(flavor = "multi_thread")]
 async fn validate_l2_block_rejects_tampered_hash_over_rpc() -> eyre::Result<()> {
diff --git a/crates/payload/types/src/lib.rs b/crates/payload/types/src/lib.rs
index 4fea843..9ded5aa 100644
--- a/crates/payload/types/src/lib.rs
+++ b/crates/payload/types/src/lib.rs
@@ -37,7 +37,7 @@ pub use attributes::{
 };
 pub use built::MorphBuiltPayload;
 pub use executable_l2_data::ExecutableL2Data;
-pub use params::{AssembleL2BlockParams, GenericResponse};
+pub use params::{AssembleL2BlockParams, AssembleL2BlockV2Params, GenericResponse};
 pub use safe_l2_data::SafeL2Data;
 
 // =============================================================================
diff --git a/crates/payload/types/src/params.rs b/crates/payload/types/src/params.rs
index 77343b4..c70d95b 100644
--- a/crates/payload/types/src/params.rs
+++ b/crates/payload/types/src/params.rs
@@ -1,6 +1,6 @@
 //! Request/response types for L2 Engine API methods.
 
-use alloy_primitives::Bytes;
+use alloy_primitives::{B256, Bytes};
 
 /// Parameters for engine_assembleL2Block.
 ///
@@ -48,6 +48,53 @@ impl AssembleL2BlockParams {
     }
 }
 
+/// Parameters for `engine_assembleL2BlockV2`.
+///
+/// V2 mirrors V1's single-struct JSON-RPC parameter style while selecting the
+/// parent by hash instead of by block number. Transactions use the normal
+/// Ethereum JSON-RPC bytes encoding (`0x...` hex strings).
+#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct AssembleL2BlockV2Params {
+    /// Parent block hash to build on.
+    pub parent_hash: B256,
+
+    /// Transactions to include in the block.
+    /// These are RLP-encoded transaction bytes.
+    #[serde(default)]
+    pub transactions: Vec<Bytes>,
+
+    /// Optional block timestamp.
+    ///
+    /// If not provided, builder can choose a local current timestamp.
+    #[serde(
+        default,
+        skip_serializing_if = "Option::is_none",
+        with = "alloy_serde::quantity::opt"
+    )]
+    pub timestamp: Option<u64>,
+}
+
+impl AssembleL2BlockV2Params {
+    /// Create a new [`AssembleL2BlockV2Params`].
+    pub fn new(parent_hash: B256, transactions: Vec<Bytes>) -> Self {
+        Self {
+            parent_hash,
+            transactions,
+            timestamp: None,
+        }
+    }
+
+    /// Create params for an empty block.
+    pub fn empty(parent_hash: B256) -> Self {
+        Self {
+            parent_hash,
+            transactions: Vec::new(),
+            timestamp: None,
+        }
+    }
+}
+
 /// Generic success/failure response for L2 Engine API methods.
 ///
 /// This is used by methods like engine_validateL2Block that return
@@ -122,6 +169,72 @@ mod tests {
         assert_eq!(params, decoded);
     }
 
+    #[test]
+    fn test_assemble_v2_params_new() {
+        let parent_hash = B256::repeat_byte(0xab);
+        let params = AssembleL2BlockV2Params::new(parent_hash, vec![Bytes::from(vec![0x01])]);
+        assert_eq!(params.parent_hash, parent_hash);
+        assert_eq!(params.transactions.len(), 1);
+        assert!(params.timestamp.is_none());
+    }
+
+    #[test]
+    fn test_assemble_v2_params_empty() {
+        let parent_hash = B256::repeat_byte(0xcd);
+        let params = AssembleL2BlockV2Params::empty(parent_hash);
+        assert_eq!(params.parent_hash, parent_hash);
+        assert!(params.transactions.is_empty());
+        assert!(params.timestamp.is_none());
+    }
+
+    #[test]
+    fn test_assemble_v2_params_serde() {
+        let parent_hash = B256::repeat_byte(0xab);
+        let params = AssembleL2BlockV2Params {
+            parent_hash,
+            transactions: vec![Bytes::from(vec![0xde, 0xad])],
+            timestamp: Some(0x6553f100),
+        };
+
+        let json = serde_json::to_string(&params).expect("serialize");
+        assert!(json.contains("parentHash"));
+        assert!(json.contains("\"0xdead\""));
+
+        let decoded: AssembleL2BlockV2Params = serde_json::from_str(&json).expect("deserialize");
+        assert_eq!(params, decoded);
+    }
+
+    #[test]
+    fn test_assemble_v2_params_serde_with_hex_timestamp_and_txs() {
+        let json = r#"{
+            "parentHash": "0xabababababababababababababababababababababababababababababababab",
+            "transactions": ["0xdead"],
+            "timestamp": "0x6553f100"
+        }"#;
+
+        let params: AssembleL2BlockV2Params = serde_json::from_str(json).expect("deserialize");
+        assert_eq!(params.parent_hash, B256::repeat_byte(0xab));
+        assert_eq!(params.transactions, vec![Bytes::from(vec![0xde, 0xad])]);
+        assert_eq!(params.timestamp, Some(0x6553f100));
+    }
+
+    #[test]
+    fn test_assemble_v2_params_serde_explicit_null_timestamp() {
+        // geth's gencodec MarshalJSON always emits the `timestamp` field, writing
+        // `null` (not omitting it) when the sequencer passes no timestamp. The reth
+        // server must decode that production payload to `None` rather than erroring.
+        let json = r#"{
+            "parentHash": "0xabababababababababababababababababababababababababababababababab",
+            "transactions": ["0xdead"],
+            "timestamp": null
+        }"#;
+
+        let params: AssembleL2BlockV2Params = serde_json::from_str(json).expect("deserialize");
+        assert_eq!(params.parent_hash, B256::repeat_byte(0xab));
+        assert_eq!(params.transactions, vec![Bytes::from(vec![0xde, 0xad])]);
+        assert!(params.timestamp.is_none());
+    }
+
     #[test]
     fn test_generic_response_serde() {
         let response = GenericResponse::success();
diff --git a/crates/payload/types/src/safe_l2_data.rs b/crates/payload/types/src/safe_l2_data.rs
index a67b553..3344887 100644
--- a/crates/payload/types/src/safe_l2_data.rs
+++ b/crates/payload/types/src/safe_l2_data.rs
@@ -2,7 +2,7 @@
 //!
 //! This type is used for NewSafeL2Block in the derivation pipeline.
 
-use alloy_primitives::Bytes;
+use alloy_primitives::{B256, Bytes};
 
 /// Safe L2 block data, used for NewSafeL2Block (derivation).
 ///
@@ -38,6 +38,16 @@ pub struct SafeL2Data {
     /// RLP-encoded transactions.
     #[serde(default)]
     pub transactions: Vec<Bytes>,
+
+    /// Optional parent hash for the derivation reorg path.
+    ///
+    /// When set, the block is executed on top of this parent (looked up by hash)
+    /// and the engine reorganizes the canonical chain onto it via forkchoice
+    /// update — used by `derivation.deriveForce` to apply the L1-canonical chain
+    /// on top of a non-head parent. When `None`, the legacy "extend the current
+    /// head" semantics apply. Mirrors go-ethereum's `SafeL2Data.ParentHash`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub parent_hash: Option<B256>,
 }
 
 impl SafeL2Data {
@@ -78,6 +88,7 @@ mod tests {
             base_fee_per_gas: Some(1_000_000_000),
             timestamp: 1234567890,
             transactions: vec![Bytes::from(vec![0x01, 0x02])],
+            parent_hash: None,
         };
 
         let json = serde_json::to_string(&data).expect("serialize");
@@ -94,6 +105,7 @@ mod tests {
             base_fee_per_gas: None,
             timestamp: 1234567890,
             transactions: vec![],
+            parent_hash: None,
         };
 
         let json = serde_json::to_string(&data).expect("serialize");
@@ -140,6 +152,62 @@ mod tests {
         assert_eq!(data.transaction_count(), 1);
     }
 
+    #[test]
+    fn test_serde_parent_hash_roundtrip() {
+        let hash = alloy_primitives::B256::repeat_byte(0xab);
+        let data = SafeL2Data {
+            number: 7,
+            gas_limit: 30_000_000,
+            base_fee_per_gas: None,
+            timestamp: 100,
+            transactions: vec![],
+            parent_hash: Some(hash),
+        };
+
+        let json = serde_json::to_string(&data).expect("serialize");
+        assert!(
+            json.contains("parentHash"),
+            "parentHash must be serialized when set: {json}"
+        );
+
+        let decoded: SafeL2Data = serde_json::from_str(&json).expect("deserialize");
+        assert_eq!(decoded.parent_hash, Some(hash));
+    }
+
+    #[test]
+    fn test_serde_without_parent_hash_omits_field() {
+        let data = SafeL2Data {
+            parent_hash: None,
+            ..Default::default()
+        };
+
+        let json = serde_json::to_string(&data).expect("serialize");
+        assert!(
+            !json.contains("parentHash"),
+            "parentHash must be omitted when None: {json}"
+        );
+
+        let decoded: SafeL2Data = serde_json::from_str(&json).expect("deserialize");
+        assert_eq!(decoded.parent_hash, None);
+    }
+
+    #[test]
+    fn test_serde_parent_hash_from_camel_case_json() {
+        let json = r#"{
+            "number": "0x64",
+            "gasLimit": "0x1c9c380",
+            "timestamp": "0x499602d2",
+            "transactions": [],
+            "parentHash": "0xabababababababababababababababababababababababababababababababab"
+        }"#;
+
+        let data: SafeL2Data = serde_json::from_str(json).expect("deserialize");
+        assert_eq!(
+            data.parent_hash,
+            Some(alloy_primitives::B256::repeat_byte(0xab))
+        );
+    }
+
     #[test]
     fn test_clone_and_equality() {
         let data = SafeL2Data {
@@ -148,6 +216,7 @@ mod tests {
             base_fee_per_gas: Some(100),
             timestamp: 999,
             transactions: vec![Bytes::from(vec![0x01, 0x02])],
+            parent_hash: None,
         };
 
         let cloned = data.clone();