diff --git a/Cargo.lock b/Cargo.lock
index 222f307bd..db03a3267 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -438,6 +438,7 @@ dependencies = [
  "alloy-pubsub",
  "alloy-rpc-client",
  "alloy-rpc-types-eth",
+ "alloy-rpc-types-txpool",
  "alloy-signer",
  "alloy-sol-types",
  "alloy-transport",
@@ -5548,9 +5549,11 @@ dependencies = [
  "futures",
  "futures-util",
  "jsonrpsee 0.25.1",
+ "jsonrpsee-types 0.25.1",
  "metrics",
  "op-alloy-consensus",
  "op-alloy-network",
+ "op-alloy-rpc-types",
  "op-alloy-rpc-types-engine",
  "op-revm",
  "parking_lot",
@@ -5586,6 +5589,7 @@ dependencies = [
  "reth-primitives-traits",
  "reth-provider",
  "reth-revm",
+ "reth-rpc-eth-types",
  "reth-rpc-layer",
  "reth-testing-utils",
  "reth-transaction-pool",
@@ -9788,7 +9792,7 @@ dependencies = [
 [[package]]
 name = "rollup-boost"
 version = "0.1.0"
-source = "git+http://github.com/flashbots/rollup-boost?rev=b9e6353d08672bd19e754a9525de47e04b34c84c#b9e6353d08672bd19e754a9525de47e04b34c84c"
+source = "git+http://github.com/flashbots/rollup-boost?rev=8506dfb7d84c65746f7c88d250983658438f59e8#8506dfb7d84c65746f7c88d250983658438f59e8"
 dependencies = [
  "alloy-primitives",
  "alloy-rpc-types-engine",
diff --git a/Cargo.toml b/Cargo.toml
index 75bfb0680..407ffbce3 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -76,6 +76,7 @@ reth-rpc-layer = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.1" }
 reth-network-peers = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.1" }
 reth-testing-utils = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.1" }
 reth-node-builder = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.1" }
+reth-rpc-eth-types =  { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.1" }
 
 # reth optimism
 reth-optimism-primitives = { git = "https://github.com/paradigmxyz/reth", tag = "v1.4.1" }
@@ -104,7 +105,7 @@ alloy-primitives = { version = "1.1.0", default-features = false }
 alloy-rlp = "0.3.10"
 alloy-chains = "0.2.0"
 alloy-evm = { version = "0.8.0", default-features = false }
-alloy-provider = { version = "1.0.3", features = ["ipc", "pubsub"] }
+alloy-provider = { version = "1.0.3", features = ["ipc", "pubsub", "txpool-api"] }
 alloy-pubsub = { version = "1.0.3" }
 alloy-eips = { version = "1.0.3" }
 alloy-rpc-types = { version = "1.0.3" }
diff --git a/crates/op-rbuilder/Cargo.toml b/crates/op-rbuilder/Cargo.toml
index e3a0ecb21..595beaf44 100644
--- a/crates/op-rbuilder/Cargo.toml
+++ b/crates/op-rbuilder/Cargo.toml
@@ -43,6 +43,7 @@ reth-network-peers.workspace = true
 reth-testing-utils.workspace = true
 reth-optimism-forks.workspace = true
 reth-node-builder.workspace = true
+reth-rpc-eth-types.workspace = true
 
 alloy-primitives.workspace = true
 alloy-consensus.workspace = true
@@ -61,6 +62,7 @@ alloy-serde.workspace = true
 alloy-op-evm.workspace = true
 op-alloy-consensus.workspace = true
 op-alloy-rpc-types-engine.workspace = true
+op-alloy-rpc-types.workspace = true
 op-alloy-network.workspace = true
 
 revm.workspace = true
@@ -73,6 +75,7 @@ serde.workspace = true
 secp256k1.workspace = true
 tokio.workspace = true
 jsonrpsee = { workspace = true }
+jsonrpsee-types.workspace = true
 async-trait = { workspace = true }
 clap_builder = { workspace = true }
 clap.workspace = true
diff --git a/crates/op-rbuilder/src/generator.rs b/crates/op-rbuilder/src/generator.rs
index 586aa313a..eefe62008 100644
--- a/crates/op-rbuilder/src/generator.rs
+++ b/crates/op-rbuilder/src/generator.rs
@@ -74,8 +74,6 @@ pub struct BlockPayloadJobGenerator<Client, Tasks, Builder> {
     last_payload: Arc<Mutex<CancellationToken>>,
     /// The extra block deadline in seconds
     extra_block_deadline: std::time::Duration,
-    /// Whether to enable revert protection
-    enable_revert_protection: bool,
 }
 
 // === impl EmptyBlockPayloadJobGenerator ===
@@ -90,7 +88,6 @@ impl<Client, Tasks, Builder> BlockPayloadJobGenerator<Client, Tasks, Builder> {
         builder: Builder,
         ensure_only_one_payload: bool,
         extra_block_deadline: std::time::Duration,
-        enable_revert_protection: bool,
     ) -> Self {
         Self {
             client,
@@ -100,7 +97,6 @@ impl<Client, Tasks, Builder> BlockPayloadJobGenerator<Client, Tasks, Builder> {
             ensure_only_one_payload,
             last_payload: Arc::new(Mutex::new(CancellationToken::new())),
             extra_block_deadline,
-            enable_revert_protection,
         }
     }
 }
@@ -186,7 +182,6 @@ where
             cancel: cancel_token,
             deadline,
             build_complete: None,
-            enable_revert_protection: self.enable_revert_protection,
         };
 
         job.spawn_build_job();
@@ -219,8 +214,6 @@ where
     pub(crate) cancel: CancellationToken,
     pub(crate) deadline: Pin<Box<Sleep>>, // Add deadline
     pub(crate) build_complete: Option<oneshot::Receiver<Result<(), PayloadBuilderError>>>,
-    /// Block building options
-    pub(crate) enable_revert_protection: bool,
 }
 
 impl<Tasks, Builder> PayloadJob for BlockPayloadJob<Tasks, Builder>
@@ -263,8 +256,6 @@ pub struct BuildArguments<Attributes, Payload: BuiltPayload> {
     pub config: PayloadConfig<Attributes, HeaderTy<Payload::Primitives>>,
     /// A marker that can be used to cancel the job.
     pub cancel: CancellationToken,
-    /// Whether to enable revert protection
-    pub enable_revert_protection: bool,
 }
 
 /// A [PayloadJob] is a future that's being polled by the `PayloadBuilderService`
@@ -280,7 +271,6 @@ where
         let payload_config = self.config.clone();
         let cell = self.cell.clone();
         let cancel = self.cancel.clone();
-        let enable_revert_protection = self.enable_revert_protection;
 
         let (tx, rx) = oneshot::channel();
         self.build_complete = Some(rx);
@@ -290,7 +280,6 @@ where
                 cached_reads: Default::default(),
                 config: payload_config,
                 cancel,
-                enable_revert_protection,
             };
 
             let result = builder.try_build(args, cell);
@@ -650,7 +639,6 @@ mod tests {
             builder.clone(),
             false,
             std::time::Duration::from_secs(1),
-            false,
         );
 
         // this is not nice but necessary
diff --git a/crates/op-rbuilder/src/main.rs b/crates/op-rbuilder/src/main.rs
index 0da7a7075..5477f1d27 100644
--- a/crates/op-rbuilder/src/main.rs
+++ b/crates/op-rbuilder/src/main.rs
@@ -1,7 +1,10 @@
 use args::CliExt;
 use clap::Parser;
 use reth_optimism_cli::{chainspec::OpChainSpecParser, Cli};
-use reth_optimism_node::{node::OpAddOnsBuilder, OpNode};
+use reth_optimism_node::{
+    node::{OpAddOnsBuilder, OpPoolBuilder},
+    OpNode,
+};
 use reth_transaction_pool::TransactionPool;
 
 /// CLI argument parsing.
@@ -9,11 +12,12 @@ pub mod args;
 pub mod generator;
 mod metrics;
 mod monitor_tx_pool;
-mod primitives;
-mod tx_signer;
-
 #[cfg(feature = "flashblocks")]
 pub mod payload_builder;
+mod primitives;
+mod revert_protection;
+mod tx;
+mod tx_signer;
 
 #[cfg(not(feature = "flashblocks"))]
 mod payload_builder_vanilla;
@@ -29,6 +33,8 @@ use metrics::{
     VERGEN_CARGO_FEATURES, VERGEN_CARGO_TARGET_TRIPLE, VERGEN_GIT_SHA,
 };
 use monitor_tx_pool::monitor_tx_pool;
+use revert_protection::{EthApiOverrideServer, RevertProtectionExt};
+use tx::FBPooledTransaction;
 
 // Prefer jemalloc for performance reasons.
 #[cfg(all(feature = "jemalloc", unix))]
@@ -53,20 +59,50 @@ fn main() {
             let op_node = OpNode::new(rollup_args.clone());
             let handle = builder
                 .with_types::<OpNode>()
-                .with_components(op_node.components().payload(CustomOpPayloadBuilder::new(
-                    builder_args.builder_signer,
-                    std::time::Duration::from_secs(builder_args.extra_block_deadline_secs),
-                    builder_args.enable_revert_protection,
-                    builder_args.flashblocks_ws_url,
-                    builder_args.chain_block_time,
-                    builder_args.flashblock_block_time,
-                )))
+                .with_components(
+                    op_node
+                        .components()
+                        .pool(
+                            OpPoolBuilder::<FBPooledTransaction>::default()
+                                .with_enable_tx_conditional(
+                                    // Revert protection uses the same internal pool logic as conditional transactions
+                                    // to garbage collect transactions out of the bundle range.
+                                    rollup_args.enable_tx_conditional
+                                        || builder_args.enable_revert_protection,
+                                )
+                                .with_supervisor(
+                                    rollup_args.supervisor_http.clone(),
+                                    rollup_args.supervisor_safety_level,
+                                ),
+                        )
+                        .payload(CustomOpPayloadBuilder::new(
+                            builder_args.builder_signer,
+                            std::time::Duration::from_secs(builder_args.extra_block_deadline_secs),
+                            builder_args.flashblocks_ws_url,
+                            builder_args.chain_block_time,
+                            builder_args.flashblock_block_time,
+                        )),
+                )
                 .with_add_ons(
                     OpAddOnsBuilder::default()
                         .with_sequencer(rollup_args.sequencer.clone())
                         .with_enable_tx_conditional(rollup_args.enable_tx_conditional)
                         .build(),
                 )
+                .extend_rpc_modules(move |ctx| {
+                    if builder_args.enable_revert_protection {
+                        tracing::info!("Revert protection enabled");
+
+                        let pool = ctx.pool().clone();
+                        let provider = ctx.provider().clone();
+                        let revert_protection_ext = RevertProtectionExt::new(pool, provider);
+
+                        ctx.modules
+                            .merge_configured(revert_protection_ext.into_rpc())?;
+                    }
+
+                    Ok(())
+                })
                 .on_node_started(move |ctx| {
                     version.register_version_metrics();
                     if builder_args.log_pool_transactions {
diff --git a/crates/op-rbuilder/src/metrics.rs b/crates/op-rbuilder/src/metrics.rs
index 87b795796..e3efa5e0c 100644
--- a/crates/op-rbuilder/src/metrics.rs
+++ b/crates/op-rbuilder/src/metrics.rs
@@ -34,6 +34,7 @@ pub struct OpRBuilderMetrics {
     #[cfg(feature = "flashblocks")]
     pub messages_sent_count: Counter,
     /// Total duration of building a block
+    #[cfg(feature = "flashblocks")]
     pub total_block_built_duration: Histogram,
     /// Flashblock build duration
     #[cfg(feature = "flashblocks")]
diff --git a/crates/op-rbuilder/src/monitor_tx_pool.rs b/crates/op-rbuilder/src/monitor_tx_pool.rs
index eba1cdde8..5bb3a4ea0 100644
--- a/crates/op-rbuilder/src/monitor_tx_pool.rs
+++ b/crates/op-rbuilder/src/monitor_tx_pool.rs
@@ -1,15 +1,15 @@
+use crate::tx::FBPooledTransaction;
 use futures_util::StreamExt;
-use reth_optimism_node::txpool::OpPooledTransaction;
 use reth_transaction_pool::{AllTransactionsEvents, FullTransactionEvent};
 use tracing::info;
 
-pub async fn monitor_tx_pool(mut new_transactions: AllTransactionsEvents<OpPooledTransaction>) {
+pub async fn monitor_tx_pool(mut new_transactions: AllTransactionsEvents<FBPooledTransaction>) {
     while let Some(event) = new_transactions.next().await {
         transaction_event_log(event);
     }
 }
 
-fn transaction_event_log(event: FullTransactionEvent<OpPooledTransaction>) {
+fn transaction_event_log(event: FullTransactionEvent<FBPooledTransaction>) {
     match event {
         FullTransactionEvent::Pending(hash) => {
             info!(
diff --git a/crates/op-rbuilder/src/payload_builder.rs b/crates/op-rbuilder/src/payload_builder.rs
index 791a56574..053165c36 100644
--- a/crates/op-rbuilder/src/payload_builder.rs
+++ b/crates/op-rbuilder/src/payload_builder.rs
@@ -100,14 +100,12 @@ pub struct CustomOpPayloadBuilder {
     chain_block_time: u64,
     flashblock_block_time: u64,
     extra_block_deadline: std::time::Duration,
-    enable_revert_protection: bool,
 }
 
 impl CustomOpPayloadBuilder {
     pub fn new(
         builder_signer: Option<Signer>,
         extra_block_deadline: std::time::Duration,
-        enable_revert_protection: bool,
         flashblocks_ws_url: String,
         chain_block_time: u64,
         flashblock_block_time: u64,
@@ -118,7 +116,6 @@ impl CustomOpPayloadBuilder {
             chain_block_time,
             flashblock_block_time,
             extra_block_deadline,
-            enable_revert_protection,
         }
     }
 }
@@ -185,7 +182,6 @@ where
     ) -> eyre::Result<PayloadBuilderHandle<<Node::Types as NodeTypes>::Payload>> {
         tracing::info!("Spawning a custom payload builder");
         let extra_block_deadline = self.extra_block_deadline;
-        let enable_revert_protection = self.enable_revert_protection;
         let payload_builder = self.build_payload_builder(ctx, pool, evm_config).await?;
         let payload_job_config = BasicPayloadJobGeneratorConfig::default();
 
@@ -196,7 +192,6 @@ where
             payload_builder,
             true,
             extra_block_deadline,
-            enable_revert_protection,
         );
 
         let (payload_service, payload_builder) =
diff --git a/crates/op-rbuilder/src/payload_builder_vanilla.rs b/crates/op-rbuilder/src/payload_builder_vanilla.rs
index e2b1e27c5..377aafce4 100644
--- a/crates/op-rbuilder/src/payload_builder_vanilla.rs
+++ b/crates/op-rbuilder/src/payload_builder_vanilla.rs
@@ -1,7 +1,8 @@
 use crate::{
-    generator::{BlockCell, BlockPayloadJobGenerator, BuildArguments, PayloadBuilder},
+    generator::BuildArguments,
     metrics::OpRBuilderMetrics,
     primitives::reth::ExecutionInfo,
+    tx::{FBPoolTransaction, MaybeRevertingTransaction},
     tx_signer::Signer,
 };
 use alloy_consensus::{
@@ -10,23 +11,17 @@ use alloy_consensus::{
 };
 use alloy_eips::{eip7685::EMPTY_REQUESTS_HASH, merge::BEACON_NONCE};
 use alloy_op_evm::block::receipt_builder::OpReceiptBuilder;
-use alloy_primitives::{private::alloy_rlp::Encodable, Address, Bytes, TxHash, TxKind, U256};
+use alloy_primitives::{private::alloy_rlp::Encodable, Address, Bytes, TxKind, U256};
 use alloy_rpc_types_engine::PayloadId;
 use alloy_rpc_types_eth::Withdrawals;
 use op_alloy_consensus::{OpDepositReceipt, OpTypedTransaction};
 use op_revm::OpSpecId;
 use reth::{
-    builder::{
-        components::{PayloadBuilderBuilder, PayloadServiceBuilder},
-        node::FullNodeTypes,
-        BuilderContext,
-    },
+    builder::{components::PayloadBuilderBuilder, node::FullNodeTypes, BuilderContext},
     core::primitives::InMemorySize,
-    payload::PayloadBuilderHandle,
 };
 use reth_basic_payload_builder::{
-    BasicPayloadJobGeneratorConfig, BuildOutcome, BuildOutcomeKind, MissingPayloadBehaviour,
-    PayloadConfig,
+    BuildOutcome, BuildOutcomeKind, MissingPayloadBehaviour, PayloadConfig,
 };
 use reth_chain_state::{ExecutedBlock, ExecutedBlockWithTrieUpdates};
 use reth_chainspec::{ChainSpecProvider, EthChainSpec, EthereumHardforks};
@@ -35,7 +30,7 @@ use reth_evm::{
     Database, Evm, EvmError, InvalidTxError,
 };
 use reth_execution_types::ExecutionOutcome;
-use reth_node_api::{NodePrimitives, NodeTypes, PrimitivesTy, TxTy};
+use reth_node_api::{NodePrimitives, NodeTypes, PrimitivesTy};
 use reth_node_builder::components::BasicPayloadServiceBuilder;
 use reth_optimism_chainspec::OpChainSpec;
 use reth_optimism_consensus::{calculate_receipt_root_no_memo_optimism, isthmus};
@@ -49,16 +44,14 @@ use reth_optimism_payload_builder::{
     OpPayloadPrimitives,
 };
 use reth_optimism_primitives::{OpPrimitives, OpReceipt, OpTransactionSigned};
-use reth_optimism_txpool::OpPooledTx;
-use reth_payload_builder::PayloadBuilderService;
 use reth_payload_builder_primitives::PayloadBuilderError;
 use reth_payload_primitives::PayloadBuilderAttributes;
 use reth_payload_util::{BestPayloadTransactions, NoopPayloadTransactions, PayloadTransactions};
 use reth_primitives::{BlockBody, SealedHeader};
 use reth_primitives_traits::{proofs, Block as _, RecoveredBlock, SignedTransaction};
 use reth_provider::{
-    CanonStateSubscriptions, HashedPostStateProvider, ProviderError, StateProviderFactory,
-    StateRootProvider, StorageRootProvider,
+    HashedPostStateProvider, ProviderError, StateProviderFactory, StateRootProvider,
+    StorageRootProvider,
 };
 use reth_revm::database::StateProviderDatabase;
 use reth_transaction_pool::{BestTransactionsAttributes, PoolTransaction, TransactionPool};
@@ -85,8 +78,8 @@ pub struct ExecutedPayload<N: NodePrimitives> {
 #[non_exhaustive]
 pub struct CustomOpPayloadBuilder {
     builder_signer: Option<Signer>,
+    #[allow(dead_code)]
     extra_block_deadline: std::time::Duration,
-    enable_revert_protection: bool,
     #[cfg(feature = "flashblocks")]
     flashblocks_ws_url: String,
     #[cfg(feature = "flashblocks")]
@@ -115,7 +108,6 @@ impl CustomOpPayloadBuilder {
     pub fn new(
         builder_signer: Option<Signer>,
         extra_block_deadline: std::time::Duration,
-        enable_revert_protection: bool,
         _flashblocks_ws_url: String,
         _chain_block_time: u64,
         _flashblock_block_time: u64,
@@ -123,7 +115,6 @@ impl CustomOpPayloadBuilder {
         BasicPayloadServiceBuilder::new(CustomOpPayloadBuilder {
             builder_signer,
             extra_block_deadline,
-            enable_revert_protection,
         })
     }
 }
@@ -137,10 +128,10 @@ where
             Primitives = OpPrimitives,
         >,
     >,
-    Pool: TransactionPool<Transaction: PoolTransaction<Consensus = TxTy<Node::Types>>>
+    Pool: TransactionPool<Transaction: FBPoolTransaction<Consensus = OpTransactionSigned>>
         + Unpin
         + 'static,
-    <Pool as TransactionPool>::Transaction: OpPooledTx,
+    <Pool as TransactionPool>::Transaction: FBPoolTransaction,
     Evm: ConfigureEvm<
             Primitives = PrimitivesTy<Node::Types>,
             NextBlockEnvCtx = OpNextBlockEnvAttributes,
@@ -159,67 +150,14 @@ where
             self.builder_signer,
             pool,
             ctx.provider().clone(),
-            self.enable_revert_protection,
         ))
     }
 }
 
-impl<Node, Pool, Evm> PayloadServiceBuilder<Node, Pool, Evm> for CustomOpPayloadBuilder
-where
-    Node: FullNodeTypes<
-        Types: NodeTypes<
-            Payload = OpEngineTypes,
-            ChainSpec = OpChainSpec,
-            Primitives = OpPrimitives,
-        >,
-    >,
-    Pool: TransactionPool<Transaction: PoolTransaction<Consensus = TxTy<Node::Types>>>
-        + Unpin
-        + 'static,
-    <Pool as TransactionPool>::Transaction: OpPooledTx,
-    Evm: ConfigureEvm<
-            Primitives = PrimitivesTy<Node::Types>,
-            NextBlockEnvCtx = OpNextBlockEnvAttributes,
-        > + 'static,
-{
-    async fn spawn_payload_builder_service(
-        self,
-        ctx: &BuilderContext<Node>,
-        pool: Pool,
-        evm_config: Evm,
-    ) -> eyre::Result<PayloadBuilderHandle<<Node::Types as NodeTypes>::Payload>> {
-        tracing::info!("Spawning a custom payload builder");
-        let extra_block_deadline = self.extra_block_deadline;
-        let enable_revert_protection = self.enable_revert_protection;
-        let payload_builder = self.build_payload_builder(ctx, pool, evm_config).await?;
-        let payload_job_config = BasicPayloadJobGeneratorConfig::default();
-
-        let payload_generator = BlockPayloadJobGenerator::with_builder(
-            ctx.provider().clone(),
-            ctx.task_executor().clone(),
-            payload_job_config,
-            payload_builder,
-            false,
-            extra_block_deadline,
-            enable_revert_protection,
-        );
-
-        let (payload_service, payload_builder) =
-            PayloadBuilderService::new(payload_generator, ctx.provider().canonical_state_stream());
-
-        ctx.task_executor()
-            .spawn_critical("custom payload builder service", Box::pin(payload_service));
-
-        tracing::info!("Custom payload service started");
-
-        Ok(payload_builder)
-    }
-}
-
 impl<Pool, Client, Txs> reth_basic_payload_builder::PayloadBuilder
     for OpPayloadBuilderVanilla<Pool, Client, Txs>
 where
-    Pool: TransactionPool<Transaction: PoolTransaction<Consensus = OpTransactionSigned>>,
+    Pool: TransactionPool<Transaction: FBPoolTransaction<Consensus = OpTransactionSigned>>,
     Client: StateProviderFactory + ChainSpecProvider<ChainSpec: EthChainSpec + OpHardforks> + Clone,
     Txs: OpPayloadTransactions<Pool::Transaction>,
 {
@@ -242,22 +180,14 @@ where
         let args = BuildArguments {
             cached_reads,
             config,
-            enable_revert_protection: self.enable_revert_protection,
             cancel: CancellationToken::new(),
         };
 
-        self.build_payload(
-            args,
-            |attrs| {
-                #[allow(clippy::unit_arg)]
-                self.best_transactions
-                    .best_transactions(pool.clone(), attrs)
-            },
-            |hashes| {
-                #[allow(clippy::unit_arg)]
-                self.best_transactions.remove_invalid(pool.clone(), hashes)
-            },
-        )
+        self.build_payload(args, |attrs| {
+            #[allow(clippy::unit_arg)]
+            self.best_transactions
+                .best_transactions(pool.clone(), attrs)
+        })
     }
 
     fn on_missing_payload(
@@ -278,13 +208,10 @@ where
             config,
             cached_reads: Default::default(),
             cancel: Default::default(),
-            enable_revert_protection: false,
         };
-        self.build_payload(
-            args,
-            |_| NoopPayloadTransactions::<Pool::Transaction>::default(),
-            |_| {},
-        )?
+        self.build_payload(args, |_| {
+            NoopPayloadTransactions::<Pool::Transaction>::default()
+        })?
         .into_payload()
         .ok_or_else(|| PayloadBuilderError::MissingPayload)
     }
@@ -308,8 +235,6 @@ pub struct OpPayloadBuilderVanilla<Pool, Client, Txs = ()> {
     pub best_transactions: Txs,
     /// The metrics for the builder
     pub metrics: OpRBuilderMetrics,
-    /// Whether we enable revert protection
-    pub enable_revert_protection: bool,
 }
 
 impl<Pool, Client> OpPayloadBuilderVanilla<Pool, Client> {
@@ -319,16 +244,8 @@ impl<Pool, Client> OpPayloadBuilderVanilla<Pool, Client> {
         builder_signer: Option<Signer>,
         pool: Pool,
         client: Client,
-        enable_revert_protection: bool,
     ) -> Self {
-        Self::with_builder_config(
-            evm_config,
-            builder_signer,
-            pool,
-            client,
-            Default::default(),
-            enable_revert_protection,
-        )
+        Self::with_builder_config(evm_config, builder_signer, pool, client, Default::default())
     }
 
     pub fn with_builder_config(
@@ -337,7 +254,6 @@ impl<Pool, Client> OpPayloadBuilderVanilla<Pool, Client> {
         pool: Pool,
         client: Client,
         config: OpBuilderConfig,
-        enable_revert_protection: bool,
     ) -> Self {
         Self {
             pool,
@@ -347,70 +263,13 @@ impl<Pool, Client> OpPayloadBuilderVanilla<Pool, Client> {
             best_transactions: (),
             metrics: Default::default(),
             builder_signer,
-            enable_revert_protection,
-        }
-    }
-}
-
-impl<Pool, Client, Txs> PayloadBuilder for OpPayloadBuilderVanilla<Pool, Client, Txs>
-where
-    Client: StateProviderFactory + ChainSpecProvider<ChainSpec: EthChainSpec + OpHardforks> + Clone,
-    Pool: TransactionPool<Transaction: PoolTransaction<Consensus = OpTransactionSigned>>,
-    Txs: OpPayloadTransactions<Pool::Transaction>,
-{
-    type Attributes = OpPayloadBuilderAttributes<OpTransactionSigned>;
-    type BuiltPayload = OpBuiltPayload;
-
-    fn try_build(
-        &self,
-        args: BuildArguments<Self::Attributes, Self::BuiltPayload>,
-        best_payload: BlockCell<Self::BuiltPayload>,
-    ) -> Result<(), PayloadBuilderError> {
-        let pool = self.pool.clone();
-        let block_build_start_time = Instant::now();
-
-        match self.build_payload(
-            args,
-            |attrs| {
-                #[allow(clippy::unit_arg)]
-                self.best_transactions
-                    .best_transactions(pool.clone(), attrs)
-            },
-            |hashes| {
-                #[allow(clippy::unit_arg)]
-                self.best_transactions.remove_invalid(pool.clone(), hashes)
-            },
-        )? {
-            BuildOutcome::Better { payload, .. } => {
-                best_payload.set(payload);
-                self.metrics
-                    .total_block_built_duration
-                    .record(block_build_start_time.elapsed());
-                self.metrics.block_built_success.increment(1);
-                Ok(())
-            }
-            BuildOutcome::Freeze(payload) => {
-                best_payload.set(payload);
-                self.metrics
-                    .total_block_built_duration
-                    .record(block_build_start_time.elapsed());
-                Ok(())
-            }
-            BuildOutcome::Cancelled => {
-                tracing::warn!("Payload build cancelled");
-                Err(PayloadBuilderError::MissingPayload)
-            }
-            _ => {
-                tracing::warn!("No better payload found");
-                Err(PayloadBuilderError::MissingPayload)
-            }
         }
     }
 }
 
 impl<Pool, Client, T> OpPayloadBuilderVanilla<Pool, Client, T>
 where
-    Pool: TransactionPool<Transaction: PoolTransaction<Consensus = OpTransactionSigned>>,
+    Pool: TransactionPool<Transaction: FBPoolTransaction<Consensus = OpTransactionSigned>>,
     Client: StateProviderFactory + ChainSpecProvider<ChainSpec: EthChainSpec + OpHardforks>,
 {
     /// Constructs an Optimism payload from the transactions sent via the
@@ -425,16 +284,14 @@ where
         &self,
         args: BuildArguments<OpPayloadBuilderAttributes<OpTransactionSigned>, OpBuiltPayload>,
         best: impl FnOnce(BestTransactionsAttributes) -> Txs + Send + Sync + 'a,
-        remove_reverted: impl FnOnce(Vec<TxHash>),
     ) -> Result<BuildOutcome<OpBuiltPayload>, PayloadBuilderError>
     where
-        Txs: PayloadTransactions<Transaction: PoolTransaction<Consensus = OpTransactionSigned>>,
+        Txs: PayloadTransactions<Transaction: FBPoolTransaction<Consensus = OpTransactionSigned>>,
     {
         let BuildArguments {
             mut cached_reads,
             config,
             cancel,
-            enable_revert_protection,
         } = args;
 
         let chain_spec = self.client.chain_spec();
@@ -475,11 +332,10 @@ where
             block_env_attributes,
             cancel,
             builder_signer: self.builder_signer,
-            metrics: Default::default(),
-            enable_revert_protection,
+            metrics: self.metrics.clone(),
         };
 
-        let builder = OpBuilder::new(best, remove_reverted);
+        let builder = OpBuilder::new(best);
 
         let state_provider = self.client.state_by_block_hash(ctx.parent().hash())?;
         let state = StateProviderDatabase::new(state_provider);
@@ -521,19 +377,12 @@ where
 pub struct OpBuilder<'a, Txs> {
     /// Yields the best transaction to include if transactions from the mempool are allowed.
     best: Box<dyn FnOnce(BestTransactionsAttributes) -> Txs + 'a>,
-    /// Removes reverted transactions from the tx pool
-    #[debug(skip)]
-    remove_invalid: Box<dyn FnOnce(Vec<TxHash>) + 'a>,
 }
 
 impl<'a, Txs> OpBuilder<'a, Txs> {
-    fn new(
-        best: impl FnOnce(BestTransactionsAttributes) -> Txs + Send + Sync + 'a,
-        remove_reverted: impl FnOnce(Vec<TxHash>) + 'a,
-    ) -> Self {
+    fn new(best: impl FnOnce(BestTransactionsAttributes) -> Txs + Send + Sync + 'a) -> Self {
         Self {
             best: Box::new(best),
-            remove_invalid: Box::new(remove_reverted),
         }
     }
 }
@@ -547,15 +396,12 @@ impl<Txs> OpBuilder<'_, Txs> {
     ) -> Result<BuildOutcomeKind<ExecutedPayload<N>>, PayloadBuilderError>
     where
         N: OpPayloadPrimitives<_TX = OpTransactionSigned>,
-        Txs: PayloadTransactions<Transaction: PoolTransaction<Consensus = N::SignedTx>>,
+        Txs: PayloadTransactions<Transaction: FBPoolTransaction<Consensus = OpTransactionSigned>>,
         ChainSpec: EthChainSpec + OpHardforks,
         DB: Database<Error = ProviderError> + AsRef<P>,
         P: StorageRootProvider,
     {
-        let Self {
-            best,
-            remove_invalid,
-        } = self;
+        let Self { best } = self;
         info!(target: "payload_builder", id=%ctx.payload_id(), parent_header = ?ctx.parent().hash(), parent_number = ctx.parent().number, "building new payload");
 
         // 1. apply pre-execution changes
@@ -636,10 +482,9 @@ impl<Txs> OpBuilder<'_, Txs> {
             .payload_num_tx
             .record(info.executed_transactions.len() as f64);
 
-        remove_invalid(info.invalid_tx_hashes.iter().copied().collect());
-
         let payload = ExecutedPayload { info };
 
+        ctx.metrics.block_built_success.increment(1);
         Ok(BuildOutcomeKind::Better { payload })
     }
 
@@ -651,7 +496,7 @@ impl<Txs> OpBuilder<'_, Txs> {
     ) -> Result<BuildOutcomeKind<OpBuiltPayload>, PayloadBuilderError>
     where
         ChainSpec: EthChainSpec + OpHardforks,
-        Txs: PayloadTransactions<Transaction: PoolTransaction<Consensus = OpTransactionSigned>>,
+        Txs: PayloadTransactions<Transaction: FBPoolTransaction<Consensus = OpTransactionSigned>>,
         DB: Database<Error = ProviderError> + AsRef<P>,
         P: StateRootProvider + HashedPostStateProvider + StorageRootProvider,
     {
@@ -813,13 +658,6 @@ pub trait OpPayloadTransactions<Transaction>: Clone + Send + Sync + Unpin + 'sta
         pool: Pool,
         attr: BestTransactionsAttributes,
     ) -> impl PayloadTransactions<Transaction = Transaction>;
-
-    /// Removes invalid transactions from the tx pool
-    fn remove_invalid<Pool: TransactionPool<Transaction = Transaction>>(
-        &self,
-        pool: Pool,
-        hashes: Vec<TxHash>,
-    );
 }
 
 impl<T: PoolTransaction> OpPayloadTransactions<T> for () {
@@ -830,14 +668,6 @@ impl<T: PoolTransaction> OpPayloadTransactions<T> for () {
     ) -> impl PayloadTransactions<Transaction = T> {
         BestPayloadTransactions::new(pool.best_transactions_with_attributes(attr))
     }
-
-    fn remove_invalid<Pool: TransactionPool<Transaction = T>>(
-        &self,
-        pool: Pool,
-        hashes: Vec<TxHash>,
-    ) {
-        pool.remove_transactions(hashes);
-    }
 }
 
 /// Container type that holds all necessities to build a new payload.
@@ -861,8 +691,6 @@ pub struct OpPayloadBuilderCtx<ChainSpec, N: NodePrimitives> {
     pub builder_signer: Option<Signer>,
     /// The metrics for the builder
     pub metrics: OpRBuilderMetrics,
-    /// Whether we enabled revert protection
-    pub enable_revert_protection: bool,
 }
 
 impl<ChainSpec, N> OpPayloadBuilderCtx<ChainSpec, N>
@@ -1124,7 +952,7 @@ where
         info: &mut ExecutionInfo<N>,
         db: &mut State<DB>,
         mut best_txs: impl PayloadTransactions<
-            Transaction: PoolTransaction<Consensus = OpTransactionSigned>,
+            Transaction: FBPoolTransaction<Consensus = OpTransactionSigned>,
         >,
         block_gas_limit: u64,
         block_da_limit: Option<u64>,
@@ -1142,6 +970,8 @@ where
         let mut evm = self.evm_config.evm_with_env(&mut *db, self.evm_env.clone());
 
         while let Some(tx) = best_txs.next(()) {
+            let exclude_reverting_txs = tx.exclude_reverting_txs();
+
             let tx = tx.into_consensus();
             num_txs_considered += 1;
             // ensure we still have capacity for this transaction
@@ -1195,10 +1025,9 @@ where
                 num_txs_simulated_success += 1;
             } else {
                 num_txs_simulated_fail += 1;
-                if self.enable_revert_protection {
+                if exclude_reverting_txs {
                     info!(target: "payload_builder", tx_hash = ?tx.tx_hash(), "skipping reverted transaction");
                     best_txs.mark_invalid(tx.signer(), tx.nonce());
-                    info.invalid_tx_hashes.insert(tx.tx_hash());
                     continue;
                 }
             }
diff --git a/crates/op-rbuilder/src/primitives/bundle.rs b/crates/op-rbuilder/src/primitives/bundle.rs
new file mode 100644
index 000000000..89c530716
--- /dev/null
+++ b/crates/op-rbuilder/src/primitives/bundle.rs
@@ -0,0 +1,26 @@
+use alloy_primitives::Bytes;
+use alloy_rpc_types_eth::erc4337::TransactionConditional;
+use serde::{Deserialize, Serialize};
+
+pub const MAX_BLOCK_RANGE_BLOCKS: u64 = 10;
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct Bundle {
+    #[serde(rename = "txs")]
+    pub transactions: Vec<Bytes>,
+
+    #[serde(rename = "maxBlockNumber")]
+    pub block_number_max: Option<u64>,
+}
+
+impl Bundle {
+    pub fn conditional(&self) -> TransactionConditional {
+        TransactionConditional {
+            block_number_min: None,
+            block_number_max: self.block_number_max,
+            known_accounts: Default::default(),
+            timestamp_max: None,
+            timestamp_min: None,
+        }
+    }
+}
diff --git a/crates/op-rbuilder/src/primitives/mod.rs b/crates/op-rbuilder/src/primitives/mod.rs
index 02615de6d..2af3ab8be 100644
--- a/crates/op-rbuilder/src/primitives/mod.rs
+++ b/crates/op-rbuilder/src/primitives/mod.rs
@@ -1 +1,3 @@
 pub mod reth;
+
+pub mod bundle;
diff --git a/crates/op-rbuilder/src/primitives/reth/execution.rs b/crates/op-rbuilder/src/primitives/reth/execution.rs
index 6ffdd2d11..3f5bf1774 100644
--- a/crates/op-rbuilder/src/primitives/reth/execution.rs
+++ b/crates/op-rbuilder/src/primitives/reth/execution.rs
@@ -1,9 +1,8 @@
 //! Heavily influenced by [reth](https://github.com/paradigmxyz/reth/blob/1e965caf5fa176f244a31c0d2662ba1b590938db/crates/optimism/payload/src/builder.rs#L570)
 use alloy_consensus::Transaction;
-use alloy_primitives::{private::alloy_rlp::Encodable, Address, TxHash, U256};
+use alloy_primitives::{private::alloy_rlp::Encodable, Address, U256};
 use reth_node_api::NodePrimitives;
 use reth_optimism_primitives::OpReceipt;
-use std::collections::HashSet;
 
 #[derive(Default, Debug)]
 pub struct ExecutionInfo<N: NodePrimitives> {
@@ -19,8 +18,6 @@ pub struct ExecutionInfo<N: NodePrimitives> {
     pub cumulative_da_bytes_used: u64,
     /// Tracks fees from executed mempool transactions
     pub total_fees: U256,
-    /// Tracks the reverted transaction hashes to remove from the transaction pool
-    pub invalid_tx_hashes: HashSet<TxHash>,
     #[cfg(feature = "flashblocks")]
     /// Index of the last consumed flashblock
     pub last_flashblock_index: usize,
@@ -36,7 +33,6 @@ impl<N: NodePrimitives> ExecutionInfo<N> {
             cumulative_gas_used: 0,
             cumulative_da_bytes_used: 0,
             total_fees: U256::ZERO,
-            invalid_tx_hashes: HashSet::new(),
             #[cfg(feature = "flashblocks")]
             last_flashblock_index: 0,
         }
diff --git a/crates/op-rbuilder/src/revert_protection.rs b/crates/op-rbuilder/src/revert_protection.rs
new file mode 100644
index 000000000..7f6f42e5c
--- /dev/null
+++ b/crates/op-rbuilder/src/revert_protection.rs
@@ -0,0 +1,97 @@
+use crate::{
+    primitives::bundle::{Bundle, MAX_BLOCK_RANGE_BLOCKS},
+    tx::{FBPooledTransaction, MaybeRevertingTransaction},
+};
+use alloy_primitives::B256;
+use jsonrpsee::{
+    core::{async_trait, RpcResult},
+    proc_macros::rpc,
+};
+use reth_optimism_txpool::{conditional::MaybeConditionalTransaction, OpPooledTransaction};
+use reth_provider::StateProviderFactory;
+use reth_rpc_eth_types::{utils::recover_raw_transaction, EthApiError};
+use reth_transaction_pool::{PoolTransaction, TransactionOrigin, TransactionPool};
+
+// Namespace overrides for revert protection support
+#[cfg_attr(not(test), rpc(server, namespace = "eth"))]
+#[cfg_attr(test, rpc(server, client, namespace = "eth"))]
+pub trait EthApiOverride {
+    #[method(name = "sendBundle")]
+    async fn send_bundle(&self, tx: Bundle) -> RpcResult<B256>;
+}
+
+pub struct RevertProtectionExt<Pool, Provider> {
+    pool: Pool,
+    provider: Provider,
+}
+
+impl<Pool, Provider> RevertProtectionExt<Pool, Provider> {
+    pub fn new(pool: Pool, provider: Provider) -> Self {
+        Self { pool, provider }
+    }
+}
+
+#[async_trait]
+impl<Pool, Provider> EthApiOverrideServer for RevertProtectionExt<Pool, Provider>
+where
+    Pool: TransactionPool<Transaction = FBPooledTransaction> + Clone + 'static,
+    Provider: StateProviderFactory + Send + Sync + Clone + 'static,
+{
+    async fn send_bundle(&self, mut bundle: Bundle) -> RpcResult<B256> {
+        let last_block_number = self
+            .provider
+            .best_block_number()
+            .map_err(|_e| EthApiError::InternalEthError)?;
+
+        // Only one transaction in the bundle is expected
+        let bundle_transaction = match bundle.transactions.len() {
+            0 => {
+                return Err(EthApiError::InvalidParams(
+                    "bundle must contain at least one transaction".into(),
+                )
+                .into());
+            }
+            1 => bundle.transactions[0].clone(),
+            _ => {
+                return Err(EthApiError::InvalidParams(
+                    "bundle must contain exactly one transaction".into(),
+                )
+                .into());
+            }
+        };
+
+        if let Some(block_number_max) = bundle.block_number_max {
+            // The max block cannot be a past block
+            if block_number_max <= last_block_number {
+                return Err(
+                    EthApiError::InvalidParams("block_number_max is a past block".into()).into(),
+                );
+            }
+
+            // Validate that it is not greater than the max_block_range
+            if block_number_max > last_block_number + MAX_BLOCK_RANGE_BLOCKS {
+                return Err(
+                    EthApiError::InvalidParams("block_number_max is too high".into()).into(),
+                );
+            }
+        } else {
+            // If no upper bound is set, use the maximum block range
+            bundle.block_number_max = Some(last_block_number + MAX_BLOCK_RANGE_BLOCKS);
+        }
+
+        let recovered = recover_raw_transaction(&bundle_transaction)?;
+        let mut pool_transaction: FBPooledTransaction =
+            OpPooledTransaction::from_pooled(recovered).into();
+
+        pool_transaction.set_exclude_reverting_txs(true);
+        pool_transaction.set_conditional(bundle.conditional());
+
+        let hash = self
+            .pool
+            .add_transaction(TransactionOrigin::Local, pool_transaction)
+            .await
+            .map_err(EthApiError::from)?;
+
+        Ok(hash)
+    }
+}
diff --git a/crates/op-rbuilder/src/tests/framework/harness.rs b/crates/op-rbuilder/src/tests/framework/harness.rs
index e5a0195d6..64c38d9ca 100644
--- a/crates/op-rbuilder/src/tests/framework/harness.rs
+++ b/crates/op-rbuilder/src/tests/framework/harness.rs
@@ -7,9 +7,9 @@ use super::{
 };
 use alloy_eips::BlockNumberOrTag;
 use alloy_network::Network;
-use alloy_primitives::hex;
+use alloy_primitives::{hex, B256};
 use alloy_provider::{
-    Identity, PendingTransactionBuilder, Provider, ProviderBuilder, RootProvider,
+    ext::TxPoolApi, Identity, PendingTransactionBuilder, Provider, ProviderBuilder, RootProvider,
 };
 use op_alloy_network::Optimism;
 use parking_lot::Mutex;
@@ -190,8 +190,72 @@ impl TestHarness {
         BUILDER_PRIVATE_KEY
     }
 
-    pub const fn builder_log_path(&self) -> &PathBuf {
-        &self.builder_log_path
+    pub async fn check_tx_in_pool(&self, tx_hash: B256) -> eyre::Result<TransactionStatus> {
+        let pool_inspect = self
+            .provider()
+            .expect("provider not available")
+            .txpool_content()
+            .await?;
+
+        let is_pending = pool_inspect.pending.iter().any(|pending_account_map| {
+            pending_account_map
+                .1
+                .iter()
+                .any(|(_, tx)| tx.as_recovered().hash() == *tx_hash)
+        });
+        if is_pending {
+            return Ok(TransactionStatus::Pending);
+        }
+
+        let is_queued = pool_inspect.queued.iter().any(|queued_account_map| {
+            queued_account_map
+                .1
+                .iter()
+                .any(|(_, tx)| tx.as_recovered().hash() == *tx_hash)
+        });
+        if is_queued {
+            return Ok(TransactionStatus::Queued);
+        }
+
+        // check that the builder emitted logs for the reverted transactions with the monitoring logic
+        // this will tell us whether the builder dropped the transaction
+        // TODO: this is not ideal, lets find a different way to detect this
+        // Each time a transaction is dropped, it emits a log like this
+        // Note that this does not tell us the reason why the transaction was dropped. Ideally
+        // we should know it at this point.
+        // 'Transaction event received target="monitoring" tx_hash="<tx_hash>" kind="discarded"'
+        let builder_logs = std::fs::read_to_string(&self.builder_log_path)?;
+        let txn_log = format!(
+            "Transaction event received target=\"monitoring\" tx_hash=\"{}\" kind=\"discarded\"",
+            tx_hash,
+        );
+        if builder_logs.contains(txn_log.as_str()) {
+            return Ok(TransactionStatus::Dropped);
+        }
+
+        Ok(TransactionStatus::NotFound)
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TransactionStatus {
+    NotFound,
+    Pending,
+    Queued,
+    Dropped,
+}
+
+impl TransactionStatus {
+    pub fn is_pending(&self) -> bool {
+        matches!(self, TransactionStatus::Pending)
+    }
+
+    pub fn is_queued(&self) -> bool {
+        matches!(self, TransactionStatus::Queued)
+    }
+
+    pub fn is_dropped(&self) -> bool {
+        matches!(self, TransactionStatus::Dropped)
     }
 }
 
diff --git a/crates/op-rbuilder/src/tests/framework/mod.rs b/crates/op-rbuilder/src/tests/framework/mod.rs
index 7ebab77fd..d8ffff3c0 100644
--- a/crates/op-rbuilder/src/tests/framework/mod.rs
+++ b/crates/op-rbuilder/src/tests/framework/mod.rs
@@ -15,6 +15,9 @@ pub use txs::*;
 const BUILDER_PRIVATE_KEY: &str =
     "0x59c6995e998f97a5a0044966f0945389dc9e86dae88c7a8412f4603b6b78690d";
 
+const FUNDED_PRIVATE_KEYS: &[&str] =
+    &["0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80"];
+
 pub const DEFAULT_JWT_TOKEN: &str =
     "688f5d737bad920bdfb2fc2f488d6b6209eebda1dae949a8de91398d932c517a";
 
diff --git a/crates/op-rbuilder/src/tests/framework/op.rs b/crates/op-rbuilder/src/tests/framework/op.rs
index 00a0f7f72..ea66f306c 100644
--- a/crates/op-rbuilder/src/tests/framework/op.rs
+++ b/crates/op-rbuilder/src/tests/framework/op.rs
@@ -137,7 +137,9 @@ impl Service for OpRbuilderConfig {
         if let Some(http_port) = self.http_port {
             cmd.arg("--http")
                 .arg("--http.port")
-                .arg(http_port.to_string());
+                .arg(http_port.to_string())
+                .arg("--http.api")
+                .arg("eth,web3,txpool");
         }
 
         if let Some(flashblocks_ws_url) = &self.flashblocks_ws_url {
diff --git a/crates/op-rbuilder/src/tests/framework/txs.rs b/crates/op-rbuilder/src/tests/framework/txs.rs
index 5c55e060f..b4049673f 100644
--- a/crates/op-rbuilder/src/tests/framework/txs.rs
+++ b/crates/op-rbuilder/src/tests/framework/txs.rs
@@ -1,7 +1,7 @@
-use crate::tx_signer::Signer;
+use crate::{primitives::bundle::Bundle, tx_signer::Signer};
 use alloy_consensus::TxEip1559;
 use alloy_eips::{eip2718::Encodable2718, BlockNumberOrTag};
-use alloy_primitives::Bytes;
+use alloy_primitives::{hex, Bytes};
 use alloy_provider::{PendingTransactionBuilder, Provider, RootProvider};
 use core::cmp::max;
 use op_alloy_consensus::{OpTxEnvelope, OpTypedTransaction};
@@ -10,7 +10,12 @@ use reth_primitives::Recovered;
 
 use alloy_eips::eip1559::MIN_PROTOCOL_BASE_FEE;
 
-use super::BUILDER_PRIVATE_KEY;
+use super::FUNDED_PRIVATE_KEYS;
+
+#[derive(Clone, Copy, Default)]
+pub struct BundleOpts {
+    pub block_number_max: Option<u64>,
+}
 
 #[derive(Clone)]
 pub struct TransactionBuilder {
@@ -19,6 +24,8 @@ pub struct TransactionBuilder {
     nonce: Option<u64>,
     base_fee: Option<u128>,
     tx: TxEip1559,
+    bundle_opts: Option<BundleOpts>,
+    key: Option<u64>,
 }
 
 impl TransactionBuilder {
@@ -33,9 +40,16 @@ impl TransactionBuilder {
                 gas_limit: 210000,
                 ..Default::default()
             },
+            bundle_opts: None,
+            key: None,
         }
     }
 
+    pub fn with_key(mut self, key: u64) -> Self {
+        self.key = Some(key);
+        self
+    }
+
     pub fn with_signer(mut self, signer: Signer) -> Self {
         self.signer = Some(signer);
         self
@@ -71,11 +85,21 @@ impl TransactionBuilder {
         self
     }
 
+    pub fn with_bundle(mut self, bundle_opts: BundleOpts) -> Self {
+        self.bundle_opts = Some(bundle_opts);
+        self
+    }
+
+    pub fn with_revert(mut self) -> Self {
+        self.tx.input = hex!("60006000fd").into();
+        self
+    }
+
     pub async fn build(mut self) -> Recovered<OpTxEnvelope> {
         let signer = match self.signer {
             Some(signer) => signer,
             None => Signer::try_from_secret(
-                BUILDER_PRIVATE_KEY
+                FUNDED_PRIVATE_KEYS[self.key.unwrap_or(0) as usize]
                     .parse()
                     .expect("invalid hardcoded builder private key"),
             )
@@ -118,10 +142,31 @@ impl TransactionBuilder {
     }
 
     pub async fn send(self) -> eyre::Result<PendingTransactionBuilder<Optimism>> {
+        let bundle_opts = self.bundle_opts.clone();
         let provider = self.provider.clone();
         let transaction = self.build().await;
+        let transaction_encoded = transaction.encoded_2718();
+
+        if let Some(bundle_opts) = bundle_opts {
+            // Send the transaction as a bundle with the bundle options
+            let bundle = Bundle {
+                transactions: vec![transaction_encoded.into()],
+                block_number_max: bundle_opts.block_number_max,
+            };
+
+            let tx_hash = provider
+                .client()
+                .request("eth_sendBundle", (bundle,))
+                .await?;
+
+            return Ok(PendingTransactionBuilder::new(
+                provider.root().clone(),
+                tx_hash,
+            ));
+        }
+
         Ok(provider
-            .send_raw_transaction(transaction.encoded_2718().as_slice())
+            .send_raw_transaction(transaction_encoded.as_slice())
             .await?)
     }
 }
diff --git a/crates/op-rbuilder/src/tests/vanilla/revert.rs b/crates/op-rbuilder/src/tests/vanilla/revert.rs
index 67d4177e3..f4710612c 100644
--- a/crates/op-rbuilder/src/tests/vanilla/revert.rs
+++ b/crates/op-rbuilder/src/tests/vanilla/revert.rs
@@ -1,51 +1,61 @@
-use crate::tests::TestHarnessBuilder;
+use alloy_provider::PendingTransactionBuilder;
+use op_alloy_network::Optimism;
 
-/// This test ensures that the transactions that get reverted an not included in the block
-/// are emitted as a log on the builder.
+use crate::{
+    primitives::bundle::MAX_BLOCK_RANGE_BLOCKS,
+    tests::{BundleOpts, TestHarness, TestHarnessBuilder},
+};
+
+/// This test ensures that the transactions that get reverted and not included in the block,
+/// are eventually dropped from the pool once their block range is reached.
+/// This test creates N transactions with different block ranges.
 #[tokio::test]
-async fn monitor_transaction_drops() -> eyre::Result<()> {
-    let harness = TestHarnessBuilder::new("monitor_transaction_drops")
+async fn revert_protection_monitor_transaction_gc() -> eyre::Result<()> {
+    let harness = TestHarnessBuilder::new("revert_protection_monitor_transaction_gc")
         .with_revert_protection()
         .build()
         .await?;
 
     let mut generator = harness.block_generator().await?;
 
-    // send 10 reverting transactions
+    // send 10 bundles with different block ranges
     let mut pending_txn = Vec::new();
-    for _ in 0..10 {
-        let txn = harness.send_revert_transaction().await?;
+    for i in 1..=10 {
+        let txn = harness
+            .create_transaction()
+            .with_revert()
+            .with_bundle(BundleOpts {
+                block_number_max: Some(i),
+            })
+            .send()
+            .await?;
         pending_txn.push(txn);
     }
 
     // generate 10 blocks
-    for _ in 0..10 {
-        generator.generate_block().await?;
-        let latest_block = harness.latest_block().await;
+    for i in 0..10 {
+        let generated_block = generator.generate_block().await?;
 
         // blocks should only include two transactions (deposit + builder)
-        assert_eq!(latest_block.transactions.len(), 2);
-    }
-
-    // check that the builder emitted logs for the reverted transactions
-    // with the monitoring logic
-    // TODO: this is not ideal, lets find a different way to detect this
-    // Each time a transaction is dropped, it emits a log like this
-    // 'Transaction event received target="monitoring" tx_hash="<tx_hash>" kind="discarded"'
-    let builder_logs = std::fs::read_to_string(harness.builder_log_path())?;
-
-    for txn in pending_txn {
-        let txn_log = format!(
-            "Transaction event received target=\"monitoring\" tx_hash=\"{}\" kind=\"discarded\"",
-            txn.tx_hash()
-        );
-
-        assert!(builder_logs.contains(txn_log.as_str()));
+        assert_eq!(generated_block.block.transactions.len(), 2);
+
+        // since we created the 10 transactions with increasing block ranges, as we generate blocks
+        // one transaction will be gc on each block.
+        // transactions from [0, i] should be dropped, transactions from [i+1, 10] should be queued
+        for j in 0..=i {
+            let status = harness.check_tx_in_pool(*pending_txn[j].tx_hash()).await?;
+            assert!(status.is_dropped());
+        }
+        for j in i + 1..10 {
+            let status = harness.check_tx_in_pool(*pending_txn[j].tx_hash()).await?;
+            assert!(status.is_queued());
+        }
     }
 
     Ok(())
 }
 
+/// If revert protection is disabled, the transactions that revert are included in the block.
 #[tokio::test]
 async fn revert_protection_disabled() -> eyre::Result<()> {
     let harness = TestHarnessBuilder::new("revert_protection_disabled")
@@ -66,22 +76,154 @@ async fn revert_protection_disabled() -> eyre::Result<()> {
     Ok(())
 }
 
+/// If revert protection is disabled, it should not be possible to send a revert bundle
+/// since the revert RPC endpoint is not available.
+#[tokio::test]
+async fn revert_protection_disabled_bundle_endpoint_error() -> eyre::Result<()> {
+    let harness = TestHarnessBuilder::new("revert_protection_disabled_bundle_endpoint_error")
+        .build()
+        .await?;
+
+    let res = harness
+        .create_transaction()
+        .with_bundle(BundleOpts::default())
+        .send()
+        .await;
+
+    assert!(
+        res.is_err(),
+        "Expected error because method is not available"
+    );
+    Ok(())
+}
+
+/// Test the behaviour of the revert protection bundle, if the bundle **does not** revert
+/// the transaction is included in the block. If the bundle reverts, the transaction
+/// is not included in the block and tried again for the next bundle range blocks
+/// when it will be dropped from the pool.
+#[tokio::test]
+async fn revert_protection_bundle() -> eyre::Result<()> {
+    let harness = TestHarnessBuilder::new("revert_protection_bundle")
+        .with_revert_protection()
+        .build()
+        .await?;
+
+    let mut generator = harness.block_generator().await?; // Block 1
+
+    // Test 1: Bundle does not revert
+    let valid_bundle = harness
+        .create_transaction()
+        .with_bundle(BundleOpts::default())
+        .send()
+        .await?;
+
+    let block_generated = generator.generate_block().await?; // Block 2
+    assert!(block_generated.includes(*valid_bundle.tx_hash()));
+
+    let bundle_opts = BundleOpts {
+        block_number_max: Some(4),
+    };
+
+    let reverted_bundle = harness
+        .create_transaction()
+        .with_revert()
+        .with_bundle(bundle_opts)
+        .send()
+        .await?;
+
+    // Test 2: Bundle reverts. It is not included in the block
+    let block_generated = generator.generate_block().await?; // Block 3
+    assert!(block_generated.not_includes(*reverted_bundle.tx_hash()));
+
+    // After the block the transaction is still pending in the pool
+    assert!(harness
+        .check_tx_in_pool(*reverted_bundle.tx_hash())
+        .await?
+        .is_pending());
+
+    // Test 3: Chain progresses beyond the bundle range. The transaction is dropped from the pool
+    generator.generate_block().await?; // Block 4
+    assert!(harness
+        .check_tx_in_pool(*reverted_bundle.tx_hash())
+        .await?
+        .is_pending());
+
+    generator.generate_block().await?; // Block 5
+    assert!(harness
+        .check_tx_in_pool(*reverted_bundle.tx_hash())
+        .await?
+        .is_dropped());
+
+    Ok(())
+}
+
+/// Test the range limits for the revert protection bundle.
 #[tokio::test]
-async fn revert_protection() -> eyre::Result<()> {
-    let harness = TestHarnessBuilder::new("revert_protection")
+async fn revert_protection_bundle_range_limits() -> eyre::Result<()> {
+    let harness = TestHarnessBuilder::new("revert_protection_bundle_range_limits")
         .with_revert_protection()
         .build()
         .await?;
 
     let mut generator = harness.block_generator().await?;
 
+    // Advance two blocks and try to send a bundle with max block = 1
+    generator.generate_block().await?; // Block 1
+    generator.generate_block().await?; // Block 2
+
+    async fn send_bundle(
+        harness: &TestHarness,
+        block_number_max: u64,
+    ) -> eyre::Result<PendingTransactionBuilder<Optimism>> {
+        harness
+            .create_transaction()
+            .with_bundle(BundleOpts {
+                block_number_max: Some(block_number_max),
+            })
+            .send()
+            .await
+    }
+
+    // Max block cannot be a past block
+    assert!(send_bundle(&harness, 1).await.is_err());
+
+    // Bundles are valid if their max block in in between the current block and the max block range
+    let next_valid_block = 3;
+
+    for i in next_valid_block..next_valid_block + MAX_BLOCK_RANGE_BLOCKS {
+        assert!(send_bundle(&harness, i).await.is_ok());
+    }
+
+    // A bundle with a block out of range is invalid
+    assert!(
+        send_bundle(&harness, next_valid_block + MAX_BLOCK_RANGE_BLOCKS + 1)
+            .await
+            .is_err()
+    );
+
+    Ok(())
+}
+
+/// If a transaction reverts and was sent as a normal transaction through the eth_sendRawTransaction
+/// bundle, the transaction should be included in the block.
+/// This behaviour is the same as the 'revert_protection_disabled' test.
+#[tokio::test]
+async fn revert_protection_allow_reverted_transactions_without_bundle() -> eyre::Result<()> {
+    let harness =
+        TestHarnessBuilder::new("revert_protection_allow_reverted_transactions_without_bundle")
+            .with_revert_protection()
+            .build()
+            .await?;
+
+    let mut generator = harness.block_generator().await?;
+
     for _ in 0..10 {
         let valid_tx = harness.send_valid_transaction().await?;
         let reverting_tx = harness.send_revert_transaction().await?;
         let block_generated = generator.generate_block().await?;
 
         assert!(block_generated.includes(*valid_tx.tx_hash()));
-        assert!(block_generated.not_includes(*reverting_tx.tx_hash()));
+        assert!(block_generated.includes(*reverting_tx.tx_hash()));
     }
 
     Ok(())
diff --git a/crates/op-rbuilder/src/tx.rs b/crates/op-rbuilder/src/tx.rs
new file mode 100644
index 000000000..023c51231
--- /dev/null
+++ b/crates/op-rbuilder/src/tx.rs
@@ -0,0 +1,259 @@
+use std::sync::Arc;
+
+use alloy_consensus::{
+    conditional::BlockConditionalAttributes, BlobTransactionSidecar, BlobTransactionValidationError,
+};
+use alloy_eips::{eip7702::SignedAuthorization, Typed2718};
+use alloy_primitives::{Address, Bytes, TxHash, TxKind, B256, U256};
+use alloy_rpc_types_eth::{erc4337::TransactionConditional, AccessList};
+use reth_optimism_primitives::OpTransactionSigned;
+use reth_optimism_txpool::{
+    conditional::MaybeConditionalTransaction, estimated_da_size::DataAvailabilitySized,
+    interop::MaybeInteropTransaction, OpPooledTransaction,
+};
+use reth_primitives::{kzg::KzgSettings, Recovered};
+use reth_primitives_traits::InMemorySize;
+use reth_transaction_pool::{EthBlobTransactionSidecar, EthPoolTransaction, PoolTransaction};
+
+pub trait FBPoolTransaction:
+    EthPoolTransaction
+    + MaybeInteropTransaction
+    + MaybeConditionalTransaction
+    + DataAvailabilitySized
+    + MaybeRevertingTransaction
+{
+}
+
+#[derive(Clone, Debug)]
+pub struct FBPooledTransaction {
+    pub inner: OpPooledTransaction,
+    pub exclude_reverting_txs: bool,
+}
+
+impl FBPoolTransaction for FBPooledTransaction {}
+
+pub trait MaybeRevertingTransaction {
+    fn set_exclude_reverting_txs(&mut self, exclude: bool);
+    fn exclude_reverting_txs(&self) -> bool;
+}
+
+impl MaybeRevertingTransaction for FBPooledTransaction {
+    fn set_exclude_reverting_txs(&mut self, exclude: bool) {
+        self.exclude_reverting_txs = exclude;
+    }
+
+    fn exclude_reverting_txs(&self) -> bool {
+        self.exclude_reverting_txs
+    }
+}
+
+impl InMemorySize for FBPooledTransaction {
+    fn size(&self) -> usize {
+        self.inner.size() + core::mem::size_of::<bool>()
+    }
+}
+
+impl PoolTransaction for FBPooledTransaction {
+    type TryFromConsensusError =
+        <op_alloy_consensus::OpPooledTransaction as TryFrom<OpTransactionSigned>>::Error;
+    type Consensus = OpTransactionSigned;
+    type Pooled = op_alloy_consensus::OpPooledTransaction;
+
+    fn clone_into_consensus(&self) -> Recovered<Self::Consensus> {
+        self.inner.clone_into_consensus()
+    }
+
+    fn into_consensus(self) -> Recovered<Self::Consensus> {
+        self.inner.into_consensus()
+    }
+
+    fn from_pooled(tx: Recovered<Self::Pooled>) -> Self {
+        let inner = OpPooledTransaction::from_pooled(tx);
+        Self {
+            inner,
+            exclude_reverting_txs: false,
+        }
+    }
+
+    fn hash(&self) -> &TxHash {
+        self.inner.hash()
+    }
+
+    fn sender(&self) -> Address {
+        self.inner.sender()
+    }
+
+    fn sender_ref(&self) -> &Address {
+        self.inner.sender_ref()
+    }
+
+    fn cost(&self) -> &U256 {
+        self.inner.cost()
+    }
+
+    fn encoded_length(&self) -> usize {
+        self.inner.encoded_length()
+    }
+}
+
+impl Typed2718 for FBPooledTransaction {
+    fn ty(&self) -> u8 {
+        self.inner.ty()
+    }
+}
+
+impl alloy_consensus::Transaction for FBPooledTransaction {
+    fn chain_id(&self) -> Option<u64> {
+        self.inner.chain_id()
+    }
+
+    fn nonce(&self) -> u64 {
+        self.inner.nonce()
+    }
+
+    fn gas_limit(&self) -> u64 {
+        self.inner.gas_limit()
+    }
+
+    fn gas_price(&self) -> Option<u128> {
+        self.inner.gas_price()
+    }
+
+    fn max_fee_per_gas(&self) -> u128 {
+        self.inner.max_fee_per_gas()
+    }
+
+    fn max_priority_fee_per_gas(&self) -> Option<u128> {
+        self.inner.max_priority_fee_per_gas()
+    }
+
+    fn max_fee_per_blob_gas(&self) -> Option<u128> {
+        self.inner.max_fee_per_blob_gas()
+    }
+
+    fn priority_fee_or_price(&self) -> u128 {
+        self.inner.priority_fee_or_price()
+    }
+
+    fn effective_gas_price(&self, base_fee: Option<u64>) -> u128 {
+        self.inner.effective_gas_price(base_fee)
+    }
+
+    fn is_dynamic_fee(&self) -> bool {
+        self.inner.is_dynamic_fee()
+    }
+
+    fn kind(&self) -> TxKind {
+        self.inner.kind()
+    }
+
+    fn is_create(&self) -> bool {
+        self.inner.is_create()
+    }
+
+    fn value(&self) -> U256 {
+        self.inner.value()
+    }
+
+    fn input(&self) -> &Bytes {
+        self.inner.input()
+    }
+
+    fn access_list(&self) -> Option<&AccessList> {
+        self.inner.access_list()
+    }
+
+    fn blob_versioned_hashes(&self) -> Option<&[B256]> {
+        self.inner.blob_versioned_hashes()
+    }
+
+    fn authorization_list(&self) -> Option<&[SignedAuthorization]> {
+        self.inner.authorization_list()
+    }
+}
+
+impl EthPoolTransaction for FBPooledTransaction {
+    fn take_blob(&mut self) -> EthBlobTransactionSidecar {
+        EthBlobTransactionSidecar::None
+    }
+
+    fn try_into_pooled_eip4844(
+        self,
+        sidecar: Arc<BlobTransactionSidecar>,
+    ) -> Option<Recovered<Self::Pooled>> {
+        self.inner.try_into_pooled_eip4844(sidecar)
+    }
+
+    fn try_from_eip4844(
+        _tx: Recovered<Self::Consensus>,
+        _sidecar: BlobTransactionSidecar,
+    ) -> Option<Self> {
+        None
+    }
+
+    fn validate_blob(
+        &self,
+        _sidecar: &BlobTransactionSidecar,
+        _settings: &KzgSettings,
+    ) -> Result<(), BlobTransactionValidationError> {
+        Err(BlobTransactionValidationError::NotBlobTransaction(
+            self.ty(),
+        ))
+    }
+}
+
+impl MaybeInteropTransaction for FBPooledTransaction {
+    fn interop_deadline(&self) -> Option<u64> {
+        self.inner.interop_deadline()
+    }
+
+    fn set_interop_deadline(&self, deadline: u64) {
+        self.inner.set_interop_deadline(deadline);
+    }
+
+    fn with_interop_deadline(self, interop: u64) -> Self
+    where
+        Self: Sized,
+    {
+        self.inner.with_interop_deadline(interop).into()
+    }
+}
+
+impl DataAvailabilitySized for FBPooledTransaction {
+    fn estimated_da_size(&self) -> u64 {
+        self.inner.estimated_da_size()
+    }
+}
+
+impl From<OpPooledTransaction> for FBPooledTransaction {
+    fn from(tx: OpPooledTransaction) -> Self {
+        Self {
+            inner: tx,
+            exclude_reverting_txs: false,
+        }
+    }
+}
+
+impl MaybeConditionalTransaction for FBPooledTransaction {
+    fn set_conditional(&mut self, conditional: TransactionConditional) {
+        self.inner.set_conditional(conditional);
+    }
+
+    fn conditional(&self) -> Option<&TransactionConditional> {
+        self.inner.conditional()
+    }
+
+    fn has_exceeded_block_attributes(&self, block_attr: &BlockConditionalAttributes) -> bool {
+        self.inner.has_exceeded_block_attributes(block_attr)
+    }
+
+    fn with_conditional(self, conditional: TransactionConditional) -> Self
+    where
+        Self: Sized,
+    {
+        FBPooledTransaction {
+            inner: self.inner.with_conditional(conditional),
+            exclude_reverting_txs: self.exclude_reverting_txs,
+        }
+    }
+}