diff --git a/crates/messaging/src/controller.rs b/crates/messaging/src/controller.rs
new file mode 100644
index 000000000..62a749e88
--- /dev/null
+++ b/crates/messaging/src/controller.rs
@@ -0,0 +1,255 @@
+//! RPC-facing checkpoint controller for the messaging service.
+//!
+//! Encapsulates DB-side checkpoint operations and signalling the running drain
+//! task to live-rewind the in-memory cursor without restarting the node.
+
+use anyhow::Context;
+use katana_provider::api::messaging::{
+    MessagingCheckpoint, MessagingCheckpointProvider, MessagingL1ToL2IndexWriter,
+};
+use katana_provider::{MutableProvider, ProviderFactory, ProviderRW, ProviderResult};
+use tokio::sync::mpsc;
+use tracing::warn;
+
+use crate::LOG_TARGET;
+
+/// Signal sent from the controller to the drain task: rewind the in-memory
+/// cursor to `(from_block, from_tx_index)`.
+#[derive(Debug, Clone, Copy)]
+pub struct RewindSignal {
+    pub from_block: u64,
+    pub from_tx_index: u64,
+}
+
+/// Operator-facing handle to the messaging checkpoint.
+///
+/// Reads/writes the persisted DB checkpoint and signals the running drain
+/// task to rewind its in-memory cursor.
+#[derive(Debug, Clone)]
+pub struct MessagingController<P> {
+    provider: P,
+    default_from_block: u64,
+    rewind_tx: mpsc::Sender<RewindSignal>,
+}
+
+impl<P> MessagingController<P> {
+    pub(crate) fn new(
+        provider: P,
+        default_from_block: u64,
+        rewind_tx: mpsc::Sender<RewindSignal>,
+    ) -> Self {
+        Self { provider, default_from_block, rewind_tx }
+    }
+}
+
+impl<P> MessagingController<P>
+where
+    P: ProviderFactory + Clone + Send + Sync + 'static,
+    <P as ProviderFactory>::ProviderMut:
+        ProviderRW + MessagingCheckpointProvider + MessagingL1ToL2IndexWriter + MutableProvider,
+{
+    /// Read the last *committed* checkpoint — the same value `resume_cursor`
+    /// reads on boot.
+    pub fn get_checkpoint(&self) -> ProviderResult<Option<MessagingCheckpoint>> {
+        let db_tx = self.provider.provider_mut();
+        let cp = db_tx.messaging_checkpoint()?;
+        MutableProvider::commit(db_tx)?;
+        Ok(cp)
+    }
+
+    /// Persist `(block, tx_index)` as the last processed checkpoint and signal
+    /// the drain task to rewind the in-memory cursor to `(block, tx_index + 1)`.
+    pub async fn set_checkpoint(&self, block: u64, tx_index: u64) -> anyhow::Result<()> {
+        let db_tx = self.provider.provider_mut();
+        db_tx
+            .set_messaging_checkpoint(&MessagingCheckpoint { block, tx_index })
+            .context("set messaging checkpoint")?;
+        MutableProvider::commit(db_tx).context("commit checkpoint write")?;
+
+        // The DB write is the source of truth — a failed channel send (server
+        // not running, or already stopped) is logged but does not fail the call.
+        // The next `start()` will resume from the new value.
+        //
+        // Why `saturating_add`: a `tx_index` of `u64::MAX` is degenerate but the
+        // RPC accepts any `u64`. Without saturation, debug builds would panic and
+        // release builds would silently wrap to 0 (re-gathering the whole block).
+        let signal = RewindSignal { from_block: block, from_tx_index: tx_index.saturating_add(1) };
+        if let Err(error) = self.rewind_tx.send(signal).await {
+            warn!(target: LOG_TARGET, %error, "Failed to send rewind signal; DB checkpoint persisted, will be picked up on next start.");
+        }
+
+        Ok(())
+    }
+
+    /// Delete the persisted checkpoint and signal the drain task to rewind to
+    /// the configured `default_from_block` (the value used by `resume_cursor`
+    /// when no checkpoint exists).
+    pub async fn reset_checkpoint(&self) -> anyhow::Result<()> {
+        let db_tx = self.provider.provider_mut();
+        db_tx.delete_messaging_checkpoint().context("delete messaging checkpoint")?;
+        MutableProvider::commit(db_tx).context("commit checkpoint delete")?;
+
+        let signal = RewindSignal { from_block: self.default_from_block, from_tx_index: 0 };
+        if let Err(error) = self.rewind_tx.send(signal).await {
+            warn!(target: LOG_TARGET, %error, "Failed to send rewind signal; DB checkpoint deleted, will be picked up on next start.");
+        }
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use katana_provider::DbProviderFactory;
+    use tokio::sync::mpsc;
+
+    use super::*;
+
+    fn setup() -> (MessagingController<DbProviderFactory>, mpsc::Receiver<RewindSignal>) {
+        let provider = DbProviderFactory::new_in_memory();
+        let (tx, rx) = mpsc::channel(1);
+        (MessagingController::new(provider, 7, tx), rx)
+    }
+
+    #[tokio::test]
+    async fn get_checkpoint_returns_none_when_absent() {
+        let (controller, _rx) = setup();
+        let cp = controller.get_checkpoint().unwrap();
+        assert!(cp.is_none());
+    }
+
+    #[tokio::test]
+    async fn set_checkpoint_persists_value_and_emits_rewind_signal_at_tx_index_plus_one() {
+        let (controller, mut rx) = setup();
+
+        controller.set_checkpoint(100, 5).await.unwrap();
+
+        let cp = controller.get_checkpoint().unwrap().expect("checkpoint persisted");
+        assert_eq!(cp.block, 100);
+        assert_eq!(cp.tx_index, 5);
+
+        let signal = rx.try_recv().expect("rewind signal sent");
+        assert_eq!(signal.from_block, 100);
+        // The DB checkpoint records the last *processed* message; the live
+        // cursor must resume one past it.
+        assert_eq!(signal.from_tx_index, 6);
+    }
+
+    #[tokio::test]
+    async fn reset_checkpoint_deletes_row_and_emits_default_from_block_signal() {
+        let (controller, mut rx) = setup();
+
+        controller.set_checkpoint(42, 9).await.unwrap();
+        let _ = rx.try_recv().expect("set signal");
+
+        controller.reset_checkpoint().await.unwrap();
+
+        let cp = controller.get_checkpoint().unwrap();
+        assert!(cp.is_none(), "row deleted");
+
+        let signal = rx.try_recv().expect("reset signal sent");
+        assert_eq!(signal.from_block, 7, "default_from_block snapshotted at construction");
+        assert_eq!(signal.from_tx_index, 0);
+    }
+
+    /// A failed channel send must not fail the call — the DB write is the
+    /// source of truth and the next start picks up the value.
+    #[tokio::test]
+    async fn set_and_reset_succeed_when_receiver_dropped() {
+        let provider = DbProviderFactory::new_in_memory();
+        let (tx, rx) = mpsc::channel(1);
+        drop(rx);
+        let controller = MessagingController::new(provider, 0, tx);
+
+        controller.set_checkpoint(1, 2).await.expect("set succeeds with dropped receiver");
+        controller.reset_checkpoint().await.expect("reset succeeds with dropped receiver");
+    }
+
+    /// Operator-rewind path: a later `setCheckpoint` to a *lower* `(block, tx_index)`
+    /// must overwrite the prior higher value. This is the canonical "I want to
+    /// re-gather from a known earlier point" use case.
+    #[tokio::test]
+    async fn set_checkpoint_overrides_prior_higher_checkpoint() {
+        let (controller, mut rx) = setup();
+
+        controller.set_checkpoint(100, 50).await.unwrap();
+        // Drain the first signal so we can inspect the second cleanly.
+        let _ = rx.try_recv().expect("first signal");
+
+        controller.set_checkpoint(20, 0).await.unwrap();
+
+        let cp = controller.get_checkpoint().unwrap().expect("checkpoint persisted");
+        assert_eq!(cp.block, 20);
+        assert_eq!(cp.tx_index, 0);
+
+        let signal = rx.try_recv().expect("second signal sent");
+        assert_eq!(signal.from_block, 20);
+        assert_eq!(signal.from_tx_index, 1);
+    }
+
+    /// `tx_index == u64::MAX` must not panic. The published rewind signal saturates
+    /// to `u64::MAX` rather than wrapping (which would silently re-gather block 0).
+    #[tokio::test]
+    async fn set_checkpoint_at_tx_index_max_saturates() {
+        let (controller, mut rx) = setup();
+
+        controller.set_checkpoint(5, u64::MAX).await.expect("no panic on u64::MAX");
+
+        let cp = controller.get_checkpoint().unwrap().expect("checkpoint persisted");
+        assert_eq!(cp.block, 5);
+        assert_eq!(cp.tx_index, u64::MAX);
+
+        let signal = rx.try_recv().expect("signal sent");
+        assert_eq!(signal.from_block, 5);
+        assert_eq!(signal.from_tx_index, u64::MAX, "saturating_add must not wrap");
+    }
+
+    /// `resetCheckpoint` on a fresh DB must succeed (the DB delete is a no-op)
+    /// AND emit the default-from-block rewind signal.
+    #[tokio::test]
+    async fn reset_checkpoint_is_idempotent_when_no_row_exists() {
+        let (controller, mut rx) = setup();
+
+        controller.reset_checkpoint().await.expect("reset on fresh DB is a no-op");
+
+        let cp = controller.get_checkpoint().unwrap();
+        assert!(cp.is_none(), "no row materialized");
+
+        let signal = rx.try_recv().expect("reset signal sent");
+        assert_eq!(signal.from_block, 7, "default_from_block from setup()");
+        assert_eq!(signal.from_tx_index, 0);
+    }
+
+    /// The controller snapshots `default_from_block` at construction; reset must
+    /// emit *that* value, independent of any prior set/get traffic.
+    #[tokio::test]
+    async fn reset_checkpoint_uses_snapshot_default_from_block() {
+        let provider = DbProviderFactory::new_in_memory();
+        let (tx, mut rx) = mpsc::channel(1);
+        let controller = MessagingController::new(provider, 42, tx);
+
+        controller.reset_checkpoint().await.unwrap();
+
+        let signal = rx.try_recv().expect("reset signal");
+        assert_eq!(signal.from_block, 42, "snapshot of construction-time default_from_block");
+        assert_eq!(signal.from_tx_index, 0);
+    }
+
+    /// `get_checkpoint` reads what was *committed*, including writes that bypass
+    /// the controller (e.g., the messaging drain task). Companion to the existing
+    /// "returns None when absent" test.
+    #[tokio::test]
+    async fn get_checkpoint_reads_committed_value() {
+        let provider = DbProviderFactory::new_in_memory();
+        let (tx, _rx) = mpsc::channel(1);
+        let controller = MessagingController::new(provider.clone(), 0, tx);
+
+        let db_tx = provider.provider_mut();
+        db_tx.set_messaging_checkpoint(&MessagingCheckpoint { block: 77, tx_index: 9 }).unwrap();
+        MutableProvider::commit(db_tx).unwrap();
+
+        let cp = controller.get_checkpoint().unwrap().expect("controller observes committed write");
+        assert_eq!(cp.block, 77);
+        assert_eq!(cp.tx_index, 9);
+    }
+}
diff --git a/crates/messaging/src/lib.rs b/crates/messaging/src/lib.rs
index 3a4343127..7ffe78073 100644
--- a/crates/messaging/src/lib.rs
+++ b/crates/messaging/src/lib.rs
@@ -10,9 +10,10 @@
 //!   interval, block subscription, etc).
 //!
 //! These are composed by [`stream::MessageStream`] into a [`Stream`] that yields
-//! [`MessagingOutcome`] items. The stream is consumed by [`server::MessagingServer`]
+//! [`MessagingOutcome`] items. The stream is consumed by [`service::MessagingService`]
 //! which adds transactions to the pool and persists checkpoints.
 
+pub mod controller;
 pub mod service;
 pub mod stream;
 
@@ -22,6 +23,7 @@ use serde::{Deserialize, Serialize};
 pub use service::{MessagingService, MessagingServiceHandle};
 use url::Url;
 
+pub use crate::controller::{MessagingController, RewindSignal};
 use crate::stream::collector::OrderedMessage;
 
 pub(crate) const LOG_TARGET: &str = "messaging";
@@ -42,8 +44,12 @@ pub struct MessagingOutcome {
 /// gathered from a settlement chain.
 ///
 /// This trait is object-safe, allowing `Box<dyn Messenger>` usage.
-pub trait Messenger: Stream<Item = MessagingOutcome> + Send + Unpin {}
-impl<T> Messenger for T where T: Stream<Item = MessagingOutcome> + Send + Unpin {}
+pub trait Messenger: Stream<Item = MessagingOutcome> + Send + Unpin {
+    /// Rewind the in-memory cursor to `(from_block, from_tx_index)`. Any
+    /// in-flight gather is abandoned — the next trigger tick re-gathers from
+    /// the new cursor.
+    fn rewind(&mut self, from_block: u64, from_tx_index: u64);
+}
 
 /// The config used to initialize the messaging service.
 #[derive(Debug, Deserialize, Clone, Serialize, PartialEq, Eq)]
diff --git a/crates/messaging/src/service.rs b/crates/messaging/src/service.rs
index 3e3381311..9d9ac675d 100644
--- a/crates/messaging/src/service.rs
+++ b/crates/messaging/src/service.rs
@@ -1,4 +1,6 @@
-use anyhow::Context;
+use std::sync::Mutex;
+
+use anyhow::{anyhow, Context};
 use futures::StreamExt;
 use katana_pool::api::TransactionPool;
 use katana_pool::TxPool;
@@ -8,38 +10,35 @@ use katana_provider::api::messaging::{
     MessagingCheckpoint, MessagingCheckpointProvider, MessagingL1ToL2IndexWriter,
 };
 use katana_provider::{MutableProvider, ProviderFactory, ProviderRW};
-use tokio::sync::oneshot;
+use tokio::sync::{mpsc, oneshot};
 use tokio::task::JoinHandle;
 use tracing::{info, warn};
 
+use crate::controller::{MessagingController, RewindSignal};
 use crate::stream::collector::ethereum::EthereumCollector;
 use crate::stream::collector::starknet::StarknetCollector;
 use crate::stream::trigger::IntervalTrigger;
 use crate::stream::MessageStream;
 use crate::{MessagingOutcome, Messenger, SettlementChainConfig, LOG_TARGET};
 
-/// Identifier used to namespace the persisted messaging checkpoint within the
-/// shared `MessagingCheckpoints` table.
-const CHECKPOINT_ID: &str = "messaging";
-
 /// Default poll interval (in seconds) between gather ticks.
 const DEFAULT_INTERVAL: u64 = 2;
 
-/// The messaging server.
+/// The messaging service.
 ///
 /// [`Self::start`] is non-consuming and mirrors `RpcServer::start`. It reads the
-/// resume checkpoint, builds the messenger, and spawns the drain loop. A
-/// settlement chain must be configured via [`settlement`](Self::settlement)
-/// before calling [`start`](Self::start); otherwise it returns an error.
+/// resume checkpoint, builds the messenger, and spawns the drain loop. Calling
+/// [`start`](Self::start) more than once returns an error: the rewind receiver
+/// is single-consumer and is taken on the first call.
 ///
-/// The server depends directly on the provider factory `P` for both reading the
-/// resume checkpoint at boot and atomically persisting the L1->L2 index entry +
-/// checkpoint after each successful pool insert.
+/// The service depends directly on the provider factory `P` for both reading
+/// the resume checkpoint at boot and atomically persisting the L1->L2 index
+/// entry + checkpoint after each successful pool insert.
 ///
-/// Configure the server using the builder-style setters
-/// ([`settlement`](Self::settlement), [`interval`](Self::interval),
-/// [`from_block`](Self::from_block), [`confirmation_depth`](Self::confirmation_depth))
-/// before calling [`start`](Self::start).
+/// Configure with builder-style setters
+/// ([`interval`](Self::interval), [`from_block`](Self::from_block),
+/// [`confirmation_depth`](Self::confirmation_depth)) before calling
+/// [`start`](Self::start). The settlement chain is required at construction.
 pub struct MessagingService<P, Pl = TxPool> {
     chain_id: ChainId,
     pool: Pl,
@@ -49,19 +48,26 @@ pub struct MessagingService<P, Pl = TxPool> {
     interval: u64,
     from_block: u64,
     confirmation_depth: u64,
+
+    /// Sender shared with all controllers; cloning is cheap.
+    rewind_tx: mpsc::Sender<RewindSignal>,
+    /// Owned by the service until [`start`](Self::start) takes it. Clones of the
+    /// service get an empty mutex so only the original can drive the drain loop.
+    /// Wrapped in a `Mutex` because `start` is `&self`.
+    rewind_rx: Mutex<Option<mpsc::Receiver<RewindSignal>>>,
 }
 
 impl<P, Pl> MessagingService<P, Pl> {
-    /// Create a new messaging server with no settlement configured.
-    ///
-    /// A settlement chain must be set via [`settlement`](Self::settlement)
-    /// before [`start`](Self::start) can be called.
+    /// Create a new messaging service for the given settlement chain.
     pub fn new(
         chain_id: ChainId,
         pool: Pl,
         provider: P,
         settlement: SettlementChainConfig,
     ) -> Self {
+        // Capacity 1 with `send().await` gives natural back-pressure on rapid
+        // rewinds; operator-issued resets don't burst.
+        let (rewind_tx, rewind_rx) = mpsc::channel(1);
         Self {
             chain_id,
             pool,
@@ -70,6 +76,8 @@ impl<P, Pl> MessagingService<P, Pl> {
             interval: DEFAULT_INTERVAL,
             from_block: 0,
             confirmation_depth: 0,
+            rewind_tx,
+            rewind_rx: Mutex::new(Some(rewind_rx)),
         }
     }
 
@@ -102,12 +110,31 @@ where
         ProviderRW + MessagingCheckpointProvider + MessagingL1ToL2IndexWriter + MutableProvider,
     Pl: TransactionPool<Transaction = ExecutableTxWithHash> + Clone + Send + Sync + 'static,
 {
-    /// Start the messaging server.
+    /// Returns a [`MessagingController`] that can read/write the persisted
+    /// checkpoint and signal a live rewind to a running drain task.
+    ///
+    /// Snapshots the configured `from_block` as the controller's default — call
+    /// [`from_block`](Self::from_block) before this.
+    pub fn controller(&self) -> MessagingController<P>
+    where
+        P: Clone,
+    {
+        MessagingController::new(self.provider.clone(), self.from_block, self.rewind_tx.clone())
+    }
+
+    /// Start the messaging service.
     ///
     /// Reads the resume checkpoint, builds the messenger, and spawns the drain
-    /// loop. Returns an error if no settlement chain has been configured via
-    /// [`settlement`](Self::settlement).
+    /// loop. Returns an error if `start()` has already been called on this
+    /// instance (the rewind receiver is single-consumer).
     pub fn start(&self) -> Result<MessagingServiceHandle, anyhow::Error> {
+        let mut rewind_rx = self
+            .rewind_rx
+            .lock()
+            .map_err(|_| anyhow!("rewind receiver mutex poisoned"))?
+            .take()
+            .ok_or_else(|| anyhow!("messaging service already started"))?;
+
         let (from_block, from_tx_index) = resume_cursor(&self.provider, self.from_block)?;
 
         let trigger = IntervalTrigger::new(self.interval);
@@ -146,6 +173,24 @@ where
 
             loop {
                 tokio::select! {
+                    // Shutdown takes priority over both gather and rewind so
+                    // a stop signal can't be starved by busy work.
+                    biased;
+
+                    _ = &mut shutdown => {
+                        break;
+                    }
+
+                    Some(sig) = rewind_rx.recv() => {
+                        info!(
+                            target: LOG_TARGET,
+                            from_block = sig.from_block,
+                            from_tx_index = sig.from_tx_index,
+                            "Rewinding messenger cursor.",
+                        );
+                        messenger.rewind(sig.from_block, sig.from_tx_index);
+                    }
+
                     outcome = messenger.next() => {
                         match outcome {
                             None => break, // Stream ended
@@ -222,10 +267,6 @@ where
                             }
                         }
                     }
-
-                    _ = &mut shutdown => {
-                        break;
-                    }
                 }
             }
         });
@@ -246,7 +287,7 @@ where
     <P as ProviderFactory>::ProviderMut: MessagingCheckpointProvider + MutableProvider,
 {
     let db_tx = provider.provider_mut();
-    let cp = db_tx.messaging_checkpoint(CHECKPOINT_ID).context("read messaging checkpoint")?;
+    let cp = db_tx.messaging_checkpoint().context("read messaging checkpoint")?;
     db_tx.commit().context("commit checkpoint read tx")?;
 
     Ok(match cp {
@@ -257,7 +298,7 @@ where
 
 impl<P, Pl> std::fmt::Debug for MessagingService<P, Pl> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("MessagingServer").finish_non_exhaustive()
+        f.debug_struct("MessagingService").finish_non_exhaustive()
     }
 }
 
@@ -271,6 +312,11 @@ impl<P: Clone, Pl: Clone> Clone for MessagingService<P, Pl> {
             interval: self.interval,
             from_block: self.from_block,
             confirmation_depth: self.confirmation_depth,
+            rewind_tx: self.rewind_tx.clone(),
+            // Clones share the sender but cannot be started — the receiver is
+            // not cloneable and only the original service can drive the drain
+            // loop.
+            rewind_rx: Mutex::new(None),
         }
     }
 }
@@ -291,7 +337,7 @@ where
 {
     let db_tx = provider.provider_mut();
     db_tx.record_l1_to_l2(l1_tx_hash, l2_tx_hash)?;
-    db_tx.set_messaging_checkpoint(CHECKPOINT_ID, &MessagingCheckpoint { block, tx_index })?;
+    db_tx.set_messaging_checkpoint(&MessagingCheckpoint { block, tx_index })?;
     db_tx.commit()?;
     Ok(())
 }
@@ -304,7 +350,7 @@ pub struct MessagingServiceHandle {
 
 impl std::fmt::Debug for MessagingServiceHandle {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("MessagingHandle").finish_non_exhaustive()
+        f.debug_struct("MessagingServiceHandle").finish_non_exhaustive()
     }
 }
 
@@ -324,12 +370,39 @@ impl MessagingServiceHandle {
 
 #[cfg(test)]
 mod tests {
+    use std::time::Duration;
+
+    use katana_pool::ordering::FiFo;
+    use katana_pool::pool::Pool;
+    use katana_pool::validation::NoopValidator;
+    use katana_primitives::transaction::ExecutableTxWithHash;
     use katana_primitives::Felt;
     use katana_provider::api::messaging::MessagingL1ToL2IndexProvider;
     use katana_provider::DbProviderFactory;
+    use url::Url;
 
     use super::*;
 
+    /// No-op pool used by the lifecycle tests. The drain task never actually inserts
+    /// transactions in these tests (the configured settlement endpoint is unroutable),
+    /// so this type only needs to satisfy the trait bounds of `start()`.
+    type NoopPool =
+        Pool<ExecutableTxWithHash, NoopValidator<ExecutableTxWithHash>, FiFo<ExecutableTxWithHash>>;
+
+    fn noop_pool() -> NoopPool {
+        Pool::new(NoopValidator::new(), FiFo::new())
+    }
+
+    /// Settlement config pointing at a non-routable URL. The drain task may try
+    /// `latest_block()` against this; it'll fail or pend, which is fine — the
+    /// lifecycle tests don't depend on any successful gather.
+    fn unroutable_settlement() -> SettlementChainConfig {
+        SettlementChainConfig::Ethereum {
+            rpc_url: Url::parse("http://127.0.0.1:1/").unwrap(),
+            contract_address: Default::default(),
+        }
+    }
+
     #[test]
     fn resume_cursor_falls_back_to_default_from_block_when_no_checkpoint_persisted() {
         let provider = DbProviderFactory::new_in_memory();
@@ -346,12 +419,7 @@ mod tests {
 
         // Persist a checkpoint marking message at (block=100, tx_index=5) as fully processed.
         let db_tx = provider.provider_mut();
-        db_tx
-            .set_messaging_checkpoint(
-                CHECKPOINT_ID,
-                &MessagingCheckpoint { block: 100, tx_index: 5 },
-            )
-            .unwrap();
+        db_tx.set_messaging_checkpoint(&MessagingCheckpoint { block: 100, tx_index: 5 }).unwrap();
         db_tx.commit().unwrap();
 
         // The default from_block is intentionally far below the persisted checkpoint —
@@ -378,8 +446,7 @@ mod tests {
 
         let db_tx = provider.provider_mut();
         let mapped = db_tx.l2_txs_for_l1(&l1).unwrap();
-        let cp =
-            db_tx.messaging_checkpoint(CHECKPOINT_ID).unwrap().expect("checkpoint should exist");
+        let cp = db_tx.messaging_checkpoint().unwrap().expect("checkpoint should exist");
         db_tx.commit().unwrap();
 
         assert_eq!(mapped, vec![l2], "L1->L2 index entry should be written");
@@ -440,10 +507,102 @@ mod tests {
         commit_message(&provider, &l1, l2, 10, 2).unwrap();
 
         let db_tx = provider.provider_mut();
-        let cp = db_tx.messaging_checkpoint(CHECKPOINT_ID).unwrap().expect("checkpoint");
+        let cp = db_tx.messaging_checkpoint().unwrap().expect("checkpoint");
         db_tx.commit().unwrap();
 
         assert_eq!(cp.block, 10, "checkpoint should reflect the latest committed message");
         assert_eq!(cp.tx_index, 2);
     }
+
+    // -------------------------------------------------------------------------
+    // Lifecycle tests
+    //
+    // These exercise `MessagingService::start` itself — that the rewind_rx is
+    // single-take, that clones can't be started, and that a closed rewind
+    // channel doesn't kill the drain task. They use a non-routable settlement
+    // endpoint; the drain task never produces work but stays alive, which is
+    // all the lifecycle invariants require.
+    // -------------------------------------------------------------------------
+
+    /// The `rewind_rx` is taken on first `start()`; a second call on the same
+    /// instance must fail with a clear "already started" error rather than
+    /// silently spawning a second drain task that competes for rewind signals.
+    #[tokio::test]
+    async fn start_twice_returns_error() {
+        let provider = DbProviderFactory::new_in_memory();
+        let pool = noop_pool();
+        let server =
+            MessagingService::new(ChainId::default(), pool, provider, unroutable_settlement())
+                .interval(60);
+
+        let mut handle = server.start().expect("first start succeeds");
+
+        let err = server.start().expect_err("second start must fail");
+        let msg = err.to_string();
+        assert!(msg.contains("already started"), "expected 'already started' in error, got: {msg}");
+
+        // Clean up the first task so the test process exits cleanly.
+        handle.stop();
+        handle.stopped().await;
+    }
+
+    /// `Clone for MessagingService` deliberately sets `rewind_rx: None` on the
+    /// clone so only the original instance can drive the drain loop. Starting
+    /// a clone must fail with the same error as a double-start.
+    #[tokio::test]
+    async fn clone_cannot_be_started() {
+        let provider = DbProviderFactory::new_in_memory();
+        let pool = noop_pool();
+        let server =
+            MessagingService::new(ChainId::default(), pool, provider, unroutable_settlement())
+                .interval(60);
+
+        let clone = server.clone();
+
+        let err = clone.start().expect_err("starting a clone must fail");
+        let msg = err.to_string();
+        assert!(msg.contains("already started"), "expected 'already started' in error, got: {msg}");
+
+        // The original is still startable (rewind_rx wasn't taken from it).
+        let mut handle = server.start().expect("original is still startable after cloning");
+        handle.stop();
+        handle.stopped().await;
+    }
+
+    /// Dropping the controller (and hence one rewind_tx sender) must not kill
+    /// the running drain task. The other arms of the `select!` (shutdown,
+    /// messenger.next) keep firing; the rewind arm just goes permanently
+    /// inactive once all senders are gone. This guards against a regression
+    /// where the loop would exit on `rewind_rx.recv() == None`.
+    #[tokio::test]
+    async fn rewind_sender_dropped_does_not_kill_task() {
+        let provider = DbProviderFactory::new_in_memory();
+        let pool = noop_pool();
+        let server =
+            MessagingService::new(ChainId::default(), pool, provider, unroutable_settlement())
+                .interval(60);
+
+        let controller = server.controller();
+        let mut handle = server.start().expect("start succeeds");
+
+        // Drop everything that holds a rewind_tx clone: the controller, the
+        // server itself (its own sender), so the receiver inside the task
+        // observes a fully-closed channel.
+        drop(controller);
+        drop(server);
+
+        // Give the runtime a moment to deliver the channel-closed notification
+        // to the task. Tokio's `mpsc::Receiver::recv` returns `None` once all
+        // senders are dropped, and the `select!` arm with that pattern simply
+        // never matches again — the other arms must keep working.
+        tokio::time::sleep(Duration::from_millis(50)).await;
+
+        assert!(
+            !handle.task_handle.is_finished(),
+            "drain task must survive a closed rewind channel"
+        );
+
+        handle.stop();
+        handle.stopped().await;
+    }
 }
diff --git a/crates/messaging/src/stream/mod.rs b/crates/messaging/src/stream/mod.rs
index a6a77777d..f27d0e131 100644
--- a/crates/messaging/src/stream/mod.rs
+++ b/crates/messaging/src/stream/mod.rs
@@ -15,7 +15,7 @@ pub mod trigger;
 use collector::{GatherResult, MessageCollector};
 use trigger::MessageTrigger;
 
-use crate::{MessagingOutcome, LOG_TARGET};
+use crate::{MessagingOutcome, Messenger, LOG_TARGET};
 
 /// Maximum number of blocks to fetch in a single gather call.
 const MAX_BLOCKS_PER_GATHER: u64 = 200;
@@ -237,6 +237,22 @@ where
     }
 }
 
+impl<C, T> Messenger for MessageStream<C, T>
+where
+    C: MessageCollector + 'static,
+    T: MessageTrigger,
+{
+    fn rewind(&mut self, from_block: u64, from_tx_index: u64) {
+        self.from_block = from_block;
+        self.from_tx_index = from_tx_index;
+        // Resetting `phase` to `Idle` abandons any in-flight gather/checking
+        // future built from the old cursor. The next trigger tick rebuilds from
+        // the new cursor — any abandoned blocks get re-fetched, and the pool's
+        // hash-level dedup absorbs duplicate inserts.
+        self.phase = MessageStreamPhase::Idle;
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::collections::VecDeque;
@@ -252,6 +268,7 @@ mod tests {
     use katana_primitives::Felt;
     use parking_lot::Mutex;
     use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
+    use tokio::sync::oneshot;
 
     use super::collector::OrderedMessage;
     use super::*;
@@ -273,6 +290,13 @@ mod tests {
         pub chain_id: ChainId,
     }
 
+    /// A queued response: either resolves immediately, or pends on a oneshot
+    /// that the test fires when it wants the underlying future to complete.
+    enum Response<T> {
+        Ready(Result<T, MockCollectorError>),
+        Pending(oneshot::Receiver<Result<T, MockCollectorError>>),
+    }
+
     /// A [`MessageCollector`] backed by canned response queues.
     ///
     /// Each test pushes responses in expected order via [`push_latest_block`] and
@@ -280,12 +304,17 @@ mod tests {
     /// the method returns [`MockCollectorError`] so tests fail loudly when they
     /// haven't enqueued enough responses.
     ///
+    /// For tests that need to hold a future across a state change (rewind, drop),
+    /// use [`push_latest_block_pending`] / [`push_gather_pending`]: each returns a
+    /// `oneshot::Sender` the test fires when it wants the underlying future to
+    /// resolve.
+    ///
     /// All calls are recorded for assertions via [`latest_block_calls`] and
     /// [`gather_calls`].
     #[derive(Default)]
     pub struct MockCollector {
-        latest_block_responses: Mutex<VecDeque<Result<u64, MockCollectorError>>>,
-        gather_responses: Mutex<VecDeque<Result<GatherResult, MockCollectorError>>>,
+        latest_block_responses: Mutex<VecDeque<Response<u64>>>,
+        gather_responses: Mutex<VecDeque<Response<GatherResult>>>,
         latest_block_call_count: AtomicU64,
         gather_calls: Mutex<Vec<GatherCall>>,
     }
@@ -297,12 +326,32 @@ mod tests {
 
         /// Push a `latest_block` response onto the queue. Called in FIFO order.
         pub fn push_latest_block(&self, response: Result<u64, MockCollectorError>) {
-            self.latest_block_responses.lock().push_back(response);
+            self.latest_block_responses.lock().push_back(Response::Ready(response));
         }
 
         /// Push a `gather` response onto the queue. Called in FIFO order.
         pub fn push_gather(&self, response: Result<GatherResult, MockCollectorError>) {
-            self.gather_responses.lock().push_back(response);
+            self.gather_responses.lock().push_back(Response::Ready(response));
+        }
+
+        /// Push a pending `latest_block` response. The returned sender resolves the
+        /// in-flight future when fired; dropping it without sending makes the future
+        /// permanently pending (used to test future-abandonment paths).
+        pub fn push_latest_block_pending(
+            &self,
+        ) -> oneshot::Sender<Result<u64, MockCollectorError>> {
+            let (tx, rx) = oneshot::channel();
+            self.latest_block_responses.lock().push_back(Response::Pending(rx));
+            tx
+        }
+
+        /// Push a pending `gather` response. See [`push_latest_block_pending`].
+        pub fn push_gather_pending(
+            &self,
+        ) -> oneshot::Sender<Result<GatherResult, MockCollectorError>> {
+            let (tx, rx) = oneshot::channel();
+            self.gather_responses.lock().push_back(Response::Pending(rx));
+            tx
         }
 
         /// Number of times `latest_block` has been called so far.
@@ -323,9 +372,14 @@ mod tests {
             &self,
         ) -> Pin<Box<dyn Future<Output = Result<u64, Self::Error>> + Send + '_>> {
             self.latest_block_call_count.fetch_add(1, Ordering::SeqCst);
-            let response =
-                self.latest_block_responses.lock().pop_front().unwrap_or(Err(MockCollectorError));
-            Box::pin(async move { response })
+            let response = self.latest_block_responses.lock().pop_front();
+            Box::pin(async move {
+                match response {
+                    Some(Response::Ready(r)) => r,
+                    Some(Response::Pending(rx)) => rx.await.unwrap_or(Err(MockCollectorError)),
+                    None => Err(MockCollectorError),
+                }
+            })
         }
 
         fn gather(
@@ -341,9 +395,14 @@ mod tests {
                 to_block,
                 chain_id,
             });
-            let response =
-                self.gather_responses.lock().pop_front().unwrap_or(Err(MockCollectorError));
-            Box::pin(async move { response })
+            let response = self.gather_responses.lock().pop_front();
+            Box::pin(async move {
+                match response {
+                    Some(Response::Ready(r)) => r,
+                    Some(Response::Pending(rx)) => rx.await.unwrap_or(Err(MockCollectorError)),
+                    None => Err(MockCollectorError),
+                }
+            })
         }
     }
 
@@ -635,4 +694,219 @@ mod tests {
         let res = tokio::time::timeout(SHORT, stream.next()).await.expect("ready promptly");
         assert!(res.is_none(), "stream should terminate when trigger ends");
     }
+
+    // -------------------------------------------------------------------------
+    // Rewind (Messenger trait)
+    // -------------------------------------------------------------------------
+
+    #[tokio::test]
+    async fn rewind_during_idle_updates_cursor() {
+        use crate::Messenger;
+
+        // Stream is idle (no trigger fired yet). Rewind moves the cursor;
+        // the next gather call must use the new (from_block, from_tx_index).
+        let (mut stream, collector, trigger) = build(100, 5, 0);
+
+        Messenger::rewind(&mut *stream, 10, 2);
+
+        collector.push_latest_block(Ok(50));
+        collector.push_gather(Ok(GatherResult { to_block: 50, messages: vec![] }));
+        trigger.fire();
+        let _ = stream.next().await.expect("yielded after rewind");
+
+        let calls = collector.gather_calls();
+        assert_eq!(calls.len(), 1);
+        assert_eq!(calls[0].from_block, 10, "gather must use rewound from_block");
+        assert_eq!(calls[0].from_tx_index, 2, "gather must use rewound from_tx_index");
+    }
+
+    /// A pending `latest_block` future suspends the stream in `CheckingBlock`.
+    /// Rewinding then must abandon that future — the next trigger tick must
+    /// rebuild from the new cursor without consuming the dangling response.
+    #[tokio::test]
+    async fn rewind_during_checking_block_drops_in_flight_future() {
+        use crate::Messenger;
+
+        let (mut stream, collector, trigger) = build(100, 0, 0);
+
+        // Queue a pending latest_block. After firing the trigger and polling,
+        // the stream parks in CheckingBlock waiting on this oneshot.
+        let lb_tx = collector.push_latest_block_pending();
+        trigger.fire();
+
+        // Drive the stream once. Since the future is pending, the stream must
+        // return Pending — we observe that via a short timeout.
+        assert_no_yield(&mut stream).await;
+        assert_eq!(collector.latest_block_calls(), 1, "future was polled");
+
+        // Rewind: phase resets to Idle, in-flight latest_block future is
+        // abandoned. Firing the pending sender after this point must NOT cause
+        // the stream to resume its old path.
+        Messenger::rewind(&mut *stream, 5, 0);
+
+        // Even if the old future is resolved late, it has been dropped: nothing
+        // happens. (Dropping the sender ensures the abandoned future would
+        // complete with an error if reused — we expect it not to be.)
+        drop(lb_tx);
+
+        // Next tick must use the rewound cursor.
+        collector.push_latest_block(Ok(30));
+        collector.push_gather(Ok(GatherResult { to_block: 30, messages: vec![] }));
+        trigger.fire();
+        let _ = tokio::time::timeout(SHORT, stream.next())
+            .await
+            .expect("woke after rewind")
+            .expect("yielded");
+
+        let calls = collector.gather_calls();
+        assert_eq!(calls.len(), 1, "only the post-rewind gather should have happened");
+        assert_eq!(calls[0].from_block, 5);
+        assert_eq!(calls[0].from_tx_index, 0);
+    }
+
+    /// Companion to `rewind_after_gather_resets_phase_to_idle`: verifies that
+    /// rewinding *while* a gather future is in flight (stream phase = Gathering)
+    /// abandons the future — no `MessagingOutcome` is yielded from the in-flight
+    /// gather, and the next gather call uses the rewound cursor.
+    #[tokio::test]
+    async fn rewind_truly_mid_gathering_abandons_future() {
+        use crate::Messenger;
+
+        let (mut stream, collector, trigger) = build(100, 0, 0);
+
+        // First tick: latest_block resolves, gather goes pending.
+        collector.push_latest_block(Ok(150));
+        let g_tx = collector.push_gather_pending();
+        trigger.fire();
+
+        // Drive the stream once — it parks in Gathering, awaiting the oneshot.
+        assert_no_yield(&mut stream).await;
+        assert_eq!(collector.gather_calls().len(), 1, "first gather was scheduled");
+
+        // Rewind while gather is in flight. Phase resets to Idle; the in-flight
+        // gather future is abandoned. The cursor jumps to (5, 0).
+        Messenger::rewind(&mut *stream, 5, 0);
+
+        // Fire the abandoned gather's oneshot. Its result must not be yielded.
+        let _ = g_tx.send(Ok(GatherResult { to_block: 150, messages: vec![msg(120, 0)] }));
+
+        // Confirm no outcome leaks from the abandoned future.
+        assert_no_yield(&mut stream).await;
+
+        // Next tick must drive a fresh gather from the rewound cursor.
+        collector.push_latest_block(Ok(30));
+        collector.push_gather(Ok(GatherResult { to_block: 30, messages: vec![] }));
+        trigger.fire();
+        let _ = tokio::time::timeout(SHORT, stream.next())
+            .await
+            .expect("woke after rewind")
+            .expect("yielded");
+
+        let calls = collector.gather_calls();
+        assert_eq!(calls.len(), 2, "first gather + post-rewind gather");
+        assert_eq!(calls[0].from_block, 100, "first gather keyed on original cursor");
+        assert_eq!(calls[1].from_block, 5, "second gather keyed on rewound cursor");
+    }
+
+    /// "Fast-forward" rewind: jump to a `from_block` *above* the current
+    /// cursor. Valid operator use case (skip blocks known to be empty).
+    #[tokio::test]
+    async fn rewind_to_higher_cursor_fast_forwards() {
+        use crate::Messenger;
+
+        let (mut stream, collector, trigger) = build(5, 0, 0);
+
+        Messenger::rewind(&mut *stream, 50, 0);
+
+        collector.push_latest_block(Ok(60));
+        collector.push_gather(Ok(GatherResult { to_block: 60, messages: vec![] }));
+        trigger.fire();
+        let _ = stream.next().await.expect("yielded");
+
+        let calls = collector.gather_calls();
+        assert_eq!(calls.len(), 1);
+        assert_eq!(calls[0].from_block, 50, "rewind fast-forwards from 5 to 50");
+        assert_eq!(calls[0].from_tx_index, 0);
+    }
+
+    /// Rewinding to the current cursor is a no-op semantically. Must not panic.
+    #[tokio::test]
+    async fn rewind_to_same_cursor_is_safe() {
+        use crate::Messenger;
+
+        let (mut stream, collector, trigger) = build(10, 0, 0);
+
+        Messenger::rewind(&mut *stream, 10, 0);
+
+        collector.push_latest_block(Ok(20));
+        collector.push_gather(Ok(GatherResult { to_block: 20, messages: vec![] }));
+        trigger.fire();
+        let _ = stream.next().await.expect("yielded");
+
+        let calls = collector.gather_calls();
+        assert_eq!(calls.len(), 1);
+        assert_eq!(calls[0].from_block, 10);
+        assert_eq!(calls[0].from_tx_index, 0);
+    }
+
+    /// Three back-to-back rewinds while idle: the last one wins. The cursor is
+    /// just two fields, so each call overwrites — there's no queueing semantics
+    /// to worry about at the stream layer.
+    #[tokio::test]
+    async fn multiple_rewinds_last_one_wins() {
+        use crate::Messenger;
+
+        let (mut stream, collector, trigger) = build(0, 0, 0);
+
+        Messenger::rewind(&mut *stream, 5, 0);
+        Messenger::rewind(&mut *stream, 15, 0);
+        Messenger::rewind(&mut *stream, 25, 0);
+
+        collector.push_latest_block(Ok(40));
+        collector.push_gather(Ok(GatherResult { to_block: 40, messages: vec![] }));
+        trigger.fire();
+        let _ = stream.next().await.expect("yielded");
+
+        let calls = collector.gather_calls();
+        assert_eq!(calls.len(), 1);
+        assert_eq!(calls[0].from_block, 25, "last rewind wins");
+        assert_eq!(calls[0].from_tx_index, 0);
+    }
+
+    /// Rewinding while a gather future is in flight resets `phase` to `Idle`,
+    /// so the next trigger tick rebuilds gather/checking futures from the new
+    /// cursor instead of continuing the old one. Verified by giving the stream
+    /// time to consume one gather, rewinding, and asserting the subsequent
+    /// gather is keyed on the new cursor.
+    #[tokio::test]
+    async fn rewind_after_gather_resets_phase_to_idle() {
+        use crate::Messenger;
+
+        let (mut stream, collector, trigger) = build(100, 0, 0);
+
+        // First tick fully consumes the queued latest_block + gather and the
+        // stream yields, ending in `Idle`. Cursor now advances to 201.
+        collector.push_latest_block(Ok(200));
+        collector.push_gather(Ok(GatherResult { to_block: 200, messages: vec![] }));
+        trigger.fire();
+        let _ = stream.next().await.expect("first gather yielded");
+
+        // Rewind: cursor jumps backward, phase reset to Idle (idempotently).
+        Messenger::rewind(&mut *stream, 5, 0);
+
+        // Next tick must use the rewound cursor (5, 0), not (201, 0).
+        collector.push_latest_block(Ok(30));
+        collector.push_gather(Ok(GatherResult { to_block: 30, messages: vec![] }));
+        trigger.fire();
+        let _ = tokio::time::timeout(SHORT, stream.next())
+            .await
+            .expect("woke after rewind")
+            .expect("yielded");
+
+        let calls = collector.gather_calls();
+        assert_eq!(calls.len(), 2);
+        assert_eq!(calls[0].from_block, 100, "first gather uses original cursor");
+        assert_eq!(calls[1].from_block, 5, "second gather uses rewound cursor");
+        assert_eq!(calls[1].from_tx_index, 0);
+    }
 }
diff --git a/crates/messaging/tests/checkpoint.rs b/crates/messaging/tests/checkpoint.rs
new file mode 100644
index 000000000..569c48719
--- /dev/null
+++ b/crates/messaging/tests/checkpoint.rs
@@ -0,0 +1,312 @@
+//! Live-rewind checkpoint tests for the messaging service against an Ethereum
+//! settlement layer.
+//!
+//! These drive checkpoint changes through [`MessagingController`] (NOT a
+//! JSON-RPC client — `katana-messaging` sits below the node/RPC layer). They
+//! exercise the operator-rewind path end-to-end against a real Anvil node and
+//! the real `StarknetMessagingLocal` Solidity contract: `set_checkpoint` /
+//! `reset_checkpoint` cause the running drain task to re-gather, and the pool's
+//! hash-level dedup must absorb the second pass without duplicating L2 txs.
+
+use std::time::{Duration, Instant};
+
+use alloy_primitives::{Uint, U256};
+use alloy_provider::ProviderBuilder;
+use alloy_sol_types::sol;
+use katana_messaging::{MessagingService, SettlementChainConfig};
+use katana_primitives::chain::ChainId;
+use katana_primitives::transaction::TxHash;
+use katana_primitives::{felt, ContractAddress};
+use rand::Rng;
+use starknet::macros::selector;
+use url::Url;
+
+mod common;
+
+sol!(
+    #[allow(missing_docs)]
+    #[sol(rpc)]
+    StarknetContract,
+    "tests/test_data/solidity/StarknetMessagingLocalCompiled.json"
+);
+
+sol!(
+    #[allow(missing_docs)]
+    #[sol(rpc)]
+    Contract1,
+    "tests/test_data/solidity/Contract1Compiled.json"
+);
+
+/// Poll the provider's messaging checkpoint until it is `Some` with
+/// `block >= min_block`, or `timeout` elapses.
+async fn wait_for_checkpoint_at_or_above(
+    provider: &katana_provider::DbProviderFactory,
+    min_block: u64,
+    timeout: Duration,
+) {
+    let deadline = Instant::now() + timeout;
+    loop {
+        if let Some(cp) = common::messaging_checkpoint(provider) {
+            if cp.block >= min_block {
+                return;
+            }
+        }
+        if Instant::now() >= deadline {
+            panic!("checkpoint did not advance to >= {min_block} within {timeout:?}");
+        }
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    }
+}
+
+/// Poll the L1->L2 index until `l1_hash` maps to a non-empty list, then return it.
+async fn wait_for_l1_to_l2_mapping(
+    provider: &katana_provider::DbProviderFactory,
+    l1_hash: &[u8; 32],
+    timeout: Duration,
+) -> Vec<TxHash> {
+    let deadline = Instant::now() + timeout;
+    loop {
+        let mapped = common::l2_txs_for_l1(provider, l1_hash);
+        if !mapped.is_empty() {
+            return mapped;
+        }
+        if Instant::now() >= deadline {
+            panic!("L1->L2 mapping never recorded within {timeout:?}");
+        }
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    }
+}
+
+/// Mid-flight set: after the service has processed a message and advanced the
+/// checkpoint, setting the checkpoint to a *prior* block must cause a
+/// re-gather. The pool's hash-level dedup must prevent duplicate L2 txs and the
+/// checkpoint must re-advance to (at least) its prior value.
+#[tokio::test(flavor = "multi_thread")]
+async fn set_checkpoint_mid_flight_causes_re_gather() {
+    let port: u16 = rand::thread_rng().gen_range(35000..65000);
+
+    let l1_provider = ProviderBuilder::new()
+        .connect_anvil_with_wallet_and_config(|anvil| anvil.port(port))
+        .expect("failed to build eth provider");
+
+    let core_contract = StarknetContract::deploy(&l1_provider).await.unwrap();
+    let l1_test = Contract1::deploy(&l1_provider, *core_contract.address()).await.unwrap();
+
+    let settlement = SettlementChainConfig::Ethereum {
+        rpc_url: Url::parse(&format!("http://localhost:{}", port)).unwrap(),
+        contract_address: *core_contract.address(),
+    };
+
+    let pool = common::build_test_pool();
+    let provider = common::build_test_provider();
+
+    let service =
+        MessagingService::new(ChainId::default(), pool.clone(), provider.clone(), settlement)
+            .interval(1)
+            .from_block(0);
+
+    // Grab the controller BEFORE `start()` takes the rewind receiver — we need
+    // it to drive a live rewind while the service runs.
+    let controller = service.controller();
+    let mut handle = service.start().expect("start messaging service");
+
+    // Send one L1->L2 message.
+    let recipient = ContractAddress::from(felt!("0xbeef"));
+    let entry_point_selector = selector!("msg_handler_value");
+    let calldata = [123u8];
+    let receipt = l1_test
+        .sendMessage(
+            recipient.into(),
+            U256::from_be_bytes(entry_point_selector.to_bytes_be()),
+            calldata.iter().map(|x| U256::from(*x)).collect::<Vec<_>>(),
+        )
+        .gas(12_000_000)
+        .value(Uint::from(1))
+        .send()
+        .await
+        .unwrap()
+        .get_receipt()
+        .await
+        .unwrap();
+    assert!(receipt.status(), "L1 sendMessage tx reverted");
+    let l1_hash: [u8; 32] = receipt.transaction_hash.0;
+
+    // Wait for the service to process it.
+    let mapped_before =
+        wait_for_l1_to_l2_mapping(&provider, &l1_hash, Duration::from_secs(15)).await;
+    assert_eq!(mapped_before.len(), 1, "single L1 tx -> single L2 tx");
+
+    let cp_before =
+        common::messaging_checkpoint(&provider).expect("checkpoint should have advanced");
+    assert!(cp_before.block > 0, "expected checkpoint to have advanced");
+
+    // Operator action: rewind below the current checkpoint to force a re-gather.
+    controller.set_checkpoint(0, 0).await.expect("set_checkpoint succeeds");
+    wait_for_checkpoint_at_or_above(&provider, cp_before.block, Duration::from_secs(15)).await;
+
+    // The L1->L2 index must still hold exactly one L2 tx for this L1 hash. The
+    // re-gather re-published the L1Handler to the pool; the pool's hash-level
+    // dedup absorbed the duplicate.
+    let mapped_after = common::l2_txs_for_l1(&provider, &l1_hash);
+    assert_eq!(
+        mapped_after.len(),
+        1,
+        "re-gather must not duplicate L1->L2 mapping (pool dedup contract)"
+    );
+    assert_eq!(mapped_after, mapped_before, "same L2 tx hash both times");
+
+    handle.stop();
+    handle.stopped().await;
+}
+
+/// Reset clears the persisted checkpoint and rewinds the service to the
+/// configured `from_block`. `get_checkpoint` must return `None` immediately
+/// after reset, and a fresh L1 message must still be processed — the service
+/// keeps running and re-gathers cleanly.
+#[tokio::test(flavor = "multi_thread")]
+async fn reset_checkpoint_resumes_from_configured_from_block() {
+    let port: u16 = rand::thread_rng().gen_range(35000..65000);
+
+    let l1_provider = ProviderBuilder::new()
+        .connect_anvil_with_wallet_and_config(|anvil| anvil.port(port))
+        .expect("failed to build eth provider");
+
+    let core_contract = StarknetContract::deploy(&l1_provider).await.unwrap();
+    let l1_test = Contract1::deploy(&l1_provider, *core_contract.address()).await.unwrap();
+
+    let settlement = SettlementChainConfig::Ethereum {
+        rpc_url: Url::parse(&format!("http://localhost:{}", port)).unwrap(),
+        contract_address: *core_contract.address(),
+    };
+
+    let pool = common::build_test_pool();
+    let provider = common::build_test_provider();
+
+    let service =
+        MessagingService::new(ChainId::default(), pool.clone(), provider.clone(), settlement)
+            .interval(1)
+            .from_block(0);
+
+    let controller = service.controller();
+    let mut handle = service.start().expect("start messaging service");
+
+    let recipient = ContractAddress::from(felt!("0xbeef"));
+    let entry_point_selector = selector!("msg_handler_value");
+
+    // Send first message and wait for the checkpoint to record it.
+    let receipt = l1_test
+        .sendMessage(
+            recipient.into(),
+            U256::from_be_bytes(entry_point_selector.to_bytes_be()),
+            vec![U256::from(7u8)],
+        )
+        .gas(12_000_000)
+        .value(Uint::from(1))
+        .send()
+        .await
+        .unwrap()
+        .get_receipt()
+        .await
+        .unwrap();
+    assert!(receipt.status());
+    wait_for_checkpoint_at_or_above(&provider, 1, Duration::from_secs(15)).await;
+
+    // Reset clears the row. `get_checkpoint` must observe `None` right away —
+    // the DB delete is synchronous (the live rewind signal is fire-and-forget).
+    controller.reset_checkpoint().await.expect("reset_checkpoint succeeds");
+    let cp_post_reset = controller.get_checkpoint().expect("get_checkpoint succeeds");
+    assert!(cp_post_reset.is_none(), "reset deletes the row; get returns None immediately after");
+
+    // Fresh message after reset: the service kept running and still processes it,
+    // re-establishing the checkpoint from the configured `from_block`.
+    let receipt = l1_test
+        .sendMessage(
+            recipient.into(),
+            U256::from_be_bytes(entry_point_selector.to_bytes_be()),
+            vec![U256::from(13u8)],
+        )
+        .gas(12_000_000)
+        .value(Uint::from(1))
+        .send()
+        .await
+        .unwrap()
+        .get_receipt()
+        .await
+        .unwrap();
+    assert!(receipt.status());
+    let l1_hash_b: [u8; 32] = receipt.transaction_hash.0;
+    let mapped = wait_for_l1_to_l2_mapping(&provider, &l1_hash_b, Duration::from_secs(15)).await;
+    assert_eq!(mapped.len(), 1, "fresh message -> single L2 tx");
+
+    handle.stop();
+    handle.stopped().await;
+}
+
+/// Pool dedup contract: re-gather of an already-processed block must NOT add a
+/// second L2 tx for the same L1 hash. Directly checks the DupSort index stays
+/// single-entry through the rewind round-trip.
+#[tokio::test(flavor = "multi_thread")]
+async fn re_gather_after_rewind_does_not_duplicate_l2_txs() {
+    let port: u16 = rand::thread_rng().gen_range(35000..65000);
+
+    let l1_provider = ProviderBuilder::new()
+        .connect_anvil_with_wallet_and_config(|anvil| anvil.port(port))
+        .expect("failed to build eth provider");
+
+    let core_contract = StarknetContract::deploy(&l1_provider).await.unwrap();
+    let l1_test = Contract1::deploy(&l1_provider, *core_contract.address()).await.unwrap();
+
+    let settlement = SettlementChainConfig::Ethereum {
+        rpc_url: Url::parse(&format!("http://localhost:{}", port)).unwrap(),
+        contract_address: *core_contract.address(),
+    };
+
+    let pool = common::build_test_pool();
+    let provider = common::build_test_provider();
+
+    let service =
+        MessagingService::new(ChainId::default(), pool.clone(), provider.clone(), settlement)
+            .interval(1)
+            .from_block(0);
+
+    let controller = service.controller();
+    let mut handle = service.start().expect("start messaging service");
+
+    // Send exactly one L1 message.
+    let recipient = ContractAddress::from(felt!("0xbeef"));
+    let entry_point_selector = selector!("msg_handler_value");
+    let receipt = l1_test
+        .sendMessage(
+            recipient.into(),
+            U256::from_be_bytes(entry_point_selector.to_bytes_be()),
+            vec![U256::from(42u8)],
+        )
+        .gas(12_000_000)
+        .value(Uint::from(1))
+        .send()
+        .await
+        .unwrap()
+        .get_receipt()
+        .await
+        .unwrap();
+    assert!(receipt.status());
+    let l1_hash: [u8; 32] = receipt.transaction_hash.0;
+
+    // Wait for the single L2 tx to be recorded, capture it.
+    let initial = wait_for_l1_to_l2_mapping(&provider, &l1_hash, Duration::from_secs(15)).await;
+    assert_eq!(initial.len(), 1);
+    let l2_hash = initial[0];
+
+    // Force a re-gather by rewinding below the current checkpoint.
+    let cp_before = common::messaging_checkpoint(&provider).expect("checkpoint exists");
+    controller.set_checkpoint(0, 0).await.expect("set_checkpoint succeeds");
+    wait_for_checkpoint_at_or_above(&provider, cp_before.block, Duration::from_secs(15)).await;
+
+    // Still exactly one mapping, still the same L2 hash.
+    let final_mapping = common::l2_txs_for_l1(&provider, &l1_hash);
+    assert_eq!(final_mapping.len(), 1, "no duplicates after rewind");
+    assert_eq!(final_mapping[0], l2_hash, "same L2 tx hash as before rewind");
+
+    handle.stop();
+    handle.stopped().await;
+}
diff --git a/crates/messaging/tests/common/mod.rs b/crates/messaging/tests/common/mod.rs
index fc9c34228..834994a58 100644
--- a/crates/messaging/tests/common/mod.rs
+++ b/crates/messaging/tests/common/mod.rs
@@ -68,7 +68,7 @@ pub fn l2_txs_for_l1(provider: &DbProviderFactory, l1_tx_hash: &[u8; 32]) -> Vec
 /// Read the messaging checkpoint for the default messaging id ("messaging") from the provider.
 pub fn messaging_checkpoint(provider: &DbProviderFactory) -> Option<MessagingCheckpoint> {
     let tx = provider.provider_mut();
-    let res = tx.messaging_checkpoint("messaging").expect("read messaging checkpoint");
+    let res = tx.messaging_checkpoint().expect("read messaging checkpoint");
     tx.commit().expect("commit read tx");
     res
 }
diff --git a/crates/node/config/src/rpc.rs b/crates/node/config/src/rpc.rs
index 64906b235..c69126700 100644
--- a/crates/node/config/src/rpc.rs
+++ b/crates/node/config/src/rpc.rs
@@ -36,6 +36,7 @@ pub enum RpcModuleKind {
     TxPool,
     Node,
     Cartridge,
+    Messaging,
 }
 
 /// Configuration for the RPC server.
@@ -108,6 +109,7 @@ impl RpcModulesList {
             RpcModuleKind::TxPool,
             RpcModuleKind::Node,
             RpcModuleKind::Cartridge,
+            RpcModuleKind::Messaging,
         ]))
     }
 
diff --git a/crates/node/sequencer/src/lib.rs b/crates/node/sequencer/src/lib.rs
index 33a56bbea..2522c52f3 100755
--- a/crates/node/sequencer/src/lib.rs
+++ b/crates/node/sequencer/src/lib.rs
@@ -45,6 +45,7 @@ use katana_provider::{
 use katana_rpc_api::cartridge::CartridgeApiServer;
 use katana_rpc_api::dev::DevApiServer;
 use katana_rpc_api::katana::KatanaApiServer;
+use katana_rpc_api::messaging::MessagingApiServer;
 use katana_rpc_api::node::NodeApiServer;
 use katana_rpc_api::paymaster::PaymasterApiServer;
 use katana_rpc_api::starknet::{StarknetApiServer, StarknetSubscriptionApiServer};
@@ -54,6 +55,7 @@ use katana_rpc_api::starknet_ext::StarknetApiExtServer;
 use katana_rpc_api::tee::TeeApiServer;
 use katana_rpc_server::cartridge::{CartridgeApi, CartridgeConfig};
 use katana_rpc_server::dev::DevApi;
+use katana_rpc_server::messaging::MessagingApiHandler;
 use katana_rpc_server::middleware::cartridge::{ControllerDeploymentLayer, VrfLayer};
 use katana_rpc_server::middleware::cors::Cors;
 use katana_rpc_server::middleware::logger::RpcLoggerLayer;
@@ -112,7 +114,7 @@ where
     block_notify: broadcast::Sender<MinedBlockOutcome>,
     gateway_server: Option<GatewayServer<TxPool, BlockProducer<P>, P>>,
     metrics_server: Option<MetricsServer<Prometheus>>,
-    messaging_service: Option<MessagingService<P>>,
+    messaging_service: Option<MessagingService<P, TxPool>>,
 }
 
 impl<P> Node<P>
@@ -311,6 +313,28 @@ where
             rpc_modules.merge(KatanaApiServer::into_rpc(starknet_api.clone()))?;
         }
 
+        // --- build messaging service (early so its controller can back the
+        //     `messaging` RPC namespace registered below).
+
+        let messaging_service = config.messaging.as_ref().map(|cfg| {
+            MessagingService::new(
+                backend.chain_spec.id(),
+                pool.clone(),
+                provider.clone(),
+                cfg.settlement.clone(),
+            )
+            .interval(cfg.interval)
+            .from_block(cfg.from_block)
+            .confirmation_depth(cfg.confirmation_depth)
+        });
+
+        if let Some(ref service) = messaging_service {
+            if config.rpc.apis.contains(&RpcModuleKind::Messaging) {
+                let handler = MessagingApiHandler::new(service.controller());
+                rpc_modules.merge(MessagingApiServer::into_rpc(handler))?;
+            }
+        }
+
         if config.rpc.apis.contains(&RpcModuleKind::Dev) {
             let api = DevApi::new(backend.clone(), block_producer.clone(), pool.clone());
             rpc_modules.merge(DevApiServer::into_rpc(api))?;
@@ -539,20 +563,6 @@ where
             None
         };
 
-        // --- build messaging server
-
-        let messaging_service = config.messaging.as_ref().map(|cfg| {
-            MessagingService::new(
-                backend.chain_spec.id(),
-                pool.clone(),
-                provider.clone(),
-                cfg.settlement.clone(),
-            )
-            .interval(cfg.interval)
-            .from_block(cfg.from_block)
-            .confirmation_depth(cfg.confirmation_depth)
-        });
-
         Ok(Node {
             db,
             provider,
@@ -842,8 +852,8 @@ where
         &self.rpc_server
     }
 
-    /// Returns a reference to the node's messaging server, if messaging is enabled.
-    pub fn messaging_server(&self) -> Option<&MessagingService<P>> {
+    /// Returns a reference to the node's messaging service, if messaging is enabled.
+    pub fn messaging_service(&self) -> Option<&MessagingService<P, TxPool>> {
         self.messaging_service.as_ref()
     }
 
diff --git a/crates/rpc/rpc-api/src/error/messaging.rs b/crates/rpc/rpc-api/src/error/messaging.rs
new file mode 100644
index 000000000..b16ed9e12
--- /dev/null
+++ b/crates/rpc/rpc-api/src/error/messaging.rs
@@ -0,0 +1,27 @@
+use jsonrpsee::types::ErrorObjectOwned;
+
+/// Error codes for the `messaging` namespace. Start at 200 to avoid collision
+/// with other Katana RPC error enums.
+#[derive(thiserror::Error, Clone, Debug)]
+#[allow(clippy::enum_variant_names)]
+pub enum MessagingApiError {
+    #[error("Messaging checkpoint storage error: {0}")]
+    StorageError(String),
+    #[error("Failed to signal messaging server for rewind: {0}")]
+    RewindSignalFailed(String),
+}
+
+impl MessagingApiError {
+    fn code(&self) -> i32 {
+        match self {
+            Self::StorageError(_) => 200,
+            Self::RewindSignalFailed(_) => 201,
+        }
+    }
+}
+
+impl From<MessagingApiError> for ErrorObjectOwned {
+    fn from(err: MessagingApiError) -> Self {
+        ErrorObjectOwned::owned(err.code(), err.to_string(), None::<()>)
+    }
+}
diff --git a/crates/rpc/rpc-api/src/error/mod.rs b/crates/rpc/rpc-api/src/error/mod.rs
index f16dfe016..47b90d51a 100644
--- a/crates/rpc/rpc-api/src/error/mod.rs
+++ b/crates/rpc/rpc-api/src/error/mod.rs
@@ -1,5 +1,6 @@
 pub mod cartridge;
 pub mod dev;
 pub mod katana;
+pub mod messaging;
 pub mod starknet;
 pub mod tee;
diff --git a/crates/rpc/rpc-api/src/lib.rs b/crates/rpc/rpc-api/src/lib.rs
index 2884fa12a..79bec19db 100644
--- a/crates/rpc/rpc-api/src/lib.rs
+++ b/crates/rpc/rpc-api/src/lib.rs
@@ -4,6 +4,7 @@ pub mod cartridge;
 pub mod dev;
 pub mod error;
 pub mod katana;
+pub mod messaging;
 pub mod node;
 pub mod starknet;
 pub mod starknet_ext;
diff --git a/crates/rpc/rpc-api/src/messaging.rs b/crates/rpc/rpc-api/src/messaging.rs
new file mode 100644
index 000000000..7b883824a
--- /dev/null
+++ b/crates/rpc/rpc-api/src/messaging.rs
@@ -0,0 +1,35 @@
+use jsonrpsee::core::RpcResult;
+use jsonrpsee::proc_macros::rpc;
+use katana_rpc_types::messaging::MessagingCheckpoint;
+
+/// Operator-facing RPC methods for the L1->L2 messaging server.
+///
+/// All three methods read or write the persisted messaging checkpoint AND
+/// signal the running messenger to live-rewind its in-memory cursor, so
+/// operators can recover missed messages without restarting the node.
+#[cfg_attr(not(feature = "client"), rpc(server, namespace = "messaging"))]
+#[cfg_attr(feature = "client", rpc(client, server, namespace = "messaging"))]
+pub trait MessagingApi {
+    /// Returns the last *committed* checkpoint — the same value the messaging
+    /// server reads on boot. Returns `null` when no checkpoint row exists.
+    ///
+    /// This reflects the DB state, not the live in-memory gather position.
+    #[method(name = "getCheckpoint")]
+    async fn get_checkpoint(&self) -> RpcResult<Option<MessagingCheckpoint>>;
+
+    /// Persist `(block, tx_index)` as the last processed message and rewind
+    /// the live cursor to `(block, tx_index + 1)`.
+    ///
+    /// Note the off-by-one: a checkpoint represents the last successfully
+    /// processed message, so the next gather resumes one past it. To re-gather
+    /// from the very beginning of block 0 use [`resetCheckpoint`] instead —
+    /// `setCheckpoint(0, 0)` would skip tx 0 of block 0.
+    #[method(name = "setCheckpoint")]
+    async fn set_checkpoint(&self, block: u64, tx_index: u64) -> RpcResult<()>;
+
+    /// Delete the persisted checkpoint and rewind the live cursor to the
+    /// messenger's configured `from_block` with `tx_index = 0`. The next boot
+    /// will also start from `from_block` since no checkpoint row exists.
+    #[method(name = "resetCheckpoint")]
+    async fn reset_checkpoint(&self) -> RpcResult<()>;
+}
diff --git a/crates/rpc/rpc-server/Cargo.toml b/crates/rpc/rpc-server/Cargo.toml
index ed9d94adf..1f55fded4 100644
--- a/crates/rpc/rpc-server/Cargo.toml
+++ b/crates/rpc/rpc-server/Cargo.toml
@@ -13,6 +13,7 @@ katana-executor.workspace = true
 katana-explorer = { workspace = true, features = [ "jsonrpsee" ], optional = true }
 katana-gas-price-oracle.workspace = true
 katana-genesis.workspace = true
+katana-messaging.workspace = true
 katana-metrics.workspace = true
 katana-pool.workspace = true
 katana-primitives.workspace = true
diff --git a/crates/rpc/rpc-server/src/lib.rs b/crates/rpc/rpc-server/src/lib.rs
index 5bf5c7935..e6d5eda0b 100644
--- a/crates/rpc/rpc-server/src/lib.rs
+++ b/crates/rpc/rpc-server/src/lib.rs
@@ -20,6 +20,7 @@ use tracing::info;
 pub mod cartridge;
 pub mod dev;
 pub mod health;
+pub mod messaging;
 pub mod middleware;
 pub mod node;
 pub mod paymaster;
diff --git a/crates/rpc/rpc-server/src/messaging.rs b/crates/rpc/rpc-server/src/messaging.rs
new file mode 100644
index 000000000..0fd48037a
--- /dev/null
+++ b/crates/rpc/rpc-server/src/messaging.rs
@@ -0,0 +1,72 @@
+use jsonrpsee::core::{async_trait, RpcResult};
+use katana_messaging::MessagingController;
+use katana_provider::api::messaging::{MessagingCheckpointProvider, MessagingL1ToL2IndexWriter};
+use katana_provider::{MutableProvider, ProviderFactory, ProviderRW};
+use katana_rpc_api::error::messaging::MessagingApiError;
+use katana_rpc_api::messaging::MessagingApiServer;
+use katana_rpc_types::messaging::MessagingCheckpoint;
+
+fn to_rpc(cp: katana_provider::api::messaging::MessagingCheckpoint) -> MessagingCheckpoint {
+    MessagingCheckpoint { block: cp.block, tx_index: cp.tx_index }
+}
+
+/// JSON-RPC handler for the `messaging` namespace. Delegates to a
+/// [`MessagingController`] obtained from a running [`MessagingServer`].
+#[allow(missing_debug_implementations)]
+pub struct MessagingApiHandler<P> {
+    controller: MessagingController<P>,
+}
+
+impl<P> MessagingApiHandler<P> {
+    pub fn new(controller: MessagingController<P>) -> Self {
+        Self { controller }
+    }
+}
+
+impl<P> MessagingApiHandler<P>
+where
+    P: ProviderFactory + Clone + Send + Sync + 'static,
+    <P as ProviderFactory>::ProviderMut:
+        ProviderRW + MessagingCheckpointProvider + MessagingL1ToL2IndexWriter + MutableProvider,
+{
+    fn get_checkpoint(&self) -> Result<Option<MessagingCheckpoint>, MessagingApiError> {
+        self.controller
+            .get_checkpoint()
+            .map(|opt| opt.map(to_rpc))
+            .map_err(|e| MessagingApiError::StorageError(e.to_string()))
+    }
+
+    async fn set_checkpoint(&self, block: u64, tx_index: u64) -> Result<(), MessagingApiError> {
+        self.controller
+            .set_checkpoint(block, tx_index)
+            .await
+            .map_err(|e| MessagingApiError::StorageError(e.to_string()))
+    }
+
+    async fn reset_checkpoint(&self) -> Result<(), MessagingApiError> {
+        self.controller
+            .reset_checkpoint()
+            .await
+            .map_err(|e| MessagingApiError::StorageError(e.to_string()))
+    }
+}
+
+#[async_trait]
+impl<P> MessagingApiServer for MessagingApiHandler<P>
+where
+    P: ProviderFactory + Clone + Send + Sync + 'static,
+    <P as ProviderFactory>::ProviderMut:
+        ProviderRW + MessagingCheckpointProvider + MessagingL1ToL2IndexWriter + MutableProvider,
+{
+    async fn get_checkpoint(&self) -> RpcResult<Option<MessagingCheckpoint>> {
+        Ok(self.get_checkpoint()?)
+    }
+
+    async fn set_checkpoint(&self, block: u64, tx_index: u64) -> RpcResult<()> {
+        Ok(self.set_checkpoint(block, tx_index).await?)
+    }
+
+    async fn reset_checkpoint(&self) -> RpcResult<()> {
+        Ok(self.reset_checkpoint().await?)
+    }
+}
diff --git a/crates/rpc/rpc-server/tests/messaging.rs b/crates/rpc/rpc-server/tests/messaging.rs
index dde809535..c003ef0eb 100644
--- a/crates/rpc/rpc-server/tests/messaging.rs
+++ b/crates/rpc/rpc-server/tests/messaging.rs
@@ -220,3 +220,160 @@ mod messages_status {
         assert!(returned.contains(&l2_b), "missing l2_b in response");
     }
 }
+
+// ==============================================================================
+// `messaging_*` checkpoint RPC tests.
+//
+// These exercise the new `messaging` namespace end-to-end against a running
+// TestNode without requiring a settlement chain. The messaging server is
+// enabled with a dummy URL — the drain task's gather calls will fail and be
+// logged, but those failures are isolated from the checkpoint RPC paths the
+// controller drives synchronously through the DB.
+// ==============================================================================
+
+mod checkpoint {
+    use katana_messaging::MessagingConfig;
+    use katana_provider::api::messaging::{MessagingCheckpoint, MessagingCheckpointProvider};
+    use katana_provider::{MutableProvider, ProviderFactory};
+    use katana_rpc_api::messaging::MessagingApiClient;
+    use katana_utils::TestNode;
+    use url::Url;
+
+    /// Build a config with messaging enabled and a deliberately unreachable
+    /// settlement URL. The drain task will log errors but the RPC handler that
+    /// drives the controller works purely off the DB and the rewind channel.
+    fn messaging_test_config() -> katana_sequencer_node::config::Config {
+        let mut config = katana_utils::node::test_config();
+        config.messaging = Some(MessagingConfig {
+            settlement: katana_messaging::SettlementChainConfig::Ethereum {
+                rpc_url: Url::parse("http://127.0.0.1:1").unwrap(),
+                contract_address: Default::default(),
+            },
+            interval: 60,
+            from_block: 42,
+            confirmation_depth: 0,
+        });
+        config
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn get_checkpoint_returns_null_when_no_row_persisted() {
+        let node = TestNode::new_with_config(messaging_test_config()).await;
+        let client = node.rpc_http_client();
+
+        let cp = client.get_checkpoint().await.expect("rpc call succeeds");
+        assert!(cp.is_none(), "fresh DB returns null checkpoint");
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn set_checkpoint_persists_row_visible_to_subsequent_get() {
+        let node = TestNode::new_with_config(messaging_test_config()).await;
+        let client = node.rpc_http_client();
+
+        client.set_checkpoint(100, 5).await.expect("set_checkpoint succeeds");
+
+        let cp = client.get_checkpoint().await.expect("get_checkpoint succeeds").expect("Some");
+        assert_eq!(cp.block, 100);
+        assert_eq!(cp.tx_index, 5);
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn reset_checkpoint_deletes_row_so_get_returns_null() {
+        let node = TestNode::new_with_config(messaging_test_config()).await;
+        let client = node.rpc_http_client();
+
+        // Pre-populate via the provider so we don't depend on `setCheckpoint`
+        // working — exercises reset in isolation.
+        let factory = node.handle().node().provider();
+        let tx = factory.provider_mut();
+        tx.set_messaging_checkpoint(&MessagingCheckpoint { block: 11, tx_index: 2 })
+            .expect("set checkpoint");
+        MutableProvider::commit(tx).expect("commit");
+
+        let pre = client.get_checkpoint().await.expect("rpc").expect("Some pre-reset");
+        assert_eq!(pre.block, 11);
+
+        client.reset_checkpoint().await.expect("reset_checkpoint succeeds");
+
+        let post = client.get_checkpoint().await.expect("rpc");
+        assert!(post.is_none(), "reset deletes the row");
+    }
+
+    /// Two consecutive `setCheckpoint` calls: the second value must overwrite
+    /// the first in the DB. This is the canonical "operator changed their mind"
+    /// path — there's no merge semantics, last write wins.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn set_checkpoint_twice_persists_latest() {
+        let node = TestNode::new_with_config(messaging_test_config()).await;
+        let client = node.rpc_http_client();
+
+        client.set_checkpoint(100, 5).await.expect("first set succeeds");
+        client.set_checkpoint(50, 0).await.expect("second set succeeds");
+
+        let cp = client.get_checkpoint().await.expect("rpc").expect("Some");
+        assert_eq!(cp.block, 50, "latest set wins");
+        assert_eq!(cp.tx_index, 0);
+    }
+
+    /// Wiring contract: the `messaging` namespace is registered only when BOTH
+    /// `config.messaging.is_some()` AND `RpcModuleKind::Messaging` is in `apis`.
+    /// These tests verify the four cells of that truth table — the "happy path"
+    /// is already covered by the tests above (both enabled + RPC works).
+    mod wiring {
+        use katana_node_config::rpc::{RpcModuleKind, RpcModulesList};
+        use katana_rpc_api::messaging::MessagingApiClient;
+        use katana_utils::TestNode;
+
+        use super::messaging_test_config;
+
+        /// Messaging server enabled, but `Messaging` not in `rpc.apis`: the
+        /// namespace must not be registered. The RPC call returns a method-not-found
+        /// (or similar) error.
+        #[tokio::test(flavor = "multi_thread")]
+        async fn messaging_server_present_but_api_disabled_does_not_register_namespace() {
+            let mut config = messaging_test_config();
+            // Default RpcModulesList omits Messaging; reset and add only the
+            // others we need for the node to launch.
+            let mut apis = RpcModulesList::new();
+            apis.add(RpcModuleKind::Starknet);
+            apis.add(RpcModuleKind::Node);
+            config.rpc.apis = apis;
+
+            let node = TestNode::new_with_config(config).await;
+            let client = node.rpc_http_client();
+
+            let res = client.get_checkpoint().await;
+            assert!(
+                res.is_err(),
+                "messaging RPC must not be reachable when Messaging is not in `apis`: got {res:?}"
+            );
+        }
+
+        /// Messaging API in `apis` but no settlement configured: the namespace
+        /// must NOT be registered (the wiring depends on `messaging_server.is_some()`).
+        /// The RPC call must fail.
+        #[tokio::test(flavor = "multi_thread")]
+        async fn messaging_api_enabled_but_no_settlement_does_not_register_namespace() {
+            let mut config = katana_utils::node::test_config();
+            // No `config.messaging = Some(...)` here — messaging server isn't
+            // built, even though `Messaging` is in `apis` (test_config uses
+            // RpcModulesList::all()).
+            assert!(config.messaging.is_none(), "precondition");
+            assert!(
+                config.rpc.apis.contains(&RpcModuleKind::Messaging),
+                "precondition: test_config exposes all APIs"
+            );
+            // Belt and suspenders: be explicit even if test_config changes.
+            config.rpc.apis.add(RpcModuleKind::Messaging);
+
+            let node = TestNode::new_with_config(config).await;
+            let client = node.rpc_http_client();
+
+            let res = client.get_checkpoint().await;
+            assert!(
+                res.is_err(),
+                "messaging RPC must not be reachable without a settlement chain: got {res:?}"
+            );
+        }
+    }
+}
diff --git a/crates/rpc/rpc-types/src/lib.rs b/crates/rpc/rpc-types/src/lib.rs
index 04cb25049..86ebaec29 100644
--- a/crates/rpc/rpc-types/src/lib.rs
+++ b/crates/rpc/rpc-types/src/lib.rs
@@ -15,6 +15,7 @@ pub mod class;
 pub mod event;
 pub mod list;
 pub mod message;
+pub mod messaging;
 pub mod node;
 pub mod outside_execution;
 pub mod receipt;
diff --git a/crates/rpc/rpc-types/src/messaging.rs b/crates/rpc/rpc-types/src/messaging.rs
new file mode 100644
index 000000000..a941caab8
--- /dev/null
+++ b/crates/rpc/rpc-types/src/messaging.rs
@@ -0,0 +1,17 @@
+use serde::{Deserialize, Serialize};
+
+/// RPC representation of the messaging service checkpoint.
+///
+/// Mirrors `katana_provider_api::messaging::MessagingCheckpoint` but lives in
+/// the RPC type crate so the public wire format stays independent of provider
+/// internals. The `From` conversion lives in `katana-rpc-server` to avoid a
+/// `rpc-types -> provider-api` dependency cycle.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MessagingCheckpoint {
+    /// The settlement chain block number last successfully processed.
+    pub block: u64,
+    /// The transaction index within `block` up to which messages have been
+    /// processed.
+    pub tx_index: u64,
+}
diff --git a/crates/storage/provider/provider-api/src/messaging.rs b/crates/storage/provider/provider-api/src/messaging.rs
index d4dde3432..0034be586 100644
--- a/crates/storage/provider/provider-api/src/messaging.rs
+++ b/crates/storage/provider/provider-api/src/messaging.rs
@@ -11,18 +11,17 @@ pub struct MessagingCheckpoint {
     pub tx_index: u64,
 }
 
-/// Provider for reading and writing messaging service checkpoints.
+/// Provider for reading and writing the messaging service checkpoint.
 #[auto_impl::auto_impl(&, Box, Arc)]
 pub trait MessagingCheckpointProvider: Send + Sync {
-    /// Returns the last successfully processed checkpoint for the given messenger.
-    fn messaging_checkpoint(&self, id: &str) -> ProviderResult<Option<MessagingCheckpoint>>;
+    /// Returns the last successfully processed checkpoint, if any.
+    fn messaging_checkpoint(&self) -> ProviderResult<Option<MessagingCheckpoint>>;
 
-    /// Sets the messaging checkpoint for the given messenger.
-    fn set_messaging_checkpoint(
-        &self,
-        id: &str,
-        checkpoint: &MessagingCheckpoint,
-    ) -> ProviderResult<()>;
+    /// Sets the messaging checkpoint.
+    fn set_messaging_checkpoint(&self, checkpoint: &MessagingCheckpoint) -> ProviderResult<()>;
+
+    /// Deletes the messaging checkpoint. No-op if no checkpoint row exists.
+    fn delete_messaging_checkpoint(&self) -> ProviderResult<()>;
 }
 
 /// Read-only access to the settlement-chain L1 -> L2 index.
diff --git a/crates/storage/provider/provider/src/providers/db/mod.rs b/crates/storage/provider/provider/src/providers/db/mod.rs
index d4cd604a8..030f648c0 100644
--- a/crates/storage/provider/provider/src/providers/db/mod.rs
+++ b/crates/storage/provider/provider/src/providers/db/mod.rs
@@ -925,25 +925,33 @@ impl<Tx: DbTxMut> StageCheckpointProvider for DbProvider<Tx> {
     }
 }
 
+/// Key under which the singleton messaging checkpoint row is stored in the
+/// shared `MessagingCheckpoints` table. An implementation detail — the trait
+/// surface intentionally hides it.
+const MESSAGING_CHECKPOINT_KEY: &str = "messaging";
+
 impl<Tx: DbTxMut> MessagingCheckpointProvider for DbProvider<Tx> {
-    fn messaging_checkpoint(
-        &self,
-        id: &str,
-    ) -> ProviderResult<Option<messaging::MessagingCheckpoint>> {
-        let result = self.0.get::<tables::MessagingCheckpoints>(id.to_string())?;
+    fn messaging_checkpoint(&self) -> ProviderResult<Option<messaging::MessagingCheckpoint>> {
+        let result =
+            self.0.get::<tables::MessagingCheckpoints>(MESSAGING_CHECKPOINT_KEY.to_string())?;
         Ok(result.map(|c| messaging::MessagingCheckpoint { block: c.block, tx_index: c.tx_index }))
     }
 
     fn set_messaging_checkpoint(
         &self,
-        id: &str,
         checkpoint: &messaging::MessagingCheckpoint,
     ) -> ProviderResult<()> {
         let value = katana_db::models::stage::MessagingCheckpoint {
             block: checkpoint.block,
             tx_index: checkpoint.tx_index,
         };
-        self.0.put::<tables::MessagingCheckpoints>(id.to_string(), value)?;
+        self.0.put::<tables::MessagingCheckpoints>(MESSAGING_CHECKPOINT_KEY.to_string(), value)?;
+        Ok(())
+    }
+
+    fn delete_messaging_checkpoint(&self) -> ProviderResult<()> {
+        self.0
+            .delete::<tables::MessagingCheckpoints>(MESSAGING_CHECKPOINT_KEY.to_string(), None)?;
         Ok(())
     }
 }
@@ -1033,7 +1041,7 @@ mod tests {
     use katana_provider_api::state::StateFactoryProvider;
     use katana_provider_api::transaction::TransactionProvider;
 
-    use crate::{DbProviderFactory, ProviderFactory};
+    use crate::{DbProviderFactory, MutableProvider, ProviderFactory};
 
     fn create_dummy_block() -> SealedBlockWithStatus {
         let header = Header { parent_hash: 199u8.into(), number: 0, ..Default::default() };
@@ -1245,4 +1253,65 @@ mod tests {
         assert_eq!(storage1, felt!("100"));
         assert_eq!(storage2, felt!("200"));
     }
+
+    #[test]
+    fn delete_messaging_checkpoint_removes_existing_row() {
+        use katana_provider_api::messaging::{MessagingCheckpoint, MessagingCheckpointProvider};
+
+        let factory = create_db_provider();
+        let provider = factory.provider_mut();
+
+        provider.set_messaging_checkpoint(&MessagingCheckpoint { block: 42, tx_index: 3 }).unwrap();
+        provider.commit().unwrap();
+
+        let provider = factory.provider_mut();
+        provider.delete_messaging_checkpoint().unwrap();
+        provider.commit().unwrap();
+
+        let provider = factory.provider_mut();
+        let cp = provider.messaging_checkpoint().unwrap();
+        assert!(cp.is_none(), "checkpoint row should be gone after delete");
+    }
+
+    #[test]
+    fn delete_messaging_checkpoint_is_noop_when_absent() {
+        use katana_provider_api::messaging::MessagingCheckpointProvider;
+
+        let factory = create_db_provider();
+        let provider = factory.provider_mut();
+
+        provider.delete_messaging_checkpoint().expect("delete of absent row succeeds");
+        provider.commit().unwrap();
+
+        let provider = factory.provider_mut();
+        let cp = provider.messaging_checkpoint().unwrap();
+        assert!(cp.is_none());
+    }
+
+    /// set → delete → set must leave the table in a usable state, ending with
+    /// the most recent write visible. Exercises the post-delete write path that
+    /// would silently break if MDBX cached the dropped key.
+    #[test]
+    fn set_then_delete_then_set_round_trip() {
+        use katana_provider_api::messaging::{MessagingCheckpoint, MessagingCheckpointProvider};
+
+        let factory = create_db_provider();
+
+        let provider = factory.provider_mut();
+        provider.set_messaging_checkpoint(&MessagingCheckpoint { block: 1, tx_index: 1 }).unwrap();
+        provider.commit().unwrap();
+
+        let provider = factory.provider_mut();
+        provider.delete_messaging_checkpoint().unwrap();
+        provider.commit().unwrap();
+
+        let provider = factory.provider_mut();
+        provider.set_messaging_checkpoint(&MessagingCheckpoint { block: 9, tx_index: 4 }).unwrap();
+        provider.commit().unwrap();
+
+        let provider = factory.provider_mut();
+        let cp = provider.messaging_checkpoint().unwrap().expect("final set visible");
+        assert_eq!(cp.block, 9);
+        assert_eq!(cp.tx_index, 4);
+    }
 }
diff --git a/crates/storage/provider/provider/src/providers/fork/mod.rs b/crates/storage/provider/provider/src/providers/fork/mod.rs
index c482a75c8..6b934d674 100644
--- a/crates/storage/provider/provider/src/providers/fork/mod.rs
+++ b/crates/storage/provider/provider/src/providers/fork/mod.rs
@@ -688,16 +688,16 @@ impl<Tx1: DbTxMut> StageCheckpointProvider for ForkedProvider<Tx1> {
 }
 
 impl<Tx1: DbTxMut> MessagingCheckpointProvider for ForkedProvider<Tx1> {
-    fn messaging_checkpoint(&self, id: &str) -> ProviderResult<Option<MessagingCheckpoint>> {
-        self.local_db.messaging_checkpoint(id)
+    fn messaging_checkpoint(&self) -> ProviderResult<Option<MessagingCheckpoint>> {
+        self.local_db.messaging_checkpoint()
     }
 
-    fn set_messaging_checkpoint(
-        &self,
-        id: &str,
-        checkpoint: &MessagingCheckpoint,
-    ) -> ProviderResult<()> {
-        self.local_db.set_messaging_checkpoint(id, checkpoint)
+    fn set_messaging_checkpoint(&self, checkpoint: &MessagingCheckpoint) -> ProviderResult<()> {
+        self.local_db.set_messaging_checkpoint(checkpoint)
+    }
+
+    fn delete_messaging_checkpoint(&self) -> ProviderResult<()> {
+        self.local_db.delete_messaging_checkpoint()
     }
 }