diff --git a/Cargo.lock b/Cargo.lock
index ad5beecf..ca6e4ea0 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4693,12 +4693,13 @@ dependencies = [
  "anyhow",
  "async-broadcast",
  "async-compatibility-layer",
- "async-lock 3.4.0",
+ "async-lock 2.8.0",
  "async-std",
  "async-trait",
  "clap",
  "committable",
  "derivative",
+ "either",
  "futures",
  "hex",
  "hotshot",
diff --git a/Cargo.toml b/Cargo.toml
index e901e0ed..f9c07811 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -10,12 +10,13 @@ async-broadcast = "0.7"
 async-compatibility-layer = { version = "1.2.1", default-features = false, features = [
     "logging-utils",
 ] }
-async-lock = "3.4"
+async-lock = "2.2"
 async-std = { version = "1.9.0", features = ["unstable", "attributes"] }
 async-trait = "0.1"
 clap = { version = "4.5", features = ["derive", "env"] }
 committable = "0.2"
 derivative = "2.2"
+either = "1.13"
 futures = "0.3"
 hex = "0.4.3"
 hotshot = { git = "https://github.com/EspressoSystems/HotShot.git", tag = "0.5.71" }
diff --git a/src/builder_state.rs b/src/builder_state.rs
index 9706c208..9b1b5cb9 100644
--- a/src/builder_state.rs
+++ b/src/builder_state.rs
@@ -25,11 +25,11 @@ use async_lock::RwLock;
 use core::panic;
 use futures::StreamExt;
 
+use std::cmp::PartialEq;
 use std::collections::{HashMap, HashSet};
 use std::fmt::Debug;
 use std::sync::Arc;
 use std::time::Instant;
-use std::{cmp::PartialEq, num::NonZeroUsize};
 use std::{collections::hash_map::Entry, time::Duration};
 
 pub type TxTimeStamp = u128;
@@ -51,7 +51,6 @@ pub struct DecideMessage<TYPES: NodeType> {
 pub struct DaProposalMessage<TYPES: NodeType> {
     pub view_number: TYPES::Time,
     pub txn_commitments: Vec<Commitment<TYPES::Transaction>>,
-    pub num_nodes: usize,
     pub sender: <TYPES as NodeType>::SignatureKey,
     pub builder_commitment: BuilderCommitment,
 }
@@ -156,9 +155,6 @@ pub struct BuilderState<TYPES: NodeType> {
     /// global state handle, defined in the service.rs
     pub global_state: Arc<RwLock<GlobalState<TYPES>>>,
 
-    /// total nodes required for the VID computation as part of block header input response
-    pub num_nodes: NonZeroUsize,
-
     /// locally spawned builder Commitements
     pub builder_commitments: HashSet<(BuilderStateId<TYPES>, BuilderCommitment)>,
 
@@ -375,7 +371,6 @@ impl<TYPES: NodeType> BuilderState<TYPES> {
         quorum_proposal: Arc<Proposal<TYPES, QuorumProposal<TYPES>>>,
         req_sender: BroadcastSender<MessageType<TYPES>>,
     ) {
-        self.num_nodes = NonZeroUsize::new(da_proposal_info.num_nodes).unwrap_or(self.num_nodes);
         self.built_from_proposed_block.view_number = quorum_proposal.data.view_number;
         self.built_from_proposed_block.vid_commitment =
             quorum_proposal.data.block_header.payload_commitment();
@@ -659,7 +654,6 @@ impl<TYPES: NodeType> BuilderState<TYPES> {
         req_receiver: BroadcastReceiver<MessageType<TYPES>>,
         tx_queue: Vec<Arc<ReceivedTransaction<TYPES>>>,
         global_state: Arc<RwLock<GlobalState<TYPES>>>,
-        num_nodes: NonZeroUsize,
         maximize_txn_capture_timeout: Duration,
         base_fee: u64,
         instance_state: Arc<TYPES::InstanceState>,
@@ -668,7 +662,6 @@ impl<TYPES: NodeType> BuilderState<TYPES> {
     ) -> Self {
         let txns_in_queue: HashSet<_> = tx_queue.iter().map(|tx| tx.commit).collect();
         BuilderState {
-            num_nodes,
             txns_in_queue,
             built_from_proposed_block,
             req_receiver,
@@ -706,7 +699,6 @@ impl<TYPES: NodeType> BuilderState<TYPES> {
             tx_queue: self.tx_queue.clone(),
             global_state: self.global_state.clone(),
             builder_commitments: self.builder_commitments.clone(),
-            num_nodes: self.num_nodes,
             maximize_txn_capture_timeout: self.maximize_txn_capture_timeout,
             base_fee: self.base_fee,
             instance_state: self.instance_state.clone(),
diff --git a/src/service.rs b/src/service.rs
index 4b52faaa..b6e252fb 100644
--- a/src/service.rs
+++ b/src/service.rs
@@ -1,26 +1,24 @@
-use anyhow::Context;
-use hotshot::{traits::election::static_committee::GeneralStaticCommittee, types::Event};
+use anyhow::bail;
+use hotshot::types::Event;
 use hotshot_builder_api::v0_3::{
     builder::BuildError,
     data_source::{AcceptsTxnSubmits, BuilderDataSource},
 };
+use hotshot_types::bundle::Bundle;
 use hotshot_types::traits::block_contents::BuilderFee;
-use hotshot_types::{bundle::Bundle, traits::node_implementation::Versions};
 use hotshot_types::{
     data::{DaProposal, Leaf, QuorumProposal, ViewNumber},
     event::EventType,
     message::Proposal,
     traits::{
         block_contents::BlockPayload,
-        election::Membership,
-        network::Topic,
         node_implementation::{ConsensusTime, NodeType},
         signature_key::{BuilderSignatureKey, SignatureKey},
     },
     utils::BuilderCommitment,
     vid::VidCommitment,
 };
-use tracing::error;
+use tracing::{error, instrument};
 
 use std::{fmt::Debug, marker::PhantomData, time::Duration};
 
@@ -31,24 +29,21 @@ use crate::{
     },
     utils::{BlockId, BuilderStateId},
 };
-use anyhow::anyhow;
 pub use async_broadcast::{broadcast, RecvError, TryRecvError};
 use async_broadcast::{InactiveReceiver, Sender as BroadcastSender, TrySendError};
 use async_lock::RwLock;
 use async_trait::async_trait;
 use committable::{Commitment, Committable};
 use derivative::Derivative;
-use futures::future::BoxFuture;
 use futures::stream::StreamExt;
-use hotshot_events_service::{events::Error as EventStreamError, events_source::StartupInfo};
+use futures::{future::BoxFuture, Stream};
 use sha2::{Digest, Sha256};
 use std::collections::HashMap;
 use std::num::NonZeroUsize;
 use std::sync::Arc;
 use std::{fmt::Display, time::Instant};
-use surf_disco::Client;
 use tagged_base64::TaggedBase64;
-use tide_disco::{method::ReadState, Url};
+use tide_disco::method::ReadState;
 
 // It holds all the necessary information for a block
 #[derive(Debug)]
@@ -132,8 +127,6 @@ impl<TYPES: NodeType> GlobalState<TYPES> {
         tx_sender: BroadcastSender<Arc<ReceivedTransaction<TYPES>>>,
         bootstrapped_builder_state_id: VidCommitment,
         bootstrapped_view_num: TYPES::Time,
-        last_garbage_collected_view_num: TYPES::Time,
-        _buffer_view_num_count: u64,
     ) -> Self {
         let mut spawned_builder_states = HashMap::new();
         let bootstrap_id = BuilderStateId {
@@ -145,7 +138,7 @@ impl<TYPES: NodeType> GlobalState<TYPES> {
             blocks: lru::LruCache::new(NonZeroUsize::new(256).unwrap()),
             spawned_builder_states,
             tx_sender,
-            last_garbage_collected_view_num,
+            last_garbage_collected_view_num: bootstrapped_view_num,
             builder_state_to_last_built_block: Default::default(),
             highest_view_num_builder_id: bootstrap_id,
         }
@@ -568,66 +561,8 @@ pub struct BroadcastSenders<TYPES: NodeType> {
     pub decide: BroadcastSender<MessageType<TYPES>>,
 }
 
-async fn connect_to_events_service<TYPES: NodeType, V: Versions>(
-    hotshot_events_api_url: Url,
-) -> Option<(
-    surf_disco::socket::Connection<
-        Event<TYPES>,
-        surf_disco::socket::Unsupported,
-        EventStreamError,
-        V::Base,
-    >,
-    GeneralStaticCommittee<TYPES, <TYPES as NodeType>::SignatureKey>,
-)> {
-    let client = Client::<hotshot_events_service::events::Error, V::Base>::new(
-        hotshot_events_api_url.clone(),
-    );
-
-    if !(client.connect(None).await) {
-        return None;
-    }
-
-    tracing::info!("Builder client connected to the hotshot events api");
-
-    // client subscrive to hotshot events
-    let subscribed_events = client
-        .socket("hotshot-events/events")
-        .subscribe::<Event<TYPES>>()
-        .await
-        .ok()?;
-
-    // handle the startup event at the start
-    let membership = if let Ok(response) = client
-        .get::<StartupInfo<TYPES>>("hotshot-events/startup_info")
-        .send()
-        .await
-    {
-        let StartupInfo {
-            known_node_with_stake,
-            non_staked_node_count,
-        } = response;
-        let membership: GeneralStaticCommittee<TYPES, <TYPES as NodeType>::SignatureKey> =
-            GeneralStaticCommittee::<TYPES, <TYPES as NodeType>::SignatureKey>::create_election(
-                known_node_with_stake.clone(),
-                known_node_with_stake.clone(),
-                Topic::Global,
-                0,
-            );
-
-        tracing::info!(
-            "Startup info: Known nodes with stake: {:?}, Non-staked node count: {:?}",
-            known_node_with_stake,
-            non_staked_node_count
-        );
-        Some(membership)
-    } else {
-        None
-    };
-    membership.map(|membership| (subscribed_events, membership))
-}
-
 #[async_trait]
-pub trait BuilderHooks<TYPES: NodeType>: Sync + Send {
+pub trait BuilderHooks<TYPES: NodeType>: Sync + Send + 'static {
     #[inline(always)]
     async fn process_transactions(
         self: &Arc<Self>,
@@ -644,114 +579,60 @@ pub struct NoHooks<TYPES: NodeType>(pub PhantomData<TYPES>);
 
 impl<TYPES: NodeType> BuilderHooks<TYPES> for NoHooks<TYPES> {}
 
-/*
-Running Non-Permissioned Builder Service
-*/
-pub async fn run_non_permissioned_standalone_builder_service<
-    TYPES: NodeType<Time = ViewNumber>,
-    V: Versions,
->(
+/// Run builder service,
+/// Refer to documentation for [`ProxyGlobalState`] for more details
+pub async fn run_builder_service<TYPES: NodeType<Time = ViewNumber>>(
     hooks: Arc<impl BuilderHooks<TYPES>>,
-
-    // Sending from the hotshot to the builder states.
     senders: BroadcastSenders<TYPES>,
-
-    // Url to (re)connect to for the events stream
-    hotshot_events_api_url: Url,
+    hotshot_event_stream: impl Stream<Item = Event<TYPES>>,
 ) -> Result<(), anyhow::Error> {
-    // connection to the events stream
-    let connected = connect_to_events_service::<TYPES, V>(hotshot_events_api_url.clone()).await;
-    if connected.is_none() {
-        return Err(anyhow!(
-            "failed to connect to API at {hotshot_events_api_url}"
-        ));
-    }
-    let (mut subscribed_events, mut membership) =
-        connected.context("Failed to connect to events service")?;
-
+    let mut hotshot_event_stream = std::pin::pin!(hotshot_event_stream);
     loop {
-        let event = subscribed_events.next().await;
-        //tracing::debug!("Builder Event received from HotShot: {:?}", event);
-        match event {
-            Some(Ok(event)) => {
-                hooks.handle_hotshot_event(&event).await;
-
-                match event.event {
-                    EventType::Error { error } => {
-                        error!("Error event in HotShot: {:?}", error);
-                    }
-                    // tx event
-                    EventType::Transactions { transactions } => {
-                        let transactions = hooks.process_transactions(transactions).await;
-
-                        for res in handle_received_txns(
-                            &senders.transactions,
-                            transactions,
-                            TransactionSource::HotShot,
-                        )
-                        .await
-                        {
-                            if let Err(e) = res {
-                                tracing::warn!("Failed to handle transactions; {:?}", e);
-                            }
-                        }
-                    }
-                    // decide event
-                    EventType::Decide {
-                        block_size: _,
-                        leaf_chain,
-                        qc: _,
-                    } => {
-                        let latest_decide_view_num = leaf_chain[0].leaf.view_number();
-                        handle_decide_event(&senders.decide, latest_decide_view_num).await;
-                    }
-                    // DA proposal event
-                    EventType::DaProposal { proposal, sender } => {
-                        // get the leader for current view
-                        let leader = membership.leader(proposal.data.view_number);
-                        // get the committee mstatked node count
-                        let total_nodes = membership.total_nodes();
-
-                        handle_da_event(
-                            &senders.da_proposal,
-                            proposal,
-                            sender,
-                            leader,
-                            NonZeroUsize::new(total_nodes).unwrap_or(NonZeroUsize::MIN),
-                        )
-                        .await;
-                    }
-                    // QC proposal event
-                    EventType::QuorumProposal { proposal, sender } => {
-                        // get the leader for current view
-                        let leader = membership.leader(proposal.data.view_number);
-                        handle_qc_event(
-                            &senders.quorum_proposal,
-                            Arc::new(proposal),
-                            sender,
-                            leader,
-                        )
-                        .await;
-                    }
-                    _ => {
-                        tracing::trace!("Unhandled event from Builder: {:?}", event.event);
+        let Some(event) = hotshot_event_stream.next().await else {
+            bail!("Event stream ended");
+        };
+
+        hooks.handle_hotshot_event(&event).await;
+
+        match event.event {
+            EventType::Error { error } => {
+                error!("Error event in HotShot: {:?}", error);
+            }
+            // tx event
+            EventType::Transactions { transactions } => {
+                let transactions = hooks.process_transactions(transactions).await;
+
+                for res in handle_received_txns(
+                    &senders.transactions,
+                    transactions,
+                    TransactionSource::HotShot,
+                )
+                .await
+                {
+                    if let Err(e) = res {
+                        tracing::warn!("Failed to handle transactions; {:?}", e);
                     }
                 }
             }
-            Some(Err(e)) => {
-                error!("Error in the event stream: {:?}", e);
+            // decide event
+            EventType::Decide {
+                block_size: _,
+                leaf_chain,
+                qc: _,
+            } => {
+                let latest_decide_view_num = leaf_chain[0].leaf.view_number();
+                handle_decide_event(&senders.decide, latest_decide_view_num).await;
             }
-            None => {
-                error!("Event stream ended");
-                let connected =
-                    connect_to_events_service::<_, V>(hotshot_events_api_url.clone()).await;
-                if connected.is_none() {
-                    return Err(anyhow!(
-                        "failed to reconnect to API at {hotshot_events_api_url}"
-                    ));
-                }
-                (subscribed_events, membership) =
-                    connected.context("Failed to reconnect to events service")?;
+            // DA proposal event
+            EventType::DaProposal { proposal, sender } => {
+                handle_da_event(&senders.da_proposal, proposal, sender).await;
+            }
+            // QC proposal event
+            EventType::QuorumProposal { proposal, sender } => {
+                handle_qc_event(&senders.quorum_proposal, Arc::new(proposal), sender).await;
+            }
+            _ => {
+                tracing::trace!("Unhandled event from Builder: {:?}", event.event);
             }
         }
     }
@@ -760,102 +641,87 @@ pub async fn run_non_permissioned_standalone_builder_service<
 /*
 Utility functions to handle the hotshot events
 */
+#[instrument(skip_all, fields(sender, da_proposal.data.view_number))]
 async fn handle_da_event<TYPES: NodeType>(
     da_channel_sender: &BroadcastSender<MessageType<TYPES>>,
     da_proposal: Proposal<TYPES, DaProposal<TYPES>>,
     sender: <TYPES as NodeType>::SignatureKey,
-    leader: <TYPES as NodeType>::SignatureKey,
-    total_nodes: NonZeroUsize,
 ) {
-    tracing::debug!(
-        "DaProposal: Leader: {:?} for the view: {:?}",
-        leader,
-        da_proposal.data.view_number
-    );
-
     // get the encoded transactions hash
     let encoded_txns_hash = Sha256::digest(&da_proposal.data.encoded_transactions);
     // check if the sender is the leader and the signature is valid; if yes, broadcast the DA proposal
-    if leader == sender && sender.validate(&da_proposal.signature, &encoded_txns_hash) {
-        let view_number = da_proposal.data.view_number;
-        tracing::debug!(
-            "Sending DA proposal to the builder states for view {:?}",
-            view_number
-        );
+    if !sender.validate(&da_proposal.signature, &encoded_txns_hash) {
+        error!("Validation Failure on DaProposal");
+        return;
+    }
 
-        // form a block payload from the encoded transactions
-        let block_payload = <TYPES::BlockPayload as BlockPayload<TYPES>>::from_bytes(
-            &da_proposal.data.encoded_transactions,
-            &da_proposal.data.metadata,
-        );
-        // get the builder commitment from the block payload
-        let builder_commitment = block_payload.builder_commitment(&da_proposal.data.metadata);
-
-        let txn_commitments = block_payload
-            .transactions(&da_proposal.data.metadata)
-            // TODO:
-            //.filter(|txn| txn.namespace_id() != namespace_id)
-            .map(|txn| txn.commit())
-            .collect();
-
-        let da_msg = DaProposalMessage {
-            view_number,
-            txn_commitments,
-            num_nodes: total_nodes.into(),
-            sender,
-            builder_commitment,
-        };
+    let view_number = da_proposal.data.view_number;
+    tracing::debug!("Sending DA proposal to the builder states",);
 
-        if let Err(e) = da_channel_sender
-            .broadcast(MessageType::DaProposalMessage(Arc::new(da_msg)))
-            .await
-        {
-            tracing::warn!(
-                "Error {e}, failed to send DA proposal to builder states for view {:?}",
-                view_number
-            );
-        }
-    } else {
-        error!("Validation Failure on DaProposal for view {:?}: Leader for the current view: {:?} and sender: {:?}", da_proposal.data.view_number, leader, sender);
+    // form a block payload from the encoded transactions
+    let block_payload = <TYPES::BlockPayload as BlockPayload<TYPES>>::from_bytes(
+        &da_proposal.data.encoded_transactions,
+        &da_proposal.data.metadata,
+    );
+    // get the builder commitment from the block payload
+    let builder_commitment = block_payload.builder_commitment(&da_proposal.data.metadata);
+
+    let txn_commitments = block_payload
+        .transactions(&da_proposal.data.metadata)
+        // TODO:
+        //.filter(|txn| txn.namespace_id() != namespace_id)
+        .map(|txn| txn.commit())
+        .collect();
+
+    let da_msg = DaProposalMessage {
+        view_number,
+        txn_commitments,
+        sender,
+        builder_commitment,
+    };
+
+    if let Err(e) = da_channel_sender
+        .broadcast(MessageType::DaProposalMessage(Arc::new(da_msg)))
+        .await
+    {
+        tracing::warn!(
+            "Error {e}, failed to send DA proposal to builder states for view {:?}",
+            view_number
+        );
     }
 }
 
+#[instrument(skip_all, fields(sender, quorum_proposal.data.view_number))]
 async fn handle_qc_event<TYPES: NodeType>(
     qc_channel_sender: &BroadcastSender<MessageType<TYPES>>,
     quorum_proposal: Arc<Proposal<TYPES, QuorumProposal<TYPES>>>,
     sender: <TYPES as NodeType>::SignatureKey,
-    leader: <TYPES as NodeType>::SignatureKey,
 ) {
-    tracing::debug!(
-        "QCProposal: Leader: {:?} for the view: {:?}",
-        leader,
-        quorum_proposal.data.view_number
-    );
-
     let leaf = Leaf::from_quorum_proposal(&quorum_proposal.data);
 
     // check if the sender is the leader and the signature is valid; if yes, broadcast the QC proposal
-    if sender == leader && sender.validate(&quorum_proposal.signature, leaf.commit().as_ref()) {
-        let qc_msg = QuorumProposalMessage::<TYPES> {
-            proposal: quorum_proposal,
-            sender: leader,
-        };
-        let view_number = qc_msg.proposal.data.view_number;
-        tracing::debug!(
-            "Sending QC proposal to the builder states for view {:?}",
+    if !sender.validate(&quorum_proposal.signature, leaf.commit().as_ref()) {
+        error!("Validation Failure on QCProposal");
+        return;
+    };
+
+    let qc_msg = QuorumProposalMessage::<TYPES> {
+        proposal: quorum_proposal,
+        sender,
+    };
+    let view_number = qc_msg.proposal.data.view_number;
+    tracing::debug!(
+        "Sending QC proposal to the builder states for view {:?}",
+        view_number
+    );
+    if let Err(e) = qc_channel_sender
+        .broadcast(MessageType::QuorumProposalMessage(qc_msg))
+        .await
+    {
+        tracing::warn!(
+            "Error {e}, failed to send QC proposal to builder states for view {:?}",
             view_number
         );
-        if let Err(e) = qc_channel_sender
-            .broadcast(MessageType::QuorumProposalMessage(qc_msg))
-            .await
-        {
-            tracing::warn!(
-                "Error {e}, failed to send QC proposal to builder states for view {:?}",
-                view_number
-            );
-        }
-    } else {
-        error!("Validation Failure on QCProposal for view {:?}: Leader for the current view: {:?} and sender: {:?}", quorum_proposal.data.view_number, leader, sender);
     }
 }
 
diff --git a/src/testing/basic_test.rs b/src/testing/basic_test.rs
index 96ff0120..34c8eedf 100644
--- a/src/testing/basic_test.rs
+++ b/src/testing/basic_test.rs
@@ -99,7 +99,6 @@ async fn test_builder() {
         let da_proposal = Arc::new(DaProposalMessage {
             view_number: ViewNumber::new(round as u64),
             txn_commitments,
-            num_nodes: NUM_STORAGE_NODES,
             sender: pub_key,
             builder_commitment: block_builder_commitment.clone(),
         });
diff --git a/src/testing/mod.rs b/src/testing/mod.rs
index 46bb130f..75983fc9 100644
--- a/src/testing/mod.rs
+++ b/src/testing/mod.rs
@@ -1,4 +1,4 @@
-use std::{hash::Hash, num::NonZeroUsize};
+use std::hash::Hash;
 
 use crate::{
     builder_state::{BuilderState, MessageType},
@@ -75,8 +75,6 @@ async fn start_builder_state(
         senders.transactions.clone(),
         genesis_vid_commitment,
         ViewNumber::genesis(),
-        ViewNumber::genesis(),
-        0,
     )));
 
     // instantiate the bootstrap builder state
@@ -86,7 +84,6 @@ async fn start_builder_state(
         bootstrap_receiver,
         Vec::new(),
         Arc::clone(&global_state),
-        NonZeroUsize::new(num_storage_nodes).unwrap(),
         Duration::from_millis(10), // max time to wait for non-zero txn block
         0,                         // base fee
         Arc::new(TestInstanceState::default()),
diff --git a/src/utils.rs b/src/utils.rs
index 181fb233..29224abe 100644
--- a/src/utils.rs
+++ b/src/utils.rs
@@ -1,13 +1,24 @@
 use std::{
     collections::HashSet,
+    future::Future,
     hash::Hash,
     mem,
+    pin::Pin,
+    task::Poll,
     time::{Duration, Instant},
 };
 
+use either::Either::{self, Left, Right};
+use futures::{FutureExt, Stream, StreamExt};
+use hotshot::types::Event;
+use hotshot_events_service::events::Error as EventStreamError;
 use hotshot_types::{
     traits::node_implementation::NodeType, utils::BuilderCommitment, vid::VidCommitment,
 };
+use surf_disco::Client;
+use tracing::error;
+use url::Url;
+use vbs::version::StaticVersionType;
 
 /// A set that allows for time-based garbage collection,
 /// implemented as three sets that are periodically shifted right.
@@ -110,3 +121,99 @@ impl<TYPES: NodeType> std::fmt::Display for BuilderStateId<TYPES> {
         )
     }
 }
+
+type EventServiceConnection<TYPES, V> = surf_disco::socket::Connection<
+    Event<TYPES>,
+    surf_disco::socket::Unsupported,
+    EventStreamError,
+    V,
+>;
+
+type EventServiceReconnect<TYPES, V> =
+    Pin<Box<dyn Future<Output = anyhow::Result<EventServiceConnection<TYPES, V>>> + Send + Sync>>;
+
+/// A wrapper around event streaming API that provides auto-reconnection capability
+pub struct EventServiceStream<TYPES: NodeType, V: StaticVersionType> {
+    api_url: Url,
+    connection: Either<EventServiceConnection<TYPES, V>, EventServiceReconnect<TYPES, V>>,
+}
+
+impl<TYPES: NodeType, V: StaticVersionType> EventServiceStream<TYPES, V> {
+    async fn connect_inner(
+        url: Url,
+    ) -> anyhow::Result<
+        surf_disco::socket::Connection<
+            Event<TYPES>,
+            surf_disco::socket::Unsupported,
+            EventStreamError,
+            V,
+        >,
+    > {
+        let client = Client::<hotshot_events_service::events::Error, V>::new(url.clone());
+
+        if !(client.connect(None).await) {
+            anyhow::bail!("Couldn't connect to API url");
+        }
+
+        tracing::info!("Builder client connected to the hotshot events api");
+
+        Ok(client
+            .socket("hotshot-events/events")
+            .subscribe::<Event<TYPES>>()
+            .await?)
+    }
+
+    /// Establish initial connection to the events service at `api_url`
+    pub async fn connect(api_url: Url) -> anyhow::Result<Self> {
+        let connection = Self::connect_inner(api_url.clone()).await?;
+
+        Ok(Self {
+            api_url,
+            connection: Left(connection),
+        })
+    }
+}
+
+impl<TYPES: NodeType, V: StaticVersionType + 'static> Stream for EventServiceStream<TYPES, V> {
+    type Item = Event<TYPES>;
+
+    fn poll_next(
+        mut self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        match &mut self.connection {
+            Left(connection) => {
+                let Poll::Ready(next) = connection.poll_next_unpin(cx) else {
+                    return Poll::Pending;
+                };
+                match next {
+                    Some(Ok(event)) => Poll::Ready(Some(event)),
+                    Some(Err(err)) => {
+                        error!("Error in the event stream: {err:?}");
+                        Poll::Pending
+                    }
+                    None => {
+                        let fut = Self::connect_inner(self.api_url.clone());
+                        let _ = std::mem::replace(&mut self.connection, Right(Box::pin(fut)));
+                        Poll::Pending
+                    }
+                }
+            }
+            Right(reconnect_future) => {
+                let Poll::Ready(ready) = reconnect_future.poll_unpin(cx) else {
+                    return Poll::Pending;
+                };
+                match ready {
+                    Ok(connection) => {
+                        let _ = std::mem::replace(&mut self.connection, Left(connection));
+                        Poll::Pending
+                    }
+                    Err(err) => {
+                        error!("Failed to reconnect to the event service: {err:?}");
+                        Poll::Ready(None)
+                    }
+                }
+            }
+        }
+    }
+}