feat(sequencer): enrich websocket transaction context

README.md

@@ -156,17 +156,18 @@ Notes:
 - `from_offset` is optional and defaults to `0`.
 - messages are JSON text frames.
 - binary fields are hex-encoded (`0x`-prefixed).
+- direct-input `block_timestamp` values are Unix seconds.
 - the current runtime enforces a subscriber cap of `64` and a catch-up cap of `50000` events.
-- if the requested catch-up window exceeds that cap, the server upgrades and then immediately closes the socket with close code `1008` (`POLICY`) and reason `catch-up window exceeded`.
+- if the requested catch-up window exceeds that cap, the server upgrades and then immediately closes the socket with close code `1008` (`POLICY`) and reason `catch-up window exceeded: live_start_offset=<u64>`; reconnecting at that offset starts from the current live head.
 
 Message shapes:
 
 ```json
-{ "kind": "user_op", "offset": 10, "sender": "0x...", "fee": 1, "data": "0x..." }
+{ "kind": "user_op", "offset": 10, "sender": "0x...", "nonce": 7, "fee": 1, "data": "0x...", "safe_block": 123, "batch_nonce": 4 }
 ```
 
 ```json
-{ "kind": "direct_input", "offset": 11, "sender": "0x...", "block_number": 123, "payload": "0x..." }
+{ "kind": "direct_input", "offset": 11, "sender": "0x...", "block_number": 123, "block_timestamp": 1700000000, "transaction_hash": "0x...", "payload": "0x...", "input_index": 42, "batch_nonce": 4 }
 ```
 
 Success response:

examples/app-core/src/application/wallet.rs

@@ -138,7 +138,7 @@ impl WalletApp {
         self.executed_input_count
     }
 
-    pub(crate) fn last_executed_safe_block(&self) -> u64 {
+    pub fn last_executed_safe_block(&self) -> u64 {
         self.last_executed_safe_block
     }
 

sequencer-core/src/broadcast.rs

@@ -1,24 +1,47 @@
 // (c) Cartesi and individual authors (see AUTHORS)
 // SPDX-License-Identifier: Apache-2.0 (see LICENSE)
 
-use crate::l2_tx::SequencedL2Tx;
+use crate::l2_tx::{DirectInput, ValidUserOp};
 use serde::{Deserialize, Serialize};
 
+/// One transaction from the sequencer's replay-then-live feed.
+///
+/// In addition to application inputs, each variant carries enough persisted
+/// ordering context for a subscriber to rebuild the soft suffix and reconcile
+/// it with L1 settlement.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(tag = "kind", rename_all = "snake_case")]
 pub enum BroadcastTxMessage {
     UserOp {
         offset: u64,
         sender: String,
+        /// Signed replay-protection nonce of the user operation.
+        /// A mirror must apply the op under exactly this nonce; synthesizing
+        /// it from mirror state mis-executes ops across feed gaps.
+        nonce: u32,
         /// Log-space fee exponent (base 129/128). See [`crate::fee`].
         fee: u16,
         data: String,
+        /// Safe L1 block committed by the covering frame.
+        safe_block: u64,
+        /// Nonce of the batch containing the covering frame.
+        batch_nonce: u64,
     },
     DirectInput {
         offset: u64,
         sender: String,
         block_number: u64,
         payload: String,
+        /// Per-application InputBox index of this direct input.
+        input_index: u64,
+        /// Nonce of the batch that drained this direct input. Lets a mirror
+        /// distinguish a direct already executed by the settled chain from one
+        /// still parked in its scheduler fridge (both have settled L1 inputs).
+        batch_nonce: u64,
+        /// Unix timestamp, in seconds, of the L1 block containing this direct input.
+        block_timestamp: u64,
+        /// Hash of the L1 transaction that carried this direct input.
+        transaction_hash: String,
     },
 }
 
@@ -30,20 +53,41 @@ impl BroadcastTxMessage {
         }
     }
 
