stub gateway pipeline

common/nym-lp-data/src/lib.rs

@@ -137,6 +137,21 @@ impl<Ts, D, Opts, NdId> PipelineData<Ts, D, Opts, NdId> {
         }
     }
 
+    /// Apply `op` to the options component, leaving the timestamp, data, and
+    /// destination unchanged.
+    ///
+    /// `No` can differ from `O`, so this also acts as a type transform.
+    pub fn options_transform<F, No>(self, mut op: F) -> PipelineData<Ts, D, No, NdId>
+    where
+        F: FnMut(Opts) -> No,
+    {
+        PipelineData {
+            data: self.data,
+            options: op(self.options),
+            dst: self.dst,
+        }
+    }
+
     /// Drop the pipeline options, producing a plain addressed payload.
     pub fn into_addressed(self) -> AddressedTimedData<Ts, D, NdId> {
         AddressedTimedData {

common/nym-lp-data/src/packet/frame.rs

@@ -15,8 +15,10 @@ pub enum LpFrameKind {
     Forward = 2,
     SphinxStream = 3,
     FragmentedData = 4,
-    SphinxPacket = 5,
-    OutfoxPacket = 6,
+    SphinxPacket = 5,        // Sphinx Packet to process, delay and forward
+    OutfoxPacket = 6,        // Outfox Packet to process, delay and forward
+    ForwardSphinxPacket = 7, // Sphinx Packet to immediately forward
+    ForwardOutfoxPacket = 8, // Outfox Packet to immediately forward
 
     #[num_enum(catch_all)]
     Unknown(u16),

nym-node/src/node/lp/data/handler/messages/gateway.rs

@@ -0,0 +1,113 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use nym_lp_data::packet::frame::{LpFrameAttributes, LpFrameHeader, LpFrameKind};
+use nym_node_metrics::mixnet::PacketKind;
+use nym_sphinx_forwarding::packet::MixPacketFormattingError;
+use nym_sphinx_params::SphinxKeyRotation;
+use nym_topology::NodeId;
+
+use crate::node::lp::data::handler::{
+    error::LpDataHandlerError,
+    messages::{MixMessage, SphinxMixMessage},
+};
+
+/// Message types supported by gateways. Gateways accept everything mixnodes
+/// do (carried as [`GatewayMessage::Mix`]) plus the forward-only variants
+/// that bypass sphinx/outfox processing.
+#[derive(Debug, Clone, Copy)]
+pub enum GatewayMessage {
+    Mix(MixMessage),
+    ForwardSphinx(ForwardSphinxMixMessage),
+    ForwardOutfox(ForwardOutfoxMixMessage),
+}
+
+impl GatewayMessage {
+    pub fn from_frame_header(header: LpFrameHeader) -> Result<Self, LpDataHandlerError> {
+        match header.kind {
+            LpFrameKind::SphinxPacket | LpFrameKind::OutfoxPacket => {
+                Ok(GatewayMessage::Mix(MixMessage::from_frame_header(header)?))
+            }
+            LpFrameKind::ForwardSphinxPacket => Ok(GatewayMessage::ForwardSphinx(
+                header.frame_attributes.try_into()?,
+            )),
+            LpFrameKind::ForwardOutfoxPacket => Ok(GatewayMessage::ForwardOutfox(
+                header.frame_attributes.try_into()?,
+            )),
+            _ => Err(LpDataHandlerError::UnsupportedLpFrameKind { typ: header.kind }),
+        }
+    }
+}
+
+impl From<GatewayMessage> for PacketKind {
+    fn from(value: GatewayMessage) -> Self {
+        match value {
+            GatewayMessage::Mix(msg) => msg.into(),
+            GatewayMessage::ForwardSphinx(_) => PacketKind::LpSphinx,
+            GatewayMessage::ForwardOutfox(_) => PacketKind::LpOutfox,
+        }
+    }
+}
+
+impl From<MixMessage> for GatewayMessage {
+    fn from(value: MixMessage) -> Self {
+        GatewayMessage::Mix(value)
+    }
+}
+
+impl From<GatewayMessage> for LpFrameHeader {
+    fn from(value: GatewayMessage) -> Self {
+        match value {
+            GatewayMessage::Mix(msg) => msg.into(),
+            GatewayMessage::ForwardSphinx(msg) => {
+                LpFrameHeader::new(LpFrameKind::ForwardSphinxPacket, msg)
+            }
+            GatewayMessage::ForwardOutfox(msg) => {
+                LpFrameHeader::new(LpFrameKind::ForwardOutfoxPacket, msg)
+            }
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct ForwardSphinxMixMessage {
+    pub key_rotation: SphinxKeyRotation,
+    pub next_hop: NodeId,
+}
+
+impl TryFrom<LpFrameAttributes> for ForwardSphinxMixMessage {
+    type Error = LpDataHandlerError;
+
+    fn try_from(value: LpFrameAttributes) -> Result<Self, Self::Error> {
+        let key_rotation = value[0]
+            .try_into()
+            .map_err(MixPacketFormattingError::InvalidKeyRotation)?;
+        // SAFETY : slice to array conversion with correct size
+        #[allow(clippy::unwrap_used)]
+        let next_hop = NodeId::from_be_bytes(value[1..5].try_into().unwrap());
+        Ok(ForwardSphinxMixMessage {
+            key_rotation,
+            next_hop,
+        })
+    }
+}
+
+impl From<ForwardSphinxMixMessage> for LpFrameAttributes {
+    fn from(value: ForwardSphinxMixMessage) -> Self {
+        let mut attrs = [0; 14];
+        attrs[0] = value.key_rotation as u8;
+        attrs[1..5].copy_from_slice(&value.next_hop.to_be_bytes());
+        attrs
+    }
+}
+
+impl From<ForwardSphinxMixMessage> for SphinxMixMessage {
+    fn from(value: ForwardSphinxMixMessage) -> Self {
+        SphinxMixMessage {
+            key_rotation: value.key_rotation,
+        }
+    }
+}
+
+// For now there are no differences. We can augment this variant when we will need it
+pub type ForwardOutfoxMixMessage = ForwardSphinxMixMessage;

nym-node/src/node/lp/data/handler/messages/mixnode.rs

@@ -2,18 +2,42 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use nym_lp_data::packet::frame::{LpFrameAttributes, LpFrameHeader, LpFrameKind};
+use nym_node_metrics::mixnet::PacketKind;
 use nym_sphinx_forwarding::packet::MixPacketFormattingError;
 use nym_sphinx_params::SphinxKeyRotation;
 
 use crate::node::lp::data::handler::error::LpDataHandlerError;
 
-/// Message types supported by mixnodes
+/// Message types supported by mixnodes.
 #[derive(Debug, Clone, Copy)]
 pub enum MixMessage {
     Sphinx(SphinxMixMessage),
     Outfox(OutfoxMixMessage),
 }
 
+impl MixMessage {
+    pub fn from_frame_header(header: LpFrameHeader) -> Result<Self, LpDataHandlerError> {
+        match header.kind {
+            LpFrameKind::SphinxPacket => {
+                Ok(MixMessage::Sphinx(header.frame_attributes.try_into()?))
+            }
+            LpFrameKind::OutfoxPacket => {
+                Ok(MixMessage::Outfox(header.frame_attributes.try_into()?))
+            }
+            _ => Err(LpDataHandlerError::UnsupportedLpFrameKind { typ: header.kind }),
+        }
+    }
+}
+
+impl From<MixMessage> for PacketKind {
+    fn from(value: MixMessage) -> Self {
+        match value {
+            MixMessage::Sphinx(_) => PacketKind::LpSphinx,
+            MixMessage::Outfox(_) => PacketKind::LpOutfox,
+        }
+    }
+}
+
 impl From<MixMessage> for LpFrameHeader {
     fn from(value: MixMessage) -> Self {
         match value {
@@ -22,17 +46,6 @@ impl From<MixMessage> for LpFrameHeader {
         }
     }
 }
-impl TryFrom<LpFrameHeader> for MixMessage {
-    type Error = LpDataHandlerError;
-
-    fn try_from(value: LpFrameHeader) -> Result<Self, Self::Error> {
-        match value.kind {
-            LpFrameKind::SphinxPacket => Ok(MixMessage::Sphinx(value.frame_attributes.try_into()?)),
-            LpFrameKind::OutfoxPacket => Ok(MixMessage::Outfox(value.frame_attributes.try_into()?)),
-            other => Err(LpDataHandlerError::UnsupportedLpFrameKind { typ: other })?,
-        }
-    }
-}
 
 #[derive(Debug, Clone, Copy)]
 pub struct SphinxMixMessage {

nym-node/src/node/lp/data/handler/messages/mod.rs

@@ -0,0 +1,8 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+mod gateway;
+mod mixnode;
+
+pub use gateway::{ForwardOutfoxMixMessage, ForwardSphinxMixMessage, GatewayMessage};
+pub use mixnode::{MixMessage, OutfoxMixMessage, SphinxMixMessage};

nym-node/src/node/lp/data/handler/mod.rs

@@ -19,7 +19,7 @@ use crate::node::lp::data::PACKET_BUFFER_SIZE;
 use crate::node::lp::data::handler::pipeline::MixnodeDataPipeline;
 use crate::node::lp::data::shared::SharedLpDataState;
 use nym_lp_data::AddressedTimedData;
-use nym_lp_data::mixnodes::traits::MixnodeProcessingPipeline;
+use nym_lp_data::nymnodes::traits::NymNodeProcessingPipeline;
 use nym_lp_data::packet::{EncryptedLpPacket, MalformedLpPacketError};
 use nym_metrics::inc;
 use rand::rngs::OsRng;

nym-node/src/node/lp/data/handler/pipeline/gateway.rs

@@ -0,0 +1,161 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use std::{net::SocketAddr, sync::Arc, time::Instant};
+
+use nym_lp_data::{
+    AddressedTimedData, PipelinePayload, TimedData, TimedPayload,
+    common::traits::{
+        Framing, FramingUnwrap, Transport, TransportUnwrap, WireUnwrappingPipeline,
+        WireWrappingPipeline,
+    },
+    nymnodes::traits::NymNodeProcessingPipeline,
+    packet::{EncryptedLpPacket, LpFrame, LpHeader, MalformedLpPacketError, frame::LpFrameHeader},
+};
+use rand::Rng;
+use tracing::warn;
+
+use crate::node::{
+    lp::data::{
+        handler::{
+            messages::GatewayMessage,
+            pipeline::{MixnodeDataPipeline, wire::WirePipeline},
+            processing,
+        },
+        shared::SharedLpDataState,
+    },
+    routing_filter::RoutingFilter,
+};
+
+pub(crate) struct GatewayDataPipeline<R> {
+    state: Arc<SharedLpDataState>,
+    wire: WirePipeline<R>,
+}
+
+impl<R: Rng> GatewayDataPipeline<R> {
+    pub(crate) fn new(state: Arc<SharedLpDataState>, rng: R) -> Self {
+        Self {
+            state: state.clone(),
+            wire: WirePipeline::new(state, rng),
+        }
+    }
+}
+
+// Processing logic
+impl<R: Rng>
+    NymNodeProcessingPipeline<
+        Instant,
+        EncryptedLpPacket,
+        GatewayMessage,
+        GatewayMessage,
+        SocketAddr,
+    > for GatewayDataPipeline<R>
+{
+    fn mix(
+        &mut self,
+        message_kind: GatewayMessage,
+        payload: TimedPayload<Instant>,
+        timestamp: Instant,
+    ) -> Vec<PipelinePayload<Instant, GatewayMessage, SocketAddr>> {
+        let processing_result = match message_kind {
+            GatewayMessage::Mix(msg) => {
+                return MixnodeDataPipeline::<R>::process_mix_packet(
+                    &self.state,
+                    msg,
+                    payload,
+                    timestamp,
+                )
+                .into_iter()
+                .map(|payload| payload.options_transform(Into::into))
+                .collect();
+            }
+
+            // SW Need to check for bandwidth and routing filter as well
+            GatewayMessage::ForwardSphinx(metadata) => {
+                processing::sphinx::process_forward(&self.state, payload, metadata)
+            }
+            GatewayMessage::ForwardOutfox(metadata) => {
+                processing::outfox::process_forward(&self.state, payload, metadata)
+            }
+        };
+
+        Vec::new()
+    }
+}
+
+// ============== Wire wrap/unwrap: pure delegation to WirePipeline ==============
+
+impl<R: Rng> Framing<Instant, GatewayMessage, SocketAddr> for GatewayDataPipeline<R> {
+    type Frame = LpFrame;
+    const OVERHEAD_SIZE: usize = LpFrameHeader::SIZE;
+
+    fn to_frame(
+        &mut self,
+        payload: PipelinePayload<Instant, GatewayMessage, SocketAddr>,
+        frame_size: usize,
+    ) -> Vec<AddressedTimedData<Instant, Self::Frame, SocketAddr>> {
+        let frame = LpFrame {
+            header: payload.options.into(),
+            content: payload.data.data.into(),
+        };
+        self.wire
+            .message_to_frame(payload.data.timestamp, frame, payload.dst, frame_size)
+    }
+}
+
+impl<R: Rng> Transport<Instant, EncryptedLpPacket, SocketAddr> for GatewayDataPipeline<R> {
+    type Frame = LpFrame;
+    const OVERHEAD_SIZE: usize = LpHeader::SIZE;
+
+    fn to_transport_packet(
+        &mut self,
+        frame: AddressedTimedData<Instant, Self::Frame, SocketAddr>,
+    ) -> AddressedTimedData<Instant, EncryptedLpPacket, SocketAddr> {
+        self.wire.frame_to_packet(frame)
+    }
+}
+
+impl<R: Rng> WireWrappingPipeline<Instant, EncryptedLpPacket, GatewayMessage, SocketAddr>
+    for GatewayDataPipeline<R>
+{
+    fn packet_size(&self) -> usize {
+        nym_lp_data::packet::MTU
+    }
+}
+
+impl<R: Rng> TransportUnwrap<Instant, EncryptedLpPacket> for GatewayDataPipeline<R> {
+    type Frame = LpFrame;
+    type Error = MalformedLpPacketError;
+
+    fn packet_to_frame(
+        &mut self,
+        packet: EncryptedLpPacket,
+        timestamp: Instant,
+    ) -> Result<TimedData<Instant, Self::Frame>, Self::Error> {
+        self.wire.packet_to_frame(packet, timestamp)
+    }
+}
+
+impl<R: Rng> FramingUnwrap<Instant, GatewayMessage> for GatewayDataPipeline<R> {
+    type Frame = LpFrame;
+
+    fn frame_to_message(
+        &mut self,
+        frame: TimedData<Instant, Self::Frame>,
+    ) -> Option<(TimedPayload<Instant>, GatewayMessage)> {
+        let reassembled = self.wire.frame_to_maybe_message(frame)?;
+        let message_kind = GatewayMessage::from_frame_header(reassembled.data.header)
+            .inspect_err(|e| warn!("{e}"))
+            .ok()?;
+        self.state.message_received(message_kind);
+        Some((
+            TimedPayload::new(reassembled.timestamp, reassembled.data.content.to_vec()),
+            message_kind,
+        ))
+    }
+}
+
+impl<R: Rng> WireUnwrappingPipeline<Instant, EncryptedLpPacket, GatewayMessage>
+    for GatewayDataPipeline<R>
+{
+}

nym-node/src/node/lp/data/handler/pipeline/mixnode.rs

@@ -9,64 +9,50 @@ use nym_lp_data::{
         Framing, FramingUnwrap, Transport, TransportUnwrap, WireUnwrappingPipeline,
         WireWrappingPipeline,
     },
-    fragmentation::fragment::fragment_lp_message,
-    mixnodes::traits::MixnodeProcessingPipeline,
-    packet::{
-        EncryptedLpPacket, LpFrame, LpHeader, LpPacket, MalformedLpPacketError,
-        frame::{LpFrameHeader, LpFrameKind},
-    },
+    nymnodes::traits::NymNodeProcessingPipeline,
+    packet::{EncryptedLpPacket, LpFrame, LpHeader, MalformedLpPacketError, frame::LpFrameHeader},
 };
 use rand::Rng;
 use tracing::warn;
 
 use crate::node::{
     lp::data::{
-        handler::{messages::MixMessage, processing},
+        handler::{messages::MixMessage, pipeline::wire::WirePipeline, processing},
         shared::SharedLpDataState,
     },
     routing_filter::RoutingFilter,
 };
 
-pub struct MixnodeDataPipeline<R>
-where
-    R: Rng,
-{
-    /// Shared data state
+pub(crate) struct MixnodeDataPipeline<R> {
     state: Arc<SharedLpDataState>,
-    rng: R,
+    wire: WirePipeline<R>,
 }
 
-impl<R> MixnodeDataPipeline<R>
-where
-    R: Rng,
-{
-    pub fn new(state: Arc<SharedLpDataState>, rng: R) -> Self {
-        Self { state, rng }
+impl<R: Rng> MixnodeDataPipeline<R> {
+    pub(crate) fn new(state: Arc<SharedLpDataState>, rng: R) -> Self {
+        Self {
+            state: state.clone(),
+            wire: WirePipeline::new(state, rng),
+        }
     }
-}
 
-// Mixing logic
-impl<R> MixnodeProcessingPipeline<Instant, EncryptedLpPacket, MixMessage, MixMessage, SocketAddr>
-    for MixnodeDataPipeline<R>
-where
-    R: Rng,
-{
-    fn mix(
-        &mut self,
+    // Processing logic so gateways can reuse them
+    pub fn process_mix_packet(
+        shared_state: &SharedLpDataState,
         message_kind: MixMessage,
         payload: TimedPayload<Instant>,
         _: Instant,
     ) -> Vec<PipelinePayload<Instant, MixMessage, SocketAddr>> {
         let processing_result = match message_kind {
             MixMessage::Sphinx(metadata) => {
-                processing::sphinx::process(&self.state, payload, metadata)
+                processing::sphinx::process(shared_state, payload, metadata)
             }
             MixMessage::Outfox(metadata) => {
-                processing::outfox::process(&self.state, payload, metadata)
+                processing::outfox::process(shared_state, payload, metadata)
             }
         };
 
-        self.state.update_processing_metrics(&processing_result);
+        shared_state.update_processing_metrics(&processing_result);
 
         let packet_to_forward = match processing_result {
             Ok(packet) => packet,
@@ -77,13 +63,13 @@ where
         };
 
         let next_hop = packet_to_forward.dst;
-        if !self.state.routing_filter.should_route(next_hop.ip()) {
+        if !shared_state.routing_filter.should_route(next_hop.ip()) {
             warn!(
                 event = "packet.dropped.routing_filter",
                 next_hop = %next_hop,
                 "dropping packet: egress address does not belong to any known node"
             );
-            self.state.routing_filter_dropped(next_hop);
+            shared_state.routing_filter_dropped(next_hop);
             Vec::new()
         } else {
             vec![packet_to_forward]
@@ -91,12 +77,25 @@ where
     }
 }
 
-impl<R> Framing<Instant, MixMessage, SocketAddr> for MixnodeDataPipeline<R>
-where
-    R: Rng,
+// Mixing logic
+impl<R: Rng>
+    NymNodeProcessingPipeline<Instant, EncryptedLpPacket, MixMessage, MixMessage, SocketAddr>
+    for MixnodeDataPipeline<R>
 {
-    type Frame = LpFrame;
+    fn mix(
+        &mut self,
+        message_kind: MixMessage,
+        payload: TimedPayload<Instant>,
+        timestamp: Instant,
+    ) -> Vec<PipelinePayload<Instant, MixMessage, SocketAddr>> {
+        MixnodeDataPipeline::<R>::process_mix_packet(&self.state, message_kind, payload, timestamp)
+    }
+}
 
+// ============== Wire wrap/unwrap: pure delegation to WirePipeline ==============
+
+impl<R: Rng> Framing<Instant, MixMessage, SocketAddr> for MixnodeDataPipeline<R> {
+    type Frame = LpFrame;
     const OVERHEAD_SIZE: usize = LpFrameHeader::SIZE;
 
     fn to_frame(
@@ -108,54 +107,32 @@ where
             header: payload.options.into(),
             content: payload.data.data.into(),
         };
-
-        let output_frames = if frame.len() > frame_size {
-            fragment_lp_message(&mut self.rng, frame, frame_size)
-                .into_iter()
-                .map(|f| f.into_lp_frame())
-                .collect()
-        } else {
-            vec![frame]
-        };
-
-        output_frames
-            .into_iter()
-            .map(|f| AddressedTimedData::new_addressed(payload.data.timestamp, f, payload.dst))
-            .collect()
+        self.wire
+            .message_to_frame(payload.data.timestamp, frame, payload.dst, frame_size)
     }
 }
 
-impl<R> Transport<Instant, EncryptedLpPacket, SocketAddr> for MixnodeDataPipeline<R>
-where
-    R: Rng,
-{
+impl<R: Rng> Transport<Instant, EncryptedLpPacket, SocketAddr> for MixnodeDataPipeline<R> {
     type Frame = LpFrame;
-
     const OVERHEAD_SIZE: usize = LpHeader::SIZE;
 
     fn to_transport_packet(
         &mut self,
         frame: AddressedTimedData<Instant, Self::Frame, SocketAddr>,
     ) -> AddressedTimedData<Instant, EncryptedLpPacket, SocketAddr> {
-        // Here be LP encryption. For not, just wrap into an EncryptedLpPacket, we don't care at reception anyway
-        frame.data_transform(|f| LpPacket::new(LpHeader::new(0, 0, 0), f).encode())
+        self.wire.frame_to_packet(frame)
     }
 }
 
-impl<R> WireWrappingPipeline<Instant, EncryptedLpPacket, MixMessage, SocketAddr>
+impl<R: Rng> WireWrappingPipeline<Instant, EncryptedLpPacket, MixMessage, SocketAddr>
     for MixnodeDataPipeline<R>
-where
-    R: Rng,
 {
     fn packet_size(&self) -> usize {
         nym_lp_data::packet::MTU
     }
 }
 
-impl<R> TransportUnwrap<Instant, EncryptedLpPacket> for MixnodeDataPipeline<R>
-where
-    R: Rng,
-{
+impl<R: Rng> TransportUnwrap<Instant, EncryptedLpPacket> for MixnodeDataPipeline<R> {
     type Frame = LpFrame;
     type Error = MalformedLpPacketError;
 
@@ -164,79 +141,31 @@ where
         packet: EncryptedLpPacket,
         timestamp: Instant,
     ) -> Result<TimedData<Instant, Self::Frame>, Self::Error> {
-        // Here be LP decryption. For now we do as is it's not encrypted
-        let lp_packet = LpPacket::decode(packet).inspect_err(|_| {
-            self.state.malformed_packet();
-        })?;
-        Ok(TimedData {
-            timestamp,
-            data: lp_packet.into_frame(),
-        })
+        self.wire.packet_to_frame(packet, timestamp)
     }
 }
 
-impl<R> FramingUnwrap<Instant, MixMessage> for MixnodeDataPipeline<R>
-where
-    R: Rng,
-{
+impl<R: Rng> FramingUnwrap<Instant, MixMessage> for MixnodeDataPipeline<R> {
     type Frame = LpFrame;
+
     fn frame_to_message(
         &mut self,
         frame: TimedData<Instant, Self::Frame>,
     ) -> Option<(TimedPayload<Instant>, MixMessage)> {
-        let reassembled_frame = if frame.data.kind() == LpFrameKind::FragmentedData {
-            let fragment = frame
-                .data
-                .try_into()
-                .inspect_err(|e| {
-                    tracing::error!("Failed to recover a fragment : {e}");
-                    self.state.malformed_packet();
-                })
-                .ok()?;
-            let message = self
-                .state
-                .message_reconstructor
-                .insert_new_fragment(fragment, frame.timestamp)?
-                .inspect_err(|e| {
-                    tracing::error!("Failed to recover a frame : {e}");
-                    self.state.malformed_packet();
-                })
-                .ok()?;
-
-            TimedData::new(frame.timestamp, message)
-        } else {
-            frame
-        };
-
-        match reassembled_frame.data.kind() {
-            LpFrameKind::FragmentedData => {
-                warn!(
-                    "Fragmented data inside fragmented data, it shouldn't happen. Dropping the message"
-                );
-                None
-            }
-            _ => {
-                let message_kind = reassembled_frame
-                    .data
-                    .header
-                    .try_into()
-                    .inspect_err(|e| warn!("{e}"))
-                    .ok()?;
-                self.state.message_received(&message_kind);
-                Some((
-                    TimedPayload::new(
-                        reassembled_frame.timestamp,
-                        reassembled_frame.data.content.to_vec(),
-                    ),
-                    message_kind,
-                ))
-            }
-        }
+        let reassembled = self.wire.frame_to_maybe_message(frame)?;
+        let message_kind = MixMessage::from_frame_header(reassembled.data.header)
+            .inspect_err(|e| warn!("{e}"))
+            .ok()?;
+        self.state.message_received(message_kind);
+        Some((
+            TimedPayload::new(reassembled.timestamp, reassembled.data.content.to_vec()),
+            message_kind,
+        ))
     }
 }
 
-impl<R> WireUnwrappingPipeline<Instant, EncryptedLpPacket, MixMessage> for MixnodeDataPipeline<R> where
-    R: Rng
+impl<R: Rng> WireUnwrappingPipeline<Instant, EncryptedLpPacket, MixMessage>
+    for MixnodeDataPipeline<R>
 {
 }
 
@@ -251,7 +180,7 @@ mod tests {
     use nym_lp_data::common::traits::WireWrappingPipeline;
     use nym_lp_data::fragmentation::fragment::fragment_lp_message;
     use nym_lp_data::fragmentation::reconstruction::MessageReconstructor;
-    use nym_lp_data::mixnodes::traits::MixnodeProcessingPipeline;
+    use nym_lp_data::nymnodes::traits::NymNodeProcessingPipeline;
     use nym_lp_data::packet::{
         EncryptedLpPacket, LpFrame, LpHeader, LpPacket, OuterHeader, frame::LpFrameHeader, version,
     };

nym-node/src/node/lp/data/handler/pipeline/mod.rs

@@ -0,0 +1,12 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+//! Per-role data pipelines (mixnode, gateway). The wire wrapping/unwrapping
+//! shared between them lives in [`wire`].
+
+mod gateway;
+mod mixnode;
+mod wire;
+
+pub(crate) use gateway::GatewayDataPipeline;
+pub(crate) use mixnode::MixnodeDataPipeline;

nym-node/src/node/lp/data/handler/pipeline/wire.rs

@@ -0,0 +1,130 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+//! Wire wrapping / unwrapping shared by the mixnode and gateway data pipelines.
+//!
+//! Everything below the application-message layer (LP packet encode/decode and
+//! fragment reassembly) is identical between roles, so it lives here. The
+//! per-role pipelines embed a [`WirePipeline`] and delegate their
+//! [`Framing`]/[`Transport`]/[`*Unwrap`] trait methods to it.
+//!
+//! [`Framing`]: nym_lp_data::common::traits::Framing
+//! [`Transport`]: nym_lp_data::common::traits::Transport
+
+use std::{net::SocketAddr, sync::Arc, time::Instant};
+
+use nym_lp_data::{
+    AddressedTimedData, TimedData,
+    fragmentation::fragment::fragment_lp_message,
+    packet::{
+        EncryptedLpPacket, LpFrame, LpHeader, LpPacket, MalformedLpPacketError, frame::LpFrameKind,
+    },
+};
+use rand::Rng;
+use tracing::warn;
+
+use crate::node::lp::data::shared::SharedLpDataState;
+
+/// Wire-layer pipeline: handles LP packet encode/decode and fragment
+/// reassembly. Has no knowledge of the application message type carried in
+/// the frame.
+pub struct WirePipeline<R> {
+    state: Arc<SharedLpDataState>,
+    rng: R,
+}
+
+impl<R: Rng> WirePipeline<R> {
+    pub fn new(state: Arc<SharedLpDataState>, rng: R) -> Self {
+        Self { state, rng }
+    }
+
+    /// Wrap an [`LpFrame`] into an [`EncryptedLpPacket`] for the wire.
+    pub fn frame_to_packet(
+        &mut self,
+        frame: AddressedTimedData<Instant, LpFrame, SocketAddr>,
+    ) -> AddressedTimedData<Instant, EncryptedLpPacket, SocketAddr> {
+        // Here be LP encryption. For now, just wrap into an EncryptedLpPacket; we don't care at reception anyway
+        frame.data_transform(|f| LpPacket::new(LpHeader::new(0, 0, 0), f).encode())
+    }
+
+    /// Unwrap an [`EncryptedLpPacket`] off the wire into an [`LpFrame`].
+    pub fn packet_to_frame(
+        &mut self,
+        packet: EncryptedLpPacket,
+        timestamp: Instant,
+    ) -> Result<TimedData<Instant, LpFrame>, MalformedLpPacketError> {
+        // Here be LP decryption. For now we do as is, it's not encrypted
+        let lp_packet = LpPacket::decode(packet).inspect_err(|_| {
+            self.state.malformed_packet();
+        })?;
+        Ok(TimedData {
+            timestamp,
+            data: lp_packet.into_frame(),
+        })
+    }
+
+    /// Wrap an [`LpFrame`] into one or more addressed frames, fragmenting it
+    /// if its serialized length would exceed `frame_size`.
+    pub fn message_to_frame(
+        &mut self,
+        timestamp: Instant,
+        frame: LpFrame,
+        dst: SocketAddr,
+        frame_size: usize,
+    ) -> Vec<AddressedTimedData<Instant, LpFrame, SocketAddr>> {
+        let output_frames = if frame.len() > frame_size {
+            fragment_lp_message(&mut self.rng, frame, frame_size)
+                .into_iter()
+                .map(|f| f.into_lp_frame())
+                .collect()
+        } else {
+            vec![frame]
+        };
+
+        output_frames
+            .into_iter()
+            .map(|f| AddressedTimedData::new_addressed(timestamp, f, dst))
+            .collect()
+    }
+
+    /// If the frame carries a fragment, attempt reassembly; otherwise return
+    /// the frame as-is. Returns `None` when more fragments are needed or
+    /// reassembly fails. The returned frame is guaranteed not to be a
+    /// fragment.
+    pub fn frame_to_maybe_message(
+        &mut self,
+        frame: TimedData<Instant, LpFrame>,
+    ) -> Option<TimedData<Instant, LpFrame>> {
+        let reassembled = if frame.data.kind() == LpFrameKind::FragmentedData {
+            let fragment = frame
+                .data
+                .try_into()
+                .inspect_err(|e| {
+                    tracing::error!("Failed to recover a fragment : {e}");
+                    self.state.malformed_packet();
+                })
+                .ok()?;
+            let message = self
+                .state
+                .message_reconstructor
+                .insert_new_fragment(fragment, frame.timestamp)?
+                .inspect_err(|e| {
+                    tracing::error!("Failed to recover a frame : {e}");
+                    self.state.malformed_packet();
+                })
+                .ok()?;
+            TimedData::new(frame.timestamp, message)
+        } else {
+            frame
+        };
+
+        if reassembled.data.kind() == LpFrameKind::FragmentedData {
+            warn!(
+                "Fragmented data inside fragmented data, it shouldn't happen. Dropping the message"
+            );
+            None
+        } else {
+            Some(reassembled)
+        }
+    }
+}

nym-node/src/node/lp/data/handler/processing/outfox.rs

@@ -10,7 +10,7 @@ use tracing::warn;
 use crate::node::lp::data::{
     handler::{
         error::LpDataHandlerError,
-        messages::{MixMessage, OutfoxMixMessage},
+        messages::{ForwardOutfoxMixMessage, GatewayMessage, MixMessage, OutfoxMixMessage},
     },
     shared::SharedLpDataState,
 };
@@ -42,3 +42,11 @@ pub(crate) fn process(
         ))
     }
 }
+
+pub(crate) fn process_forward(
+    shared_state: &SharedLpDataState,
+    outfox_packet: TimedPayload<Instant>,
+    metadata: ForwardOutfoxMixMessage,
+) -> Result<PipelinePayload<Instant, GatewayMessage, SocketAddr>, LpDataHandlerError> {
+    todo!()
+}

nym-node/src/node/lp/data/handler/processing/sphinx.rs

@@ -12,7 +12,7 @@ use tracing::{error, warn};
 use crate::node::lp::data::{
     handler::{
         error::LpDataHandlerError,
-        messages::{MixMessage, SphinxMixMessage},
+        messages::{ForwardSphinxMixMessage, GatewayMessage, MixMessage, SphinxMixMessage},
     },
     shared::SharedLpDataState,
 };
@@ -87,3 +87,11 @@ pub(crate) fn process(
         }
     }
 }
+
+pub(crate) fn process_forward(
+    shared_state: &SharedLpDataState,
+    sphinx_packet: TimedPayload<Instant>,
+    metadata: ForwardSphinxMixMessage,
+) -> Result<PipelinePayload<Instant, GatewayMessage, SocketAddr>, LpDataHandlerError> {
+    todo!()
+}

nym-node/src/node/lp/data/shared.rs

@@ -1,26 +1,25 @@
 // Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: GPL-3.0-only
 
-use crate::config::Config;
-use crate::config::LpConfig;
-use crate::node::key_rotation::active_keys::ActiveSphinxKeys;
-use crate::node::key_rotation::active_keys::SphinxKeyGuard;
-use crate::node::lp::data::handler::error::LpDataHandlerError;
-use crate::node::lp::data::handler::messages::MixMessage;
-use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters;
-use crate::node::routing_filter::network_filter::NetworkRoutingFilter;
-use nym_lp_data::PipelinePayload;
-use nym_lp_data::fragmentation::reconstruction::MessageReconstructor;
-use nym_node_metrics::NymNodeMetrics;
-use nym_node_metrics::mixnet::PacketKind;
+use crate::{
+    config::{Config, LpConfig, NodeModes},
+    node::{
+        key_rotation::active_keys::{ActiveSphinxKeys, SphinxKeyGuard},
+        lp::data::handler::error::LpDataHandlerError,
+        replay_protection::bloomfilter::ReplayProtectionBloomfilters,
+        routing_filter::network_filter::NetworkRoutingFilter,
+    },
+};
+use nym_lp_data::{PipelinePayload, fragmentation::reconstruction::MessageReconstructor};
+use nym_node_metrics::{NymNodeMetrics, mixnet::PacketKind};
 use nym_sphinx_framing::processing::PacketProcessingError;
 use nym_sphinx_params::SphinxKeyRotation;
 use nym_task::ShutdownToken;
-use std::net::SocketAddr;
-use std::time::Duration;
-use std::time::Instant;
-use tracing::Span;
-use tracing::warn;
+use std::{
+    net::SocketAddr,
+    time::{Duration, Instant},
+};
+use tracing::{Span, warn};
 
 #[derive(Clone, Copy)]
 pub(crate) struct ProcessingConfig {
@@ -56,13 +55,6 @@ pub(crate) struct SharedLpDataState {
     pub shutdown_token: ShutdownToken,
 }
 
-fn message_kind_to_packet_kind(message_kind: &MixMessage) -> PacketKind {
-    match message_kind {
-        MixMessage::Sphinx { .. } => PacketKind::LpSphinx,
-        MixMessage::Outfox { .. } => PacketKind::LpOutfox,
-    }
-}
-
 impl SharedLpDataState {
     pub(crate) fn new(
         config: &Config,
@@ -123,10 +115,8 @@ impl SharedLpDataState {
         self.metrics.mixnet.lp_malformed_packet()
     }
 
-    pub(super) fn message_received(&self, message_kind: &MixMessage) {
-        self.metrics
-            .mixnet
-            .lp_message_received(message_kind_to_packet_kind(message_kind))
+    pub(super) fn message_received(&self, message_kind: impl Into<PacketKind>) {
+        self.metrics.mixnet.lp_message_received(message_kind.into())
     }
 
     pub(super) fn packet_forwarded(&self, dst: SocketAddr) {
@@ -163,7 +153,7 @@ impl SharedLpDataState {
     pub(super) fn update_processing_metrics(
         &self,
         processing_result: &Result<
-            PipelinePayload<Instant, MixMessage, SocketAddr>,
+            PipelinePayload<Instant, impl Clone + Into<PacketKind>, SocketAddr>,
             LpDataHandlerError,
         >,
     ) {
@@ -171,7 +161,7 @@ impl SharedLpDataState {
             Ok(packet) => {
                 self.metrics
                     .mixnet
-                    .lp_processed_message(message_kind_to_packet_kind(&packet.options));
+                    .lp_processed_message(packet.options.clone().into());
             }
             Err(LpDataHandlerError::PacketProcessingError(PacketProcessingError::PacketReplay)) => {
                 self.metrics.mixnet.lp_processing_replayed_packet();