diff --git a/Cargo.lock b/Cargo.lock
index ab4fcc25744..2c8fa2679b3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7628,6 +7628,7 @@ dependencies = [
  "reth-revm",
  "reth-storage-api",
  "reth-tasks",
+ "reth-trie-parallel",
  "serde",
  "tokio",
  "tracing",
@@ -9503,6 +9504,7 @@ dependencies = [
  "reth-payload-builder-primitives",
  "reth-payload-primitives",
  "reth-primitives-traits",
+ "reth-trie-parallel",
  "tokio",
  "tokio-stream",
  "tracing",
@@ -10559,6 +10561,8 @@ dependencies = [
 name = "reth-trie-parallel"
 version = "1.11.3"
 dependencies = [
+ "alloy-eip7928",
+ "alloy-evm",
  "alloy-primitives",
  "alloy-rlp",
  "crossbeam-channel",
@@ -10578,6 +10582,7 @@ dependencies = [
  "reth-trie",
  "reth-trie-db",
  "reth-trie-sparse",
+ "revm-state",
  "thiserror 2.0.18",
  "tokio",
  "tracing",
diff --git a/crates/engine/primitives/src/config.rs b/crates/engine/primitives/src/config.rs
index a53f5f67899..078b08d9e3f 100644
--- a/crates/engine/primitives/src/config.rs
+++ b/crates/engine/primitives/src/config.rs
@@ -153,6 +153,8 @@ pub struct TreeConfig {
     state_root_task_timeout: Option<Duration>,
     /// Whether to share execution cache with the payload builder.
     share_execution_cache_with_payload_builder: bool,
+    /// Whether to share sparse trie with the payload builder.
+    share_sparse_trie_with_payload_builder: bool,
     /// Maximum random jitter applied before each proof computation (trie-debug only).
     /// When set, each proof worker sleeps for a random duration up to this value
     /// before starting a proof calculation.
@@ -189,6 +191,7 @@ impl Default for TreeConfig {
             disable_sparse_trie_cache_pruning: false,
             state_root_task_timeout: Some(DEFAULT_STATE_ROOT_TASK_TIMEOUT),
             share_execution_cache_with_payload_builder: false,
+            share_sparse_trie_with_payload_builder: false,
             #[cfg(feature = "trie-debug")]
             proof_jitter: None,
         }
@@ -224,6 +227,7 @@ impl TreeConfig {
         slow_block_threshold: Option<Duration>,
         state_root_task_timeout: Option<Duration>,
         share_execution_cache_with_payload_builder: bool,
+        share_sparse_trie_with_payload_builder: bool,
     ) -> Self {
         Self {
             persistence_threshold,
@@ -252,6 +256,7 @@ impl TreeConfig {
             disable_sparse_trie_cache_pruning: false,
             state_root_task_timeout,
             share_execution_cache_with_payload_builder,
+            share_sparse_trie_with_payload_builder,
             #[cfg(feature = "trie-debug")]
             proof_jitter: None,
         }
@@ -569,6 +574,11 @@ impl TreeConfig {
         self.share_execution_cache_with_payload_builder
     }
 
+    /// Returns whether to share sparse trie with the payload builder.
+    pub const fn share_sparse_trie_with_payload_builder(&self) -> bool {
+        self.share_sparse_trie_with_payload_builder
+    }
+
     /// Setter for whether to share execution cache with the payload builder.
     pub const fn with_share_execution_cache_with_payload_builder(
         mut self,
@@ -579,6 +589,15 @@ impl TreeConfig {
         self
     }
 
+    /// Setter for whether to share sparse trie with the payload builder.
+    pub const fn with_share_sparse_trie_with_payload_builder(
+        mut self,
+        share_sparse_trie_with_payload_builder: bool,
+    ) -> Self {
+        self.share_sparse_trie_with_payload_builder = share_sparse_trie_with_payload_builder;
+        self
+    }
+
     /// Returns the proof jitter duration, if configured (trie-debug only).
     #[cfg(feature = "trie-debug")]
     pub const fn proof_jitter(&self) -> Option<Duration> {
diff --git a/crates/engine/tree/src/tree/mod.rs b/crates/engine/tree/src/tree/mod.rs
index 675ed263428..e65f0e70c49 100644
--- a/crates/engine/tree/src/tree/mod.rs
+++ b/crates/engine/tree/src/tree/mod.rs
@@ -3101,12 +3101,23 @@ where
             None
         };
 
+        let trie_handle = if self.config.share_sparse_trie_with_payload_builder() {
+            self.payload_validator.sparse_trie_handle_for(
+                state.head_block_hash,
+                head.state_root(),
+                &self.state,
+            )
+        } else {
+            None
+        };
+
         // send the payload to the builder and return the receiver for the pending payload
         // id, initiating payload job is handled asynchronously
         let pending_payload_id = self.payload_builder.send_new_payload(BuildNewPayload {
             parent_hash: state.head_block_hash,
             attributes,
             cache,
+            trie_handle,
         });
 
         // Client software MUST respond to this method call in the following way:
diff --git a/crates/engine/tree/src/tree/payload_processor/mod.rs b/crates/engine/tree/src/tree/payload_processor/mod.rs
index 9ddd0260c47..e93c0673eac 100644
--- a/crates/engine/tree/src/tree/payload_processor/mod.rs
+++ b/crates/engine/tree/src/tree/payload_processor/mod.rs
@@ -2,17 +2,13 @@
 
 use super::precompile_cache::PrecompileCacheMap;
 use crate::tree::{
-    payload_processor::{
-        prewarm::{PrewarmCacheTask, PrewarmContext, PrewarmMode, PrewarmTaskEvent},
-        sparse_trie::StateRootComputeOutcome,
-    },
+    payload_processor::prewarm::{PrewarmCacheTask, PrewarmContext, PrewarmMode, PrewarmTaskEvent},
     sparse_trie::SparseTrieCacheTask,
     CacheWaitDurations, CachedStateMetrics, ExecutionCache, PayloadExecutionCache, SavedCache,
     StateProviderBuilder, TreeConfig, WaitForCaches,
 };
 use alloy_eip7928::BlockAccessList;
 use alloy_eips::{eip1898::BlockWithParent, eip4895::Withdrawal};
-use alloy_evm::block::StateChangeSource;
 use alloy_primitives::B256;
 use crossbeam_channel::{Receiver as CrossbeamReceiver, Sender as CrossbeamSender};
 use multiproof::*;
@@ -28,7 +24,7 @@ use reth_primitives_traits::{FastInstant as Instant, NodePrimitives};
 use reth_provider::{
     BlockExecutionOutput, BlockReader, DatabaseProviderROFactory, StateProviderFactory, StateReader,
 };
-use reth_revm::{db::BundleState, state::EvmState};
+use reth_revm::db::BundleState;
 use reth_tasks::{utils::increase_thread_priority, ForEachOrdered, Runtime};
 use reth_trie::{hashed_cursor::HashedCursorFactory, trie_cursor::TrieCursorFactory};
 use reth_trie_parallel::{
@@ -271,12 +267,18 @@ where
 
         let span = Span::current();
 
-        let state_root_handle = self.spawn_state_root(multiproof_provider_factory, &env, config);
+        let halve_workers = env.transaction_count <= Self::SMALL_BLOCK_PROOF_WORKER_TX_THRESHOLD;
+        let state_root_handle = self.spawn_state_root(
+            multiproof_provider_factory,
+            env.parent_state_root,
+            halve_workers,
+            config,
+        );
         let prewarm_handle = self.spawn_caching_with(
             env,
             prewarm_rx,
             provider_builder,
-            Some(state_root_handle.to_multi_proof.clone()),
+            Some(state_root_handle.updates_tx().clone()),
             bal,
         );
 
@@ -322,11 +324,15 @@ where
     ///   state root.
     ///
     /// The state hook **must** be dropped after execution to signal the end of state updates.
+    ///
+    /// When `halve_workers` is true, the proof worker pool is halved (for small blocks where
+    /// fewer transactions produce fewer state changes and most workers would be idle).
     #[instrument(level = "debug", target = "engine::tree::payload_processor", skip_all)]
     pub fn spawn_state_root<F>(
-        &mut self,
+        &self,
         multiproof_provider_factory: F,
-        env: &ExecutionEnv<Evm>,
+        parent_state_root: B256,
+        halve_workers: bool,
         config: &TreeConfig,
     ) -> StateRootHandle
     where
@@ -336,12 +342,11 @@ where
             + Sync
             + 'static,
     {
-        let (to_multi_proof, from_multi_proof) = crossbeam_channel::unbounded();
+        let (updates_tx, from_multi_proof) = crossbeam_channel::unbounded();
 
         let task_ctx = ProofTaskCtx::new(multiproof_provider_factory);
         #[cfg(feature = "trie-debug")]
         let task_ctx = task_ctx.with_proof_jitter(config.proof_jitter());
-        let halve_workers = env.transaction_count <= Self::SMALL_BLOCK_PROOF_WORKER_TX_THRESHOLD;
         let proof_handle = ProofWorkerHandle::new(&self.executor, task_ctx, halve_workers);
 
         let (state_root_tx, state_root_rx) = channel();
@@ -350,11 +355,11 @@ where
             proof_handle,
             state_root_tx,
             from_multi_proof,
-            env.parent_state_root,
+            parent_state_root,
             config.multiproof_chunk_size(),
         );
 
-        StateRootHandle::new(to_multi_proof, state_root_rx)
+        StateRootHandle::new(parent_state_root, updates_tx, state_root_rx)
     }
 
     /// Transaction count threshold below which proof workers are halved, since fewer transactions
@@ -465,7 +470,7 @@ where
         env: ExecutionEnv<Evm>,
         transactions: mpsc::Receiver<(usize, impl ExecutableTxFor<Evm> + Clone + Send + 'static)>,
         provider_builder: StateProviderBuilder<N, P>,
-        to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
+        to_multi_proof: Option<CrossbeamSender<StateRootMessage>>,
         bal: Option<Arc<BlockAccessList>>,
     ) -> CacheTaskHandle<N::Receipt>
     where
@@ -542,7 +547,7 @@ where
         &self,
         proof_worker_handle: ProofWorkerHandle,
         state_root_tx: mpsc::Sender<Result<StateRootComputeOutcome, ParallelStateRootError>>,
-        from_multi_proof: CrossbeamReceiver<MultiProofMessage>,
+        from_multi_proof: CrossbeamReceiver<StateRootMessage>,
         parent_state_root: B256,
         chunk_size: usize,
     ) {
@@ -742,67 +747,6 @@ fn convert_serial<RawTx, Tx, TxEnv, InnerTx, Recovered, Err, C>(
     }
 }
 
-/// Handle to a background state root computation task.
-///
-/// Unlike [`PayloadHandle`], this does not include transaction iteration or cache prewarming.
-/// It only provides access to the state root computation via [`Self::state_hook`] and
-/// [`Self::state_root`].
-///
-/// Created by [`PayloadProcessor::spawn_state_root`].
-#[derive(Debug)]
-pub struct StateRootHandle {
-    /// Channel for evm state updates to the multiproof pipeline.
-    to_multi_proof: CrossbeamSender<MultiProofMessage>,
-    /// Receiver for the computed state root.
-    state_root_rx: Option<mpsc::Receiver<Result<StateRootComputeOutcome, ParallelStateRootError>>>,
-}
-
-impl StateRootHandle {
-    /// Creates a new state root handle.
-    pub const fn new(
-        to_multi_proof: CrossbeamSender<MultiProofMessage>,
-        state_root_rx: mpsc::Receiver<Result<StateRootComputeOutcome, ParallelStateRootError>>,
-    ) -> Self {
-        Self { to_multi_proof, state_root_rx: Some(state_root_rx) }
-    }
-
-    /// Returns a state hook that streams state updates to the background state root task.
-    ///
-    /// The hook must be dropped after execution completes to signal the end of state updates.
-    pub fn state_hook(&self) -> impl OnStateHook {
-        let to_multi_proof = StateHookSender::new(self.to_multi_proof.clone());
-
-        move |source: StateChangeSource, state: &EvmState| {
-            let _ =
-                to_multi_proof.send(MultiProofMessage::StateUpdate(source.into(), state.clone()));
-        }
-    }
-
-    /// Awaits the state root computation result.
-    ///
-    /// # Panics
-    ///
-    /// If called more than once.
-    pub fn state_root(&mut self) -> Result<StateRootComputeOutcome, ParallelStateRootError> {
-        self.state_root_rx
-            .take()
-            .expect("state_root already taken")
-            .recv()
-            .map_err(|_| ParallelStateRootError::Other("sparse trie task dropped".to_string()))?
-    }
-
-    /// Takes the state root receiver for use with custom waiting logic (e.g., timeouts).
-    ///
-    /// # Panics
-    ///
-    /// If called more than once.
-    pub const fn take_state_root_rx(
-        &mut self,
-    ) -> mpsc::Receiver<Result<StateRootComputeOutcome, ParallelStateRootError>> {
-        self.state_root_rx.take().expect("state_root already taken")
-    }
-}
-
 /// Handle to all the spawned tasks.
 ///
 /// Generic over `R` (receipt type) to allow sharing `Arc<ExecutionOutcome<R>>` with the
diff --git a/crates/engine/tree/src/tree/payload_processor/multiproof.rs b/crates/engine/tree/src/tree/payload_processor/multiproof.rs
index 5e2f04baeb7..0f4046d1d7a 100644
--- a/crates/engine/tree/src/tree/payload_processor/multiproof.rs
+++ b/crates/engine/tree/src/tree/payload_processor/multiproof.rs
@@ -1,120 +1,17 @@
 //! Multiproof task related functionality.
 
-use alloy_evm::block::StateChangeSource;
-use alloy_primitives::{keccak256, B256};
-use crossbeam_channel::Sender as CrossbeamSender;
-use derive_more::derive::Deref;
 use metrics::{Gauge, Histogram};
 use reth_metrics::Metrics;
-use reth_revm::state::EvmState;
-use reth_trie::{HashedPostState, HashedStorage};
-use reth_trie_common::MultiProofTargetsV2;
-use std::sync::Arc;
-use tracing::trace;
 
-/// Source of state changes, either from EVM execution or from a Block Access List.
-#[derive(Clone, Copy)]
-pub enum Source {
-    /// State changes from EVM execution.
-    Evm(StateChangeSource),
-    /// State changes from Block Access List (EIP-7928).
-    BlockAccessList,
-}
-
-impl std::fmt::Debug for Source {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self {
-            Self::Evm(source) => source.fmt(f),
-            Self::BlockAccessList => f.write_str("BlockAccessList"),
-        }
-    }
-}
-
-impl From<StateChangeSource> for Source {
-    fn from(source: StateChangeSource) -> Self {
-        Self::Evm(source)
-    }
-}
+pub use reth_trie_parallel::state_root_task::{
+    evm_state_to_hashed_post_state, Source, StateHookSender, StateRootComputeOutcome,
+    StateRootHandle, StateRootMessage,
+};
 
 /// The default max targets, for limiting the number of account and storage proof targets to be
 /// fetched by a single worker. If exceeded, chunking is forced regardless of worker availability.
 pub(crate) const DEFAULT_MAX_TARGETS_FOR_CHUNKING: usize = 300;
 
-/// Messages used internally by the multi proof task.
-#[derive(Debug)]
-pub enum MultiProofMessage {
-    /// Prefetch proof targets
-    PrefetchProofs(MultiProofTargetsV2),
-    /// New state update from transaction execution with its source
-    StateUpdate(Source, EvmState),
-    /// Pre-hashed state update from BAL conversion that can be applied directly without proofs.
-    HashedStateUpdate(HashedPostState),
-    /// Block Access List (EIP-7928; BAL) containing complete state changes for the block.
-    ///
-    /// When received, the task generates a single state update from the BAL and processes it.
-    /// No further messages are expected after receiving this variant.
-    BlockAccessList(Arc<alloy_eip7928::BlockAccessList>),
-    /// Signals state update stream end.
-    ///
-    /// This is triggered by block execution, indicating that no additional state updates are
-    /// expected.
-    FinishedStateUpdates,
-}
-
-/// A wrapper for the sender that signals completion when dropped.
-///
-/// This type is intended to be used in combination with the evm executor statehook.
-/// This should trigger once the block has been executed (after) the last state update has been
-/// sent. This triggers the exit condition of the multi proof task.
-#[derive(Deref, Debug)]
-pub struct StateHookSender(CrossbeamSender<MultiProofMessage>);
-
-impl StateHookSender {
-    /// Creates a new [`StateHookSender`] wrapping the given channel sender.
-    pub const fn new(inner: CrossbeamSender<MultiProofMessage>) -> Self {
-        Self(inner)
-    }
-}
-
-impl Drop for StateHookSender {
-    fn drop(&mut self) {
-        // Send completion signal when the sender is dropped
-        let _ = self.0.send(MultiProofMessage::FinishedStateUpdates);
-    }
-}
-
-pub(crate) fn evm_state_to_hashed_post_state(update: EvmState) -> HashedPostState {
-    let mut hashed_state = HashedPostState::with_capacity(update.len());
-
-    for (address, account) in update {
-        if account.is_touched() {
-            let hashed_address = keccak256(address);
-            trace!(target: "engine::tree::payload_processor::multiproof", ?address, ?hashed_address, "Adding account to state update");
-
-            let destroyed = account.is_selfdestructed();
-            let info = if destroyed { None } else { Some(account.info.into()) };
-            hashed_state.accounts.insert(hashed_address, info);
-
-            let mut changed_storage_iter = account
-                .storage
-                .into_iter()
-                .filter(|(_slot, value)| value.is_changed())
-                .map(|(slot, value)| (keccak256(B256::from(slot)), value.present_value))
-                .peekable();
-
-            if destroyed {
-                hashed_state.storages.insert(hashed_address, HashedStorage::new(true));
-            } else if changed_storage_iter.peek().is_some() {
-                hashed_state
-                    .storages
-                    .insert(hashed_address, HashedStorage::from_iter(false, changed_storage_iter));
-            }
-        }
-    }
-
-    hashed_state
-}
-
 #[derive(Metrics, Clone)]
 #[metrics(scope = "tree.root")]
 pub(crate) struct MultiProofTaskMetrics {
diff --git a/crates/engine/tree/src/tree/payload_processor/prewarm.rs b/crates/engine/tree/src/tree/payload_processor/prewarm.rs
index 886f18b1cc1..19d4c25c3b2 100644
--- a/crates/engine/tree/src/tree/payload_processor/prewarm.rs
+++ b/crates/engine/tree/src/tree/payload_processor/prewarm.rs
@@ -12,7 +12,7 @@
 //! 3. When actual block execution happens, it benefits from the warmed cache
 
 use crate::tree::{
-    payload_processor::{bal, multiproof::MultiProofMessage},
+    payload_processor::{bal, multiproof::StateRootMessage},
     precompile_cache::{CachedPrecompile, PrecompileCacheMap},
     CachedStateProvider, ExecutionEnv, PayloadExecutionCache, SavedCache, StateProviderBuilder,
 };
@@ -69,7 +69,7 @@ where
     /// Context provided to execution tasks
     ctx: PrewarmContext<N, P, Evm>,
     /// Sender to emit evm state outcome messages, if any.
-    to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
+    to_multi_proof: Option<CrossbeamSender<StateRootMessage>>,
     /// Receiver for events produced by tx execution
     actions_rx: Receiver<PrewarmTaskEvent<N::Receipt>>,
     /// Parent span for tracing
@@ -87,7 +87,7 @@ where
         executor: Runtime,
         execution_cache: PayloadExecutionCache,
         ctx: PrewarmContext<N, P, Evm>,
-        to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
+        to_multi_proof: Option<CrossbeamSender<StateRootMessage>>,
     ) -> (Self, Sender<PrewarmTaskEvent<N::Receipt>>) {
         let (actions_tx, actions_rx) = channel();
 
@@ -121,7 +121,7 @@ where
         &self,
         pending: mpsc::Receiver<(usize, Tx)>,
         actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
-        to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
+        to_multi_proof: Option<CrossbeamSender<StateRootMessage>>,
     ) where
         Tx: ExecutableTxFor<Evm> + Send + 'static,
     {
@@ -181,7 +181,7 @@ where
                     !withdrawals.is_empty()
                 {
                     let targets = multiproof_targets_from_withdrawals(withdrawals);
-                    let _ = to_multi_proof.send(MultiProofMessage::PrefetchProofs(targets));
+                    let _ = to_multi_proof.send(StateRootMessage::PrefetchProofs(targets));
                 }
             });
 
@@ -201,7 +201,7 @@ where
         ctx: &PrewarmContext<N, P, Evm>,
         index: usize,
         tx: Tx,
-        to_multi_proof: Option<&CrossbeamSender<MultiProofMessage>>,
+        to_multi_proof: Option<&CrossbeamSender<StateRootMessage>>,
     ) where
         Tx: ExecutableTxFor<Evm>,
     {
@@ -248,7 +248,7 @@ where
                 let (targets, storage_targets) = multiproof_targets_from_state(res.state);
                 ctx.metrics.prefetch_storage_targets.record(storage_targets as f64);
                 if let Some(to_multi_proof) = to_multi_proof {
-                    let _ = to_multi_proof.send(MultiProofMessage::PrefetchProofs(targets));
+                    let _ = to_multi_proof.send(StateRootMessage::PrefetchProofs(targets));
                 }
             }
 
@@ -402,8 +402,8 @@ where
                     storages = hashed_state.storages.len(),
                     "Converted BAL to hashed post state"
                 );
-                let _ = to_multi_proof.send(MultiProofMessage::HashedStateUpdate(hashed_state));
-                let _ = to_multi_proof.send(MultiProofMessage::FinishedStateUpdates);
+                let _ = to_multi_proof.send(StateRootMessage::HashedStateUpdate(hashed_state));
+                let _ = to_multi_proof.send(StateRootMessage::FinishedStateUpdates);
             }
             Err(err) => {
                 warn!(
diff --git a/crates/engine/tree/src/tree/payload_processor/sparse_trie.rs b/crates/engine/tree/src/tree/payload_processor/sparse_trie.rs
index a3103787fa4..27f38f53f72 100644
--- a/crates/engine/tree/src/tree/payload_processor/sparse_trie.rs
+++ b/crates/engine/tree/src/tree/payload_processor/sparse_trie.rs
@@ -4,8 +4,8 @@ use std::sync::Arc;
 
 use crate::tree::{
     multiproof::{
-        dispatch_with_chunking, evm_state_to_hashed_post_state, MultiProofMessage,
-        DEFAULT_MAX_TARGETS_FOR_CHUNKING,
+        dispatch_with_chunking, evm_state_to_hashed_post_state, StateRootComputeOutcome,
+        StateRootMessage, DEFAULT_MAX_TARGETS_FOR_CHUNKING,
     },
     payload_processor::multiproof::MultiProofTaskMetrics,
 };
@@ -26,8 +26,6 @@ use reth_trie_parallel::{
     },
     root::ParallelStateRootError,
 };
-#[cfg(feature = "trie-debug")]
-use reth_trie_sparse::debug_recorder::TrieDebugRecorder;
 use reth_trie_sparse::{
     errors::SparseTrieResult, ConfigurableSparseTrie, DeferredDrops, LeafUpdate,
     RevealableSparseTrie, SparseStateTrie, SparseTrie,
@@ -118,7 +116,7 @@ where
     /// Creates a new sparse trie, pre-populating with an existing [`SparseStateTrie`].
     pub(super) fn new_with_trie(
         executor: &Runtime,
-        updates: CrossbeamReceiver<MultiProofMessage>,
+        updates: CrossbeamReceiver<StateRootMessage>,
         proof_worker_handle: ProofWorkerHandle,
         metrics: MultiProofTaskMetrics,
         trie: SparseStateTrie<A, S>,
@@ -164,7 +162,7 @@ where
     /// Runs the hashing task that drains updates from the channel and converts them to
     /// `HashedPostState` in parallel.
     fn run_hashing_task(
-        updates: CrossbeamReceiver<MultiProofMessage>,
+        updates: CrossbeamReceiver<StateRootMessage>,
         hashed_state_tx: CrossbeamSender<SparseTrieTaskMessage>,
         metrics: MultiProofTaskMetrics,
     ) {
@@ -175,22 +173,22 @@ where
             total_idle_time += idle_start.elapsed();
 
             let msg = match message {
-                MultiProofMessage::PrefetchProofs(targets) => {
+                StateRootMessage::PrefetchProofs(targets) => {
                     SparseTrieTaskMessage::PrefetchProofs(targets)
                 }
-                MultiProofMessage::StateUpdate(_, state) => {
+                StateRootMessage::StateUpdate(_, state) => {
                     let _span = debug_span!(target: "engine::tree::payload_processor::sparse_trie", "hashing_state_update", n = state.len()).entered();
                     let hashed = evm_state_to_hashed_post_state(state);
                     SparseTrieTaskMessage::HashedState(hashed)
                 }
-                MultiProofMessage::FinishedStateUpdates => {
+                StateRootMessage::FinishedStateUpdates => {
                     SparseTrieTaskMessage::FinishedStateUpdates
                 }
-                MultiProofMessage::BlockAccessList(_) => {
+                StateRootMessage::BlockAccessList(_) => {
                     idle_start = Instant::now();
                     continue;
                 }
-                MultiProofMessage::HashedStateUpdate(state) => {
+                StateRootMessage::HashedStateUpdate(state) => {
                     SparseTrieTaskMessage::HashedState(state)
                 }
             };
@@ -863,20 +861,6 @@ enum SparseTrieTaskMessage {
     FinishedStateUpdates,
 }
 
-/// Outcome of the state root computation, including the state root itself with
-/// the trie updates.
-#[derive(Debug, Clone)]
-pub struct StateRootComputeOutcome {
-    /// The state root.
-    pub state_root: B256,
-    /// The trie updates.
-    pub trie_updates: Arc<TrieUpdates>,
-    /// Debug recorders taken from the sparse tries, keyed by `None` for account trie
-    /// and `Some(address)` for storage tries.
-    #[cfg(feature = "trie-debug")]
-    pub debug_recorders: Vec<(Option<B256>, TrieDebugRecorder)>,
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -911,8 +895,8 @@ mod tests {
             );
         });
 
-        updates_tx.send(MultiProofMessage::HashedStateUpdate(hashed_state)).unwrap();
-        updates_tx.send(MultiProofMessage::FinishedStateUpdates).unwrap();
+        updates_tx.send(StateRootMessage::HashedStateUpdate(hashed_state)).unwrap();
+        updates_tx.send(StateRootMessage::FinishedStateUpdates).unwrap();
         drop(updates_tx);
 
         let SparseTrieTaskMessage::HashedState(received) = hashed_state_rx.recv().unwrap() else {
diff --git a/crates/engine/tree/src/tree/payload_validator.rs b/crates/engine/tree/src/tree/payload_validator.rs
index 9ece726513b..b4edc74630d 100644
--- a/crates/engine/tree/src/tree/payload_validator.rs
+++ b/crates/engine/tree/src/tree/payload_validator.rs
@@ -41,9 +41,9 @@
 use crate::tree::{
     error::{InsertBlockError, InsertBlockErrorKind, InsertPayloadError},
     instrumented_state::{InstrumentedStateProvider, StateProviderStats},
+    multiproof::{StateRootComputeOutcome, StateRootHandle},
     payload_processor::PayloadProcessor,
     precompile_cache::{CachedPrecompile, CachedPrecompileMetrics, PrecompileCacheMap},
-    sparse_trie::StateRootComputeOutcome,
     CacheWaitDurations, CachedStateProvider, EngineApiMetrics, EngineApiTreeState, ExecutionEnv,
     PayloadHandle, StateProviderBuilder, StateProviderDatabase, TreeConfig, WaitForCaches,
 };
@@ -1944,6 +1944,14 @@ pub trait EngineValidator<
 
     /// Returns [`SavedCache`] for the given block hash.
     fn cache_for(&self, _block_hash: B256) -> Option<SavedCache>;
+
+    /// Spawns a sparse trie pipeline and returns a handle for the payload builder.
+    fn sparse_trie_handle_for(
+        &self,
+        parent_hash: B256,
+        parent_state_root: B256,
+        state: &EngineApiTreeState<N>,
+    ) -> Option<StateRootHandle>;
 }
 
 impl<N, Types, P, Evm, V> EngineValidator<Types> for BasicEngineValidator<P, Evm, V>
@@ -2011,6 +2019,27 @@ where
     fn cache_for(&self, block_hash: B256) -> Option<SavedCache> {
         Some(self.payload_processor.cache_for(block_hash))
     }
+
+    fn sparse_trie_handle_for(
+        &self,
+        parent_hash: B256,
+        parent_state_root: B256,
+        state: &EngineApiTreeState<N>,
+    ) -> Option<StateRootHandle> {
+        let (lazy_overlay, anchor_hash) = Self::get_parent_lazy_overlay(parent_hash, state);
+        let overlay_factory =
+            OverlayStateProviderFactory::new(self.provider.clone(), self.changeset_cache.clone())
+                .with_block_hash(Some(anchor_hash))
+                .with_lazy_overlay(lazy_overlay);
+
+        Some(self.payload_processor.spawn_state_root(
+            overlay_factory,
+            parent_state_root,
+            // Full proof workers — tx count unknown at FCU time (block built incrementally)
+            false,
+            &self.config,
+        ))
+    }
 }
 
 impl<P, Evm, V> WaitForCaches for BasicEngineValidator<P, Evm, V>
diff --git a/crates/engine/util/src/reorg.rs b/crates/engine/util/src/reorg.rs
index 2b82a40a887..adfec89ba0e 100644
--- a/crates/engine/util/src/reorg.rs
+++ b/crates/engine/util/src/reorg.rs
@@ -302,7 +302,7 @@ where
         cumulative_gas_used += gas_used;
     }
 
-    let BlockBuilderOutcome { block, .. } = builder.finish(&state_provider)?;
+    let BlockBuilderOutcome { block, .. } = builder.finish(&state_provider, None)?;
 
     Ok(block.into_sealed_block())
 }
diff --git a/crates/ethereum/node/tests/e2e/eth.rs b/crates/ethereum/node/tests/e2e/eth.rs
index 43b393a6ca9..4555f7c3fc6 100644
--- a/crates/ethereum/node/tests/e2e/eth.rs
+++ b/crates/ethereum/node/tests/e2e/eth.rs
@@ -255,6 +255,52 @@ async fn test_testing_build_block_v1_osaka() -> eyre::Result<()> {
     Ok(())
 }
 
+/// Tests that the sparse trie pipeline can be shared with the payload builder.
+///
+/// Enables both `share_execution_cache_with_payload_builder` and
+/// `share_sparse_trie_with_payload_builder`, then advances multiple blocks with random
+/// transactions. Each FCU spawns a `StateRootHandle` that the payload builder uses for
+/// incremental state root computation instead of blocking `state_root_with_updates()`.
+///
+/// The test validates that all blocks are successfully built and their state roots are
+/// accepted by the engine (newPayload returns VALID).
+#[tokio::test]
+async fn test_share_sparse_trie_with_payload_builder() -> eyre::Result<()> {
+    reth_tracing::init_test_tracing();
+
+    let tree_config = TreeConfig::default()
+        .with_legacy_state_root(false)
+        .with_share_execution_cache_with_payload_builder(true)
+        .with_share_sparse_trie_with_payload_builder(true);
+
+    let (mut nodes, _wallet) = setup_engine::<EthereumNode>(
+        1,
+        Arc::new(
+            ChainSpecBuilder::default()
+                .chain(MAINNET.chain)
+                .genesis(serde_json::from_str(include_str!("../assets/genesis.json")).unwrap())
+                .cancun_activated()
+                .prague_activated()
+                .build(),
+        ),
+        false,
+        tree_config,
+        eth_payload_attributes,
+    )
+    .await?;
+
+    let mut node = nodes.pop().unwrap();
+    let mut rng = rand::rng();
+
+    let num_blocks = 5;
+    advance_with_random_transactions(&mut node, num_blocks, &mut rng, true).await?;
+
+    let best_block = node.inner.provider.best_block_number()?;
+    assert_eq!(best_block, num_blocks as u64, "Expected {} blocks, got {}", num_blocks, best_block);
+
+    Ok(())
+}
+
 /// Tests that sparse trie allocation reuse works correctly across consecutive blocks.
 ///
 /// This test exercises the sparse trie allocation reuse path by:
diff --git a/crates/ethereum/payload/src/lib.rs b/crates/ethereum/payload/src/lib.rs
index 4f6342374af..d71d7490b8a 100644
--- a/crates/ethereum/payload/src/lib.rs
+++ b/crates/ethereum/payload/src/lib.rs
@@ -22,7 +22,7 @@ use reth_consensus_common::validation::MAX_RLP_BLOCK_SIZE;
 use reth_errors::{BlockExecutionError, BlockValidationError, ConsensusError};
 use reth_ethereum_primitives::{EthPrimitives, TransactionSigned};
 use reth_evm::{
-    execute::{BlockBuilder, BlockBuilderOutcome},
+    execute::{BlockBuilder, BlockBuilderOutcome, BlockExecutor},
     ConfigureEvm, Evm, NextBlockEnvAttributes,
 };
 use reth_evm_ethereum::EthEvmConfig;
@@ -119,6 +119,7 @@ where
         let args = BuildArguments::new(
             Default::default(),
             Default::default(),
+            None,
             config,
             Default::default(),
             None,
@@ -157,7 +158,14 @@ where
     Pool: TransactionPool<Transaction: PoolTransaction<Consensus = TransactionSigned>>,
     F: FnOnce(BestTransactionsAttributes) -> BestTransactionsIter<Pool>,
 {
-    let BuildArguments { mut cached_reads, execution_cache, config, cancel, best_payload } = args;
+    let BuildArguments {
+        mut cached_reads,
+        execution_cache,
+        trie_handle,
+        config,
+        cancel,
+        best_payload,
+    } = args;
     let PayloadConfig { parent_header, attributes, payload_id } = config;
 
     let mut state_provider = client.state_by_block_hash(parent_header.hash())?;
@@ -201,6 +209,12 @@ where
     ));
     let mut total_fees = U256::ZERO;
 
+    // If we have a sparse trie handle, wire a state hook that streams per-tx state diffs
+    // to the background trie pipeline for incremental state root computation.
+    if let Some(ref handle) = trie_handle {
+        builder.executor_mut().set_state_hook(Some(Box::new(handle.state_hook())));
+    }
+
     builder.apply_pre_execution_changes().map_err(|err| {
         warn!(target: "payload_builder", %err, "failed to apply pre-execution changes");
         PayloadBuilderError::Internal(err.into())
@@ -378,8 +392,31 @@ where
         return Ok(BuildOutcome::Aborted { fees: total_fees, cached_reads })
     }
 
-    let BlockBuilderOutcome { execution_result, block, .. } =
-        builder.finish(state_provider.as_ref())?;
+    let BlockBuilderOutcome { execution_result, block, .. } = if let Some(mut handle) = trie_handle
+    {
+        // Drop the state hook, which drops the StateHookSender and triggers
+        // FinishedStateUpdates via its Drop impl, signaling the trie task to finalize.
+        builder.executor_mut().set_state_hook(None);
+
+        // The sparse trie has been computing incrementally alongside tx execution.
+        // This recv() waits for the final root hash — most work is already done.
+        // Fall back to sync state root if the trie pipeline fails.
+        match handle.state_root() {
+            Ok(outcome) => {
+                debug!(target: "payload_builder", id=%payload_id, state_root=?outcome.state_root, "received state root from sparse trie");
+                builder.finish(
+                    state_provider.as_ref(),
+                    Some((outcome.state_root, Arc::unwrap_or_clone(outcome.trie_updates))),
+                )?
+            }
+            Err(err) => {
+                warn!(target: "payload_builder", id=%payload_id, %err, "sparse trie failed, falling back to sync state root");
+                builder.finish(state_provider.as_ref(), None)?
+            }
+        }
+    } else {
+        builder.finish(state_provider.as_ref(), None)?
+    };
 
     let requests = chain_spec
         .is_prague_active_at_timestamp(attributes.timestamp)
diff --git a/crates/evm/evm/src/execute.rs b/crates/evm/evm/src/execute.rs
index 595097df347..76c9f3808b3 100644
--- a/crates/evm/evm/src/execute.rs
+++ b/crates/evm/evm/src/execute.rs
@@ -358,9 +358,14 @@ pub trait BlockBuilder {
     }
 
     /// Completes the block building process and returns the [`BlockBuilderOutcome`].
+    ///
+    /// When `state_root_precomputed` is `None`, the state root is computed internally via
+    /// `state_root_with_updates()`. When `Some`, the provided root and trie updates are used
+    /// directly, skipping the expensive computation (e.g. when using the sparse trie pipeline).
     fn finish(
         self,
         state_provider: impl StateProvider,
+        state_root_precomputed: Option<(B256, TrieUpdates)>,
     ) -> Result<BlockBuilderOutcome<Self::Primitives>, BlockExecutionError>;
 
     /// Provides mutable access to the inner [`BlockExecutor`].
@@ -477,6 +482,7 @@ where
     fn finish(
         self,
         state: impl StateProvider,
+        state_root_precomputed: Option<(B256, TrieUpdates)>,
     ) -> Result<BlockBuilderOutcome<N>, BlockExecutionError> {
         let (evm, result) = self.executor.finish()?;
         let (db, evm_env) = evm.finish();
@@ -484,11 +490,13 @@ where
         // merge all transitions into bundle state
         db.merge_transitions(BundleRetention::Reverts);
 
-        // calculate the state root
         let hashed_state = state.hashed_post_state(&db.bundle_state);
-        let (state_root, trie_updates) = state
-            .state_root_with_updates(hashed_state.clone())
-            .map_err(BlockExecutionError::other)?;
+        let (state_root, trie_updates) = match state_root_precomputed {
+            Some(precomputed) => precomputed,
+            None => state
+                .state_root_with_updates(hashed_state.clone())
+                .map_err(BlockExecutionError::other)?,
+        };
 
         let (transactions, senders) =
             self.transactions.into_iter().map(|tx| tx.into_parts()).unzip();
diff --git a/crates/evm/evm/src/lib.rs b/crates/evm/evm/src/lib.rs
index ee19f749b3b..7b01da49e99 100644
--- a/crates/evm/evm/src/lib.rs
+++ b/crates/evm/evm/src/lib.rs
@@ -146,7 +146,7 @@ pub use alloy_evm::{
 /// }
 ///
 /// // Finish block building and get the outcome (block)
-/// let outcome = builder.finish(state_provider)?;
+/// let outcome = builder.finish(state_provider, None)?;
 /// let block = outcome.block;
 /// ```
 ///
@@ -399,7 +399,7 @@ pub trait ConfigureEvm: Clone + Debug + Send + Sync + Unpin {
     /// }
     ///
     /// // Complete block building
-    /// let outcome = builder.finish(state_provider)?;
+    /// let outcome = builder.finish(state_provider, None)?;
     /// ```
     fn builder_for_next_block<'a, DB: Database + 'a>(
         &'a self,
diff --git a/crates/node/core/src/args/engine.rs b/crates/node/core/src/args/engine.rs
index 2d451b84915..c52f342c3e3 100644
--- a/crates/node/core/src/args/engine.rs
+++ b/crates/node/core/src/args/engine.rs
@@ -46,6 +46,7 @@ pub struct DefaultEngineValues {
     disable_sparse_trie_cache_pruning: bool,
     state_root_task_timeout: Option<String>,
     share_execution_cache_with_payload_builder: bool,
+    share_sparse_trie_with_payload_builder: bool,
 }
 
 impl DefaultEngineValues {
@@ -211,6 +212,12 @@ impl DefaultEngineValues {
         self.share_execution_cache_with_payload_builder = v;
         self
     }
+
+    /// Set whether to share the sparse trie with the payload builder by default
+    pub const fn with_share_sparse_trie_with_payload_builder(mut self, v: bool) -> Self {
+        self.share_sparse_trie_with_payload_builder = v;
+        self
+    }
 }
 
 impl Default for DefaultEngineValues {
@@ -241,6 +248,7 @@ impl Default for DefaultEngineValues {
             disable_sparse_trie_cache_pruning: false,
             state_root_task_timeout: Some("1s".to_string()),
             share_execution_cache_with_payload_builder: false,
+            share_sparse_trie_with_payload_builder: false,
         }
     }
 }
@@ -419,6 +427,23 @@ pub struct EngineArgs {
     )]
     pub share_execution_cache_with_payload_builder: bool,
 
+    /// Whether to share the sparse trie with the payload builder.
+    ///
+    /// Replaces the payload builder's blocking `state_root_with_updates()` call with the
+    /// sparse trie, computing the state root concurrently with transaction execution.
+    ///
+    /// The engine and payload builder contend for the same trie — if a builder task is
+    /// still running when `newPayload` arrives, the engine will block until the trie is
+    /// stored back.
+    ///
+    /// The builder also anchors the trie at the built block's state root, so if the next
+    /// `newPayload` is not on top of that block, the trie cache is invalidated and cleared.
+    #[arg(
+        long = "engine.share-sparse-trie-with-payload-builder",
+        default_value_t = DefaultEngineValues::get_global().share_sparse_trie_with_payload_builder,
+    )]
+    pub share_sparse_trie_with_payload_builder: bool,
+
     /// Add random jitter before each proof computation (trie-debug only).
     /// Each proof worker sleeps for a random duration up to this value before
     /// starting work. Useful for stress-testing timing-sensitive proof logic.
@@ -462,6 +487,7 @@ impl Default for EngineArgs {
             disable_sparse_trie_cache_pruning,
             state_root_task_timeout,
             share_execution_cache_with_payload_builder,
+            share_sparse_trie_with_payload_builder,
         } = DefaultEngineValues::get_global().clone();
         Self {
             persistence_threshold,
@@ -495,6 +521,7 @@ impl Default for EngineArgs {
                 .as_deref()
                 .map(|s| humantime::parse_duration(s).expect("valid default duration")),
             share_execution_cache_with_payload_builder,
+            share_sparse_trie_with_payload_builder,
             #[cfg(feature = "trie-debug")]
             proof_jitter: None,
         }
@@ -529,6 +556,9 @@ impl EngineArgs {
             .with_state_root_task_timeout(self.state_root_task_timeout.filter(|d| !d.is_zero()))
             .with_share_execution_cache_with_payload_builder(
                 self.share_execution_cache_with_payload_builder,
+            )
+            .with_share_sparse_trie_with_payload_builder(
+                self.share_sparse_trie_with_payload_builder,
             );
         #[cfg(feature = "trie-debug")]
         let config = config.with_proof_jitter(self.proof_jitter);
@@ -588,6 +618,7 @@ mod tests {
             disable_sparse_trie_cache_pruning: true,
             state_root_task_timeout: Some(Duration::from_secs(2)),
             share_execution_cache_with_payload_builder: false,
+            share_sparse_trie_with_payload_builder: false,
             #[cfg(feature = "trie-debug")]
             proof_jitter: None,
         };
@@ -663,4 +694,19 @@ mod tests {
         .args;
         assert_eq!(args.slow_block_threshold, Some(Duration::from_millis(500)));
     }
+
+    #[test]
+    fn test_parse_share_sparse_trie_flag() {
+        let args = CommandParser::<EngineArgs>::parse_from(["reth"]).args;
+        assert!(!args.share_sparse_trie_with_payload_builder);
+        assert!(!args.tree_config().share_sparse_trie_with_payload_builder());
+
+        let args = CommandParser::<EngineArgs>::parse_from([
+            "reth",
+            "--engine.share-sparse-trie-with-payload-builder",
+        ])
+        .args;
+        assert!(args.share_sparse_trie_with_payload_builder);
+        assert!(args.tree_config().share_sparse_trie_with_payload_builder());
+    }
 }
diff --git a/crates/payload/basic/Cargo.toml b/crates/payload/basic/Cargo.toml
index 9aa2b478ced..cc83a1f1322 100644
--- a/crates/payload/basic/Cargo.toml
+++ b/crates/payload/basic/Cargo.toml
@@ -14,6 +14,7 @@ workspace = true
 [dependencies]
 # reth
 reth-execution-cache.workspace = true
+reth-trie-parallel.workspace = true
 reth-primitives-traits.workspace = true
 reth-payload-builder.workspace = true
 reth-payload-builder-primitives.workspace = true
diff --git a/crates/payload/basic/src/lib.rs b/crates/payload/basic/src/lib.rs
index cf50c1c165e..86a710837e3 100644
--- a/crates/payload/basic/src/lib.rs
+++ b/crates/payload/basic/src/lib.rs
@@ -24,6 +24,7 @@ use reth_primitives_traits::{HeaderTy, NodePrimitives, SealedHeader};
 use reth_revm::{cached::CachedReads, cancelled::CancelOnDrop};
 use reth_storage_api::{BlockReaderIdExt, StateProviderFactory};
 use reth_tasks::Runtime;
+use reth_trie_parallel::state_root_task::StateRootHandle;
 use std::{
     fmt,
     future::Future,
@@ -177,6 +178,7 @@ where
             pending_block: None,
             cached_reads,
             execution_cache: input.cache,
+            trie_handle: input.trie_handle,
             payload_task_guard: self.payload_task_guard.clone(),
             metrics: Default::default(),
             builder: self.builder.clone(),
@@ -327,6 +329,8 @@ where
     cached_reads: Option<CachedReads>,
     /// Optional execution cache shared with the engine.
     execution_cache: Option<SavedCache>,
+    /// Optional state root task handle, shared with the engine.
+    trie_handle: Option<StateRootHandle>,
     /// metrics for this type
     metrics: PayloadBuilderMetrics,
     /// The type responsible for building payloads.
@@ -353,6 +357,7 @@ where
         self.metrics.inc_initiated_payload_builds();
         let cached_reads = self.cached_reads.take().unwrap_or_default();
         let execution_cache = self.execution_cache.clone();
+        let trie_handle = self.trie_handle.take();
         let builder = self.builder.clone();
         self.executor.spawn_blocking_task(async move {
             // acquire the permit for executing the task
@@ -360,6 +365,7 @@ where
             let args = BuildArguments {
                 cached_reads,
                 execution_cache,
+                trie_handle,
                 config: payload_config,
                 cancel,
                 best_payload,
@@ -494,6 +500,7 @@ where
             let args = BuildArguments {
                 cached_reads: self.cached_reads.take().unwrap_or_default(),
                 execution_cache: self.execution_cache.clone(),
+                trie_handle: None,
                 config: self.config.clone(),
                 cancel: CancelOnDrop::default(),
                 best_payload: None,
@@ -823,6 +830,13 @@ pub struct BuildArguments<Attributes, Payload: BuiltPayload> {
     pub cached_reads: CachedReads,
     /// Optional execution cache shared with the engine.
     pub execution_cache: Option<SavedCache>,
+    /// Optional state root task handle, shared with the engine.
+    ///
+    /// The preserved trie is shared with the engine, so a concurrent `newPayload` will
+    /// block until this task completes. The trie is anchored at the built block's state
+    /// root, so if the next `newPayload` is not on top of that block, the trie cache is
+    /// invalidated and cleared.
+    pub trie_handle: Option<StateRootHandle>,
     /// How to configure the payload.
     pub config: PayloadConfig<Attributes, HeaderTy<Payload::Primitives>>,
     /// A marker that can be used to cancel the job.
@@ -836,11 +850,12 @@ impl<Attributes, Payload: BuiltPayload> BuildArguments<Attributes, Payload> {
     pub const fn new(
         cached_reads: CachedReads,
         execution_cache: Option<SavedCache>,
+        trie_handle: Option<StateRootHandle>,
         config: PayloadConfig<Attributes, HeaderTy<Payload::Primitives>>,
         cancel: CancelOnDrop,
         best_payload: Option<Payload>,
     ) -> Self {
-        Self { cached_reads, execution_cache, config, cancel, best_payload }
+        Self { cached_reads, execution_cache, trie_handle, config, cancel, best_payload }
     }
 }
 
diff --git a/crates/payload/basic/src/stack.rs b/crates/payload/basic/src/stack.rs
index fde236bff8d..57d70775253 100644
--- a/crates/payload/basic/src/stack.rs
+++ b/crates/payload/basic/src/stack.rs
@@ -153,7 +153,14 @@ where
         &self,
         args: BuildArguments<Self::Attributes, Self::BuiltPayload>,
     ) -> Result<BuildOutcome<Self::BuiltPayload>, PayloadBuilderError> {
-        let BuildArguments { cached_reads, execution_cache, config, cancel, best_payload } = args;
+        let BuildArguments {
+            cached_reads,
+            execution_cache,
+            trie_handle,
+            config,
+            cancel,
+            best_payload,
+        } = args;
         let PayloadConfig { parent_header, attributes, payload_id } = config;
 
         match attributes {
@@ -161,6 +168,7 @@ where
                 let left_args: BuildArguments<L::Attributes, L::BuiltPayload> = BuildArguments {
                     cached_reads,
                     execution_cache,
+                    trie_handle,
                     config: PayloadConfig { parent_header, attributes: left_attr, payload_id },
                     cancel,
                     best_payload: best_payload.and_then(|payload| {
@@ -177,6 +185,7 @@ where
                 let right_args = BuildArguments {
                     cached_reads,
                     execution_cache,
+                    trie_handle,
                     config: PayloadConfig { parent_header, attributes: right_attr, payload_id },
                     cancel,
                     best_payload: best_payload.and_then(|payload| {
diff --git a/crates/payload/builder/Cargo.toml b/crates/payload/builder/Cargo.toml
index 1de07938b6c..8da58d65a77 100644
--- a/crates/payload/builder/Cargo.toml
+++ b/crates/payload/builder/Cargo.toml
@@ -19,6 +19,7 @@ reth-chain-state.workspace = true
 reth-payload-builder-primitives.workspace = true
 reth-payload-primitives.workspace = true
 reth-ethereum-engine-primitives.workspace = true
+reth-trie-parallel.workspace = true
 
 # alloy
 alloy-consensus.workspace = true
@@ -47,4 +48,5 @@ test-utils = [
     "reth-primitives-traits/test-utils",
     "tokio/rt",
     "reth-execution-cache/test-utils",
+    "reth-trie-parallel/test-utils",
 ]
diff --git a/crates/payload/builder/src/service.rs b/crates/payload/builder/src/service.rs
index af19ad318bc..229bf96153a 100644
--- a/crates/payload/builder/src/service.rs
+++ b/crates/payload/builder/src/service.rs
@@ -16,6 +16,7 @@ use reth_execution_cache::SavedCache;
 use reth_payload_builder_primitives::{Events, PayloadBuilderError, PayloadEvents};
 use reth_payload_primitives::{BuiltPayload, PayloadAttributes, PayloadKind, PayloadTypes};
 use reth_primitives_traits::{FastInstant as Instant, NodePrimitives};
+use reth_trie_parallel::state_root_task::StateRootHandle;
 use std::{
     future::Future,
     pin::Pin,
@@ -522,7 +523,7 @@ pub enum PayloadServiceCommand<T: PayloadTypes> {
 }
 
 /// A request to build a new payload.
-#[derive(Debug, Clone)]
+#[derive(Debug)]
 pub struct BuildNewPayload<T> {
     /// The attributes for the new payload
     pub attributes: T,
@@ -532,6 +533,8 @@ pub struct BuildNewPayload<T> {
     ///
     /// Only provided if `--engine.share-execution-cache-with-payload-builder` is enabled.
     pub cache: Option<SavedCache>,
+    /// Optional handle to a background sparse trie task.
+    pub trie_handle: Option<StateRootHandle>,
 }
 
 impl<T: PayloadAttributes> BuildNewPayload<T> {
diff --git a/crates/rpc/rpc-eth-api/src/helpers/pending_block.rs b/crates/rpc/rpc-eth-api/src/helpers/pending_block.rs
index 77586278e2c..1b3cbf05706 100644
--- a/crates/rpc/rpc-eth-api/src/helpers/pending_block.rs
+++ b/crates/rpc/rpc-eth-api/src/helpers/pending_block.rs
@@ -378,7 +378,7 @@ pub trait LoadPendingBlock:
         }
 
         let BlockBuilderOutcome { execution_result, block, hashed_state, trie_updates } =
-            builder.finish(NoopProvider::default()).map_err(Self::Error::from_eth_err)?;
+            builder.finish(NoopProvider::default(), None).map_err(Self::Error::from_eth_err)?;
 
         let execution_outcome =
             BlockExecutionOutput { state: db.take_bundle(), result: execution_result };
diff --git a/crates/rpc/rpc-eth-types/src/simulate.rs b/crates/rpc/rpc-eth-types/src/simulate.rs
index ab58a95319a..c4908c2e072 100644
--- a/crates/rpc/rpc-eth-types/src/simulate.rs
+++ b/crates/rpc/rpc-eth-types/src/simulate.rs
@@ -203,7 +203,7 @@ where
     }
 
     // Pass noop provider to skip state root calculations.
-    let result = builder.finish(NoopProvider::default())?;
+    let result = builder.finish(NoopProvider::default(), None)?;
 
     Ok((result, results))
 }
diff --git a/crates/rpc/rpc/src/testing.rs b/crates/rpc/rpc/src/testing.rs
index d6932454682..6700c026972 100644
--- a/crates/rpc/rpc/src/testing.rs
+++ b/crates/rpc/rpc/src/testing.rs
@@ -205,7 +205,7 @@ where
                     block_transactions_rlp_length += tx_rlp_len;
                     total_fees += U256::from(tip) * U256::from(gas_used);
                 }
-                let outcome = builder.finish(&state).map_err(Eth::Error::from_eth_err)?;
+                let outcome = builder.finish(&state, None).map_err(Eth::Error::from_eth_err)?;
 
                 let has_requests = outcome.block.requests_hash().is_some();
                 let sealed_block = Arc::new(outcome.block.into_sealed_block());
diff --git a/crates/trie/parallel/Cargo.toml b/crates/trie/parallel/Cargo.toml
index 216e71a1fd3..ac80c5ba41c 100644
--- a/crates/trie/parallel/Cargo.toml
+++ b/crates/trie/parallel/Cargo.toml
@@ -22,12 +22,17 @@ reth-tasks = { workspace = true, features = ["rayon"] }
 reth-trie.workspace = true
 
 # alloy
-alloy-rlp.workspace = true
+alloy-eip7928.workspace = true
+alloy-evm.workspace = true
 alloy-primitives.workspace = true
+alloy-rlp.workspace = true
 
 # tracing
 tracing.workspace = true
 
+# revm
+revm-state.workspace = true
+
 # misc
 thiserror.workspace = true
 derive_more.workspace = true
@@ -58,7 +63,7 @@ tokio = { workspace = true, features = ["rt", "rt-multi-thread", "macros"] }
 [features]
 default = ["metrics"]
 metrics = ["reth-metrics", "dep:metrics", "reth-trie/metrics", "reth-trie-sparse/metrics"]
-trie-debug = ["dep:rand"]
+trie-debug = ["dep:rand", "reth-trie-sparse/trie-debug"]
 test-utils = [
     "reth-primitives-traits/test-utils",
     "reth-provider/test-utils",
diff --git a/crates/trie/parallel/src/lib.rs b/crates/trie/parallel/src/lib.rs
index 020e3266ea4..5f10c5ca314 100644
--- a/crates/trie/parallel/src/lib.rs
+++ b/crates/trie/parallel/src/lib.rs
@@ -20,6 +20,9 @@ pub mod root;
 /// Implementation of parallel proof computation.
 pub mod proof_task;
 
+/// State root task interface types shared between the engine tree and the payload builder.
+pub mod state_root_task;
+
 /// Async value encoder for V2 proofs.
 pub(crate) mod value_encoder;
 
diff --git a/crates/trie/parallel/src/state_root_task.rs b/crates/trie/parallel/src/state_root_task.rs
new file mode 100644
index 00000000000..24a1e3269d9
--- /dev/null
+++ b/crates/trie/parallel/src/state_root_task.rs
@@ -0,0 +1,201 @@
+//! State root task interface types shared between the engine tree and the payload builder.
+
+use crate::root::ParallelStateRootError;
+use alloy_eip7928::BlockAccessList;
+use alloy_evm::block::StateChangeSource;
+use alloy_primitives::{keccak256, B256};
+use derive_more::derive::Deref;
+use reth_trie::{updates::TrieUpdates, HashedPostState, HashedStorage, MultiProofTargetsV2};
+use revm_state::EvmState;
+use std::sync::Arc;
+use tracing::trace;
+
+/// Source of state changes, either from EVM execution or from a Block Access List.
+#[derive(Clone, Copy)]
+pub enum Source {
+    /// State changes from EVM execution.
+    Evm(StateChangeSource),
+    /// State changes from Block Access List (EIP-7928).
+    BlockAccessList,
+}
+
+impl std::fmt::Debug for Source {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Evm(source) => source.fmt(f),
+            Self::BlockAccessList => f.write_str("BlockAccessList"),
+        }
+    }
+}
+
+impl From<StateChangeSource> for Source {
+    fn from(source: StateChangeSource) -> Self {
+        Self::Evm(source)
+    }
+}
+
+/// Messages used internally by the multi proof task.
+#[derive(Debug)]
+pub enum StateRootMessage {
+    /// Prefetch proof targets
+    PrefetchProofs(MultiProofTargetsV2),
+    /// New state update from transaction execution with its source
+    StateUpdate(Source, EvmState),
+    /// Pre-hashed state update from BAL conversion that can be applied directly without proofs.
+    HashedStateUpdate(HashedPostState),
+    /// Block Access List (EIP-7928; BAL) containing complete state changes for the block.
+    ///
+    /// When received, the task generates a single state update from the BAL and processes it.
+    /// No further messages are expected after receiving this variant.
+    BlockAccessList(Arc<BlockAccessList>),
+    /// Signals state update stream end.
+    ///
+    /// This is triggered by block execution, indicating that no additional state updates are
+    /// expected.
+    FinishedStateUpdates,
+}
+
+/// Outcome of the state root computation, including the state root itself with
+/// the trie updates.
+#[derive(Debug, Clone)]
+pub struct StateRootComputeOutcome {
+    /// The state root.
+    pub state_root: B256,
+    /// The trie updates.
+    pub trie_updates: Arc<TrieUpdates>,
+    /// Debug recorders taken from the sparse tries, keyed by `None` for account trie
+    /// and `Some(address)` for storage tries.
+    #[cfg(feature = "trie-debug")]
+    pub debug_recorders: Vec<(Option<B256>, reth_trie_sparse::debug_recorder::TrieDebugRecorder)>,
+}
+
+/// Handle to a background sparse trie state root computation.
+///
+/// Used by both the engine (during `newPayload`) and the payload builder (during `FCU`-triggered
+/// block building). Provides channels for streaming state updates into the pipeline and receiving
+/// the final computed state root.
+///
+/// Created by `PayloadProcessor::spawn_state_root`.
+#[derive(Debug)]
+pub struct StateRootHandle {
+    /// The state root that the cached sparse trie is anchored at (parent block's state root).
+    cached_trie_state_root: B256,
+    /// Channel for streaming state updates and proof targets into the sparse trie pipeline.
+    updates_tx: crossbeam_channel::Sender<StateRootMessage>,
+    /// Receiver for the final state root result.
+    state_root_rx:
+        Option<std::sync::mpsc::Receiver<Result<StateRootComputeOutcome, ParallelStateRootError>>>,
+}
+
+impl StateRootHandle {
+    /// Creates a new [`StateRootHandle`].
+    pub const fn new(
+        cached_trie_state_root: B256,
+        updates_tx: crossbeam_channel::Sender<StateRootMessage>,
+        state_root_rx: std::sync::mpsc::Receiver<
+            Result<StateRootComputeOutcome, ParallelStateRootError>,
+        >,
+    ) -> Self {
+        Self { cached_trie_state_root, updates_tx, state_root_rx: Some(state_root_rx) }
+    }
+
+    /// Returns the state root that the cached sparse trie is anchored at.
+    pub const fn cached_trie_state_root(&self) -> B256 {
+        self.cached_trie_state_root
+    }
+
+    /// Returns a reference to the updates sender channel.
+    pub const fn updates_tx(&self) -> &crossbeam_channel::Sender<StateRootMessage> {
+        &self.updates_tx
+    }
+
+    /// Returns a state hook that streams state updates to the background state root task.
+    ///
+    /// The hook must be dropped after execution completes to signal the end of state updates.
+    pub fn state_hook(&self) -> impl alloy_evm::block::OnStateHook {
+        let sender = StateHookSender::new(self.updates_tx.clone());
+
+        move |source: StateChangeSource, state: &EvmState| {
+            let _ = sender.send(StateRootMessage::StateUpdate(source.into(), state.clone()));
+        }
+    }
+
+    /// Awaits the state root computation result.
+    ///
+    /// # Panics
+    ///
+    /// If called more than once.
+    pub fn state_root(&mut self) -> Result<StateRootComputeOutcome, ParallelStateRootError> {
+        self.state_root_rx
+            .take()
+            .expect("state_root already taken")
+            .recv()
+            .map_err(|_| ParallelStateRootError::Other("sparse trie task dropped".to_string()))?
+    }
+
+    /// Takes the state root receiver for use with custom waiting logic (e.g., timeouts).
+    ///
+    /// # Panics
+    ///
+    /// If called more than once.
+    pub const fn take_state_root_rx(
+        &mut self,
+    ) -> std::sync::mpsc::Receiver<Result<StateRootComputeOutcome, ParallelStateRootError>> {
+        self.state_root_rx.take().expect("state_root already taken")
+    }
+}
+
+/// A wrapper for the sender that signals completion when dropped.
+///
+/// This type is intended to be used in combination with the evm executor statehook.
+/// This should trigger once the block has been executed (after) the last state update has been
+/// sent. This triggers the exit condition of the multi proof task.
+#[derive(Deref, Debug)]
+pub struct StateHookSender(crossbeam_channel::Sender<StateRootMessage>);
+
+impl StateHookSender {
+    /// Creates a new [`StateHookSender`] wrapping the given channel sender.
+    pub const fn new(inner: crossbeam_channel::Sender<StateRootMessage>) -> Self {
+        Self(inner)
+    }
+}
+
+impl Drop for StateHookSender {
+    fn drop(&mut self) {
+        // Send completion signal when the sender is dropped
+        let _ = self.0.send(StateRootMessage::FinishedStateUpdates);
+    }
+}
+
+/// Converts [`EvmState`] to [`HashedPostState`] by keccak256-hashing addresses and storage slots.
+pub fn evm_state_to_hashed_post_state(update: EvmState) -> HashedPostState {
+    let mut hashed_state = HashedPostState::with_capacity(update.len());
+
+    for (address, account) in update {
+        if account.is_touched() {
+            let hashed_address = keccak256(address);
+            trace!(target: "trie::parallel::sparse", ?address, ?hashed_address, "Adding account to state update");
+
+            let destroyed = account.is_selfdestructed();
+            let info = if destroyed { None } else { Some(account.info.into()) };
+            hashed_state.accounts.insert(hashed_address, info);
+
+            let mut changed_storage_iter = account
+                .storage
+                .into_iter()
+                .filter(|(_slot, value)| value.is_changed())
+                .map(|(slot, value)| (keccak256(B256::from(slot)), value.present_value))
+                .peekable();
+
+            if destroyed {
+                hashed_state.storages.insert(hashed_address, HashedStorage::new(true));
+            } else if changed_storage_iter.peek().is_some() {
+                hashed_state
+                    .storages
+                    .insert(hashed_address, HashedStorage::from_iter(false, changed_storage_iter));
+            }
+        }
+    }
+
+    hashed_state
+}
diff --git a/docs/vocs/docs/pages/cli/reth/node.mdx b/docs/vocs/docs/pages/cli/reth/node.mdx
index 64c59f1907f..e138938632c 100644
--- a/docs/vocs/docs/pages/cli/reth/node.mdx
+++ b/docs/vocs/docs/pages/cli/reth/node.mdx
@@ -1027,6 +1027,15 @@ Engine:
 
           Note: this should only be enabled if node would not be requested to process any payloads in parallel with payload building.
 
+      --engine.share-sparse-trie-with-payload-builder
+          Whether to share the sparse trie with the payload builder.
+
+          Replaces the payload builder's blocking `state_root_with_updates()` call with the sparse trie, computing the state root concurrently with transaction execution.
+
+          The engine and payload builder contend for the same trie — if a builder task is still running when `newPayload` arrives, the engine will block until the trie is stored back.
+
+          The builder also anchors the trie at the built block's state root, so if the next `newPayload` is not on top of that block, the trie cache is invalidated and cleared.
+
 ERA:
       --era.enable
           Enable import from ERA1 files
diff --git a/examples/custom-engine-types/src/main.rs b/examples/custom-engine-types/src/main.rs
index bae94409d45..7c6d6de4fbc 100644
--- a/examples/custom-engine-types/src/main.rs
+++ b/examples/custom-engine-types/src/main.rs
@@ -326,7 +326,14 @@ where
         &self,
         args: BuildArguments<Self::Attributes, Self::BuiltPayload>,
     ) -> Result<BuildOutcome<Self::BuiltPayload>, PayloadBuilderError> {
-        let BuildArguments { cached_reads, execution_cache, config, cancel, best_payload } = args;
+        let BuildArguments {
+            cached_reads,
+            execution_cache,
+            trie_handle,
+            config,
+            cancel,
+            best_payload,
+        } = args;
         let PayloadConfig { parent_header, attributes, payload_id } = config;
 
         // This reuses the default EthereumPayloadBuilder to build the payload
@@ -334,6 +341,7 @@ where
         self.inner.try_build(BuildArguments {
             cached_reads,
             execution_cache,
+            trie_handle,
             config: PayloadConfig { parent_header, attributes: attributes.inner, payload_id },
             cancel,
             best_payload,