-    pub fn from_offset_and_tx(offset: u64, tx: SequencedL2Tx) -> Self {
-        match tx {
-            SequencedL2Tx::UserOp(user_op) => Self::UserOp {
-                offset,
-                sender: user_op.sender.to_string(),
-                fee: user_op.fee,
-                data: alloy_primitives::hex::encode_prefixed(user_op.data.as_slice()),
-            },
-            SequencedL2Tx::Direct(direct) => Self::DirectInput {
-                offset,
-                sender: direct.sender.to_string(),
-                block_number: direct.block_number,
-                payload: alloy_primitives::hex::encode_prefixed(direct.payload.as_slice()),
-            },
+    pub fn from_user_op(
+        offset: u64,
+        user_op: ValidUserOp,
+        nonce: u32,
+        safe_block: u64,
+        batch_nonce: u64,
+    ) -> Self {
+        Self::UserOp {
+            offset,
+            sender: user_op.sender.to_string(),
+            nonce,
+            fee: user_op.fee,
+            data: alloy_primitives::hex::encode_prefixed(user_op.data.as_slice()),
+            safe_block,
+            batch_nonce,
+        }
+    }
+
+    pub fn from_direct_input(
+        offset: u64,
+        direct: DirectInput,
+        input_index: u64,
+        batch_nonce: u64,
+        block_timestamp: u64,
+        transaction_hash: alloy_primitives::B256,
+    ) -> Self {
+        Self::DirectInput {
+            offset,
+            sender: direct.sender.to_string(),
+            block_number: direct.block_number,
+            payload: alloy_primitives::hex::encode_prefixed(direct.payload.as_slice()),
+            input_index,
+            batch_nonce,
+            block_timestamp,
+            transaction_hash: alloy_primitives::hex::encode_prefixed(transaction_hash),
         }
     }
 }

sequencer/src/egress/api/subscribe.rs

@@ -72,12 +72,10 @@ async fn run_ws_session(
                 max_catchup_events,
                 "ws catch-up window exceeded; closing subscriber"
             );
-            close_with_frame(
-                &mut socket,
-                close_code::POLICY,
-                WS_CATCHUP_WINDOW_EXCEEDED_REASON,
-            )
-            .await;
+            let reason = format!(
+                "{WS_CATCHUP_WINDOW_EXCEEDED_REASON}: live_start_offset={live_start_offset}"
+            );
+            close_with_frame(&mut socket, close_code::POLICY, reason.as_str()).await;
             return;
         }
         Err(SubscribeError::OpenStorage { source }) => {

sequencer/src/egress/l2_tx_feed/mod.rs

@@ -14,11 +14,10 @@ pub use sequencer_core::broadcast::BroadcastTxMessage;
 use std::time::Duration;
 
 use alloy_primitives::Address;
-use sequencer_core::l2_tx::SequencedL2Tx;
 use tokio::sync::mpsc;
 
 use crate::runtime::shutdown::ShutdownSignal;
-use crate::storage::Storage;
+use crate::storage::{OrderedL2TxRow, Storage};
 
 #[derive(Debug, Clone, Copy)]
 pub struct L2TxFeedConfig {
@@ -182,7 +181,7 @@ fn run_subscription(
         }
 
         let txs = storage
-            .ordered_l2_txs_page_from(next_offset, page_size)
+            .ordered_l2_tx_rows_page_from(next_offset, page_size)
             .map_err(|source| SubscriptionError::LoadReplay {
                 offset: next_offset,
                 source,
@@ -193,35 +192,45 @@ fn run_subscription(
             continue;
         }
 
-        // The frame safe_block (third element) is a replay-only concern;
-        // the WS message shape doesn't carry it (review F7/WP5 owns any
-        // feed-protocol extension).
-        for (db_offset, tx, _frame_safe_block) in txs {
+        for row in txs {
             if shutdown.is_shutdown_requested() || events_tx.is_closed() {
                 return Ok(());
             }
 
-            next_offset = db_offset;
-
-            if should_filter_from_broadcast(&tx, batch_submitter_address) {
-                continue;
-            }
-
-            let event = BroadcastTxMessage::from_offset_and_tx(db_offset, tx);
+            next_offset = row.offset();
+            let event = match row {
+                OrderedL2TxRow::UserOp {
+                    offset,
+                    tx,
+                    nonce,
+                    safe_block,
+                    batch_nonce,
+                } => BroadcastTxMessage::from_user_op(offset, tx, nonce, safe_block, batch_nonce),
+                OrderedL2TxRow::DirectInput {
+                    offset,
+                    tx,
+                    input_index,
+                    batch_nonce,
+                    block_timestamp,
+                    transaction_hash,
+                    ..
+                } => {
+                    if batch_submitter_address == Some(tx.sender) {
+                        continue;
+                    }
+                    BroadcastTxMessage::from_direct_input(
+                        offset,
+                        tx,
+                        input_index,
+                        batch_nonce,
+                        block_timestamp,
+                        transaction_hash,
+                    )
+                }
+            };
             if events_tx.blocking_send(event).is_err() {
                 return Ok(());
             }
         }
     }
 }
-
-fn should_filter_from_broadcast(
-    tx: &SequencedL2Tx,
-    batch_submitter_address: Option<Address>,
-) -> bool {
-    matches!(
-        (tx, batch_submitter_address),
-        (SequencedL2Tx::Direct(direct), Some(batch_submitter_address))
-            if direct.sender == batch_submitter_address
-    )
-}