diff --git a/CHANGELOG.md b/CHANGELOG.md
index 43c8cc66f..ca990dfe0 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -16,14 +16,15 @@ Versioning: [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   + `streamDescriptorsForData(dataIndex, descriptors)`, the
   `DataStreamHandle` record, the offline `Muxer.pushData` / `pushDataTo`
   pair plus the `dataHandles()` / `dataStreamHandle(index)` accessors,
-  `MuxerFileSink.pushData` / `pushDataTo`, and `pushData` / `pushDataTo` /
-  `dataHandle()` on the srt and rtp `MuxSender`s. Pass-through / PTS /
-  size-ceiling semantics are those of the Rust `push_data` family (see the
-  muxer + pipeline entry below). Not yet exposed (recorded follow-ups): the
-  rtp `MountHandle` and srt `ManagedMuxSender` data push families, and
+  `MuxerFileSink.pushData` / `pushDataTo`, `pushData` / `pushDataTo` /
+  `dataHandle()` on the srt and rtp `MuxSender`s, the srt
+  `ManagedMuxSender.pushData` / `pushDataTo` / `dataHandle()` pair, and the
+  rtp `MountHandle.pushData` / `pushDataTo` / `dataHandle()` / `dataHandles()`
+  family. Pass-through / PTS / size-ceiling semantics are those of the Rust
+  `push_data` family (see the muxer + pipeline entry below). Not yet exposed:
   `data_handles_for_program` (the JVM `MuxerConfig` is single-program).
 
-### Added — C data-stream surface (ABI minor 12)
+### Added — C data-stream surface (ABI minor 12 → 13)
 
 - **tst-c** gains the data-stream mux surface, binding the Rust
   muxer/pipeline family below: `tst_data_stream_handle_t` plus seven new
@@ -36,6 +37,11 @@ Versioning: [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   those of the Rust `push_data` family (see the muxer + pipeline entry
   below). ABI minor 11 → 12; additive — no symbol removed, no signature or
   struct layout changed.
+- **tst-c** further gains the managed-sender and RTSP-mount data push pairs:
+  `tst_managed_mux_sender_send_data` / `_to` (behind `TST_HAS_SRT`) and
+  `tst_rtsp_mount_push_data` / `_to` (behind `TST_HAS_RTP`), mirroring the
+  `send_klv` / `push_klv` families. ABI minor 12 → 13; additive — no symbol
+  removed, no signature or struct layout changed.
 
 ### Added — Python data-stream surface (`tstrans.mpegts` + srt/rtp `MuxSender`)
 
@@ -48,6 +54,9 @@ Versioning: [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 - **`tstrans.srt.MuxSender` / `tstrans.rtp.MuxSender`** gain `push_data` /
   `push_data_to` / `data_handle()` — push raw private-PES payloads through a
   live sender, following the `push_klv` family shape.
+- **`tstrans.srt.ManagedMuxSender`** and **`tstrans.rtp.MountHandle`** gain the
+  same `push_data` / `push_data_to` / `data_handle()` trio, completing data
+  parity across every live sender / mount shell.
 
 ### Changed — `tio.transmux` passes private/application data streams through
 
@@ -60,6 +69,18 @@ Versioning: [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   always carry PTS and the demuxer substitutes 0 for a PTS-less source PES,
   so a source sample with no PTS re-emerges with a literal PTS of 0.
 
+### Changed — tst-jni panic safety + handle-decode hardening (`org.tstrans`)
+
+- An unexpected Rust panic inside any `org.tstrans` JVM native now surfaces as
+  a `java.lang.RuntimeException` instead of aborting the JVM process — every
+  native is wrapped in a `jni_catch` panic boundary (the JVM twin of the C
+  binding's `ffi_catch`).
+- `Muxer.pull`'s JNI-bridge failure cases now throw an unchecked
+  `RuntimeException` rather than a checked `MuxException` (the hot drain path
+  keeps no `throws` clause), and every handle-targeted `*To` push native now
+  rejects out-of-range / negative stream handles instead of silently
+  truncating them to a valid handle.
+
 ### Added — private/application data streams (muxer + pipeline)
 
 Arbitrary private PES carriage: declare a data elementary stream with an
diff --git a/bindings/c/core/src/lib.rs b/bindings/c/core/src/lib.rs
index 5b4a248c2..ec5eecd03 100644
--- a/bindings/c/core/src/lib.rs
+++ b/bindings/c/core/src/lib.rs
@@ -13,7 +13,7 @@
 //! transport surfaces are gated on the `rtp` cargo feature. The
 //! offline byte-feeding `tst_demuxer_*` surface is unconditional (no
 //! feature gate), as is the offline `tst_muxer_*` surface (un-gated from
-//! `srt` in ABI 0.9). ABI minor is `0.12` (see [`TST_ABI_VERSION_MINOR`]).
+//! `srt` in ABI 0.9). ABI minor is `0.13` (see [`TST_ABI_VERSION_MINOR`]).
 
 #![cfg_attr(not(feature = "std"), no_std)]
 #![allow(clippy::missing_safety_doc)] // every extern "C" fn has a /// header documenting the contract
@@ -186,7 +186,7 @@ pub const TST_ABI_VERSION_MAJOR: crate::c_types::c_int = 0;
 /// Minor version of the C ABI contract. See [`TST_ABI_VERSION_MAJOR`]
 /// for the bump policy.
 ///
-/// Cbindgen emits this as `#define TST_ABI_VERSION_MINOR 12` in the
+/// Cbindgen emits this as `#define TST_ABI_VERSION_MINOR 13` in the
 /// generated header. Runtime accessor: [`tst_get_abi_version_minor`].
 ///
 /// History (additive bumps only — major stays at 0 pre-1.0):
@@ -253,7 +253,14 @@ pub const TST_ABI_VERSION_MAJOR: crate::c_types::c_int = 0;
 ///   alongside video/KLV, mirroring the Rust `push_data` family.
 ///   Additive — no symbol removed, no signature or struct layout
 ///   changed.
-pub const TST_ABI_VERSION_MINOR: crate::c_types::c_int = 12;
+/// - `13` — private-data push through the managed-sender and RTSP-mount
+///   pipeline shells: the srt-gated pair `tst_managed_mux_sender_send_data`
+///   / `tst_managed_mux_sender_send_data_to` (behind `TST_HAS_SRT`) and the
+///   rtp-gated pair `tst_rtsp_mount_push_data` / `tst_rtsp_mount_push_data_to`
+///   (behind `TST_HAS_RTP`). Completes the data-stream surface parity with the
+///   video/klv/audio/subtitle push families on both shells. Additive — no
+///   symbol removed, no signature or struct layout changed.
+pub const TST_ABI_VERSION_MINOR: crate::c_types::c_int = 13;
 
 // =========================================================================
 // Runtime version accessors
diff --git a/bindings/c/core/src/rtsp/server/mount.rs b/bindings/c/core/src/rtsp/server/mount.rs
index 7269e7c78..ac1a3cead 100644
--- a/bindings/c/core/src/rtsp/server/mount.rs
+++ b/bindings/c/core/src/rtsp/server/mount.rs
@@ -46,7 +46,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
 
 use tst_core::mpegts::common::Pts90khz;
 use tst_core::mpegts::mux::{
-    AudioStreamHandle, KlvStreamHandle, SubtitleStreamHandle, VideoStreamHandle,
+    AudioStreamHandle, DataStreamHandle, KlvStreamHandle, SubtitleStreamHandle, VideoStreamHandle,
 };
 
 use crate::config::TstMuxConfig;
@@ -54,7 +54,8 @@ use crate::error::{
     TstError, mount_error_to_code, record_mux_error, rtsp_server_error_to_code, set_last_error,
 };
 use crate::handle::{
-    TstAudioStreamHandle, TstKlvStreamHandle, TstSubtitleStreamHandle, TstVideoStreamHandle,
+    TstAudioStreamHandle, TstDataStreamHandle, TstKlvStreamHandle, TstSubtitleStreamHandle,
+    TstVideoStreamHandle,
 };
 use crate::panic::ffi_catch;
 use crate::rtsp::server::types::{TstRtspMountHandle, TstRtspServer};
@@ -544,6 +545,53 @@ pub unsafe extern "C" fn tst_rtsp_mount_push_subtitle(
     })
 }
 
+/// Push one data payload through the mount's single data stream and out the
+/// RTSP broadcast fanout (single-stream shorthand).
+///
+/// Pass-through: `data` lands verbatim as one PES packet on `stream_id`
+/// 0xBD. PTS is written only for `carries_pts = true` streams.
+///
+/// Returns `0` on success, `TST_E_CLOSED` after `tst_rtsp_mount_cancel`,
+/// `TST_E_RTSP_MOUNT` on muxer or mount errors, `TST_E_INVALID_CONFIG` if
+/// `handle` is null.
+///
+/// # Safety
+///
+/// `handle` must be a valid non-freed `*mut tst_rtsp_mount_handle_t`.
+/// `data` must be readable for `len` bytes.
+#[cfg(feature = "rtp")]
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn tst_rtsp_mount_push_data(
+    handle: *mut TstRtspMountHandle,
+    data: *const u8,
+    len: usize,
+    pts_90khz: i64,
+) -> libc::c_int {
+    ffi_catch(TstError::Internal as i32, || {
+        let Some(h) = (unsafe { handle.as_ref() }) else {
+            set_last_error(TstError::InvalidConfig, "null mount handle pointer");
+            return TstError::InvalidConfig as i32;
+        };
+        if h.cancelled.load(Ordering::Acquire) {
+            set_last_error(TstError::Closed, "mount handle has been cancelled");
+            return TstError::Closed as i32;
+        }
+        let slice = match unsafe { crate::ffi_slice::ffi_slice(data, len, "data") } {
+            Ok(s) => s,
+            Err(code) => return code,
+        };
+        let pts = Pts90khz::new(pts_90khz);
+        match h.inner.push_data(slice, pts) {
+            Ok(()) => 0,
+            Err(e) => {
+                let code = mount_error_to_code(&e);
+                set_last_error(code, &format!("push_data failed: {e}"));
+                code as i32
+            }
+        }
+    })
+}
+
 // ---------------------------------------------------------------------------
 // Push — multi-stream (_to) variants
 // ---------------------------------------------------------------------------
@@ -764,6 +812,56 @@ pub unsafe extern "C" fn tst_rtsp_mount_push_subtitle_to(
     })
 }
 
+/// Push one data payload targeting a specific data elementary stream.
+///
+/// On a single-stream mount, prefer `tst_rtsp_mount_push_data`.
+///
+/// # Safety
+///
+/// `handle` must be a valid non-freed `*mut tst_rtsp_mount_handle_t`.
+/// `data` must be readable for `len` bytes.
+#[cfg(feature = "rtp")]
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn tst_rtsp_mount_push_data_to(
+    handle: *mut TstRtspMountHandle,
+    stream_handle: TstDataStreamHandle,
+    data: *const u8,
+    len: usize,
+    pts_90khz: i64,
+) -> libc::c_int {
+    ffi_catch(TstError::Internal as i32, || {
+        let Some(h) = (unsafe { handle.as_ref() }) else {
+            set_last_error(TstError::InvalidConfig, "null mount handle pointer");
+            return TstError::InvalidConfig as i32;
+        };
+        if h.cancelled.load(Ordering::Acquire) {
+            set_last_error(TstError::Closed, "mount handle has been cancelled");
+            return TstError::Closed as i32;
+        }
+        let slice = match unsafe { crate::ffi_slice::ffi_slice(data, len, "data") } {
+            Ok(s) => s,
+            Err(code) => return code,
+        };
+        // Trust-boundary validation — see push_video_to rationale above.
+        let stream = match DataStreamHandle::try_from_raw(stream_handle) {
+            Ok(s) => s,
+            Err(e) => {
+                crate::error::record_mux_error(&e);
+                return unsafe { crate::error::tst_get_last_error() };
+            }
+        };
+        let pts = Pts90khz::new(pts_90khz);
+        match h.inner.push_data_to(stream, slice, pts) {
+            Ok(()) => 0,
+            Err(e) => {
+                let code = mount_error_to_code(&e);
+                set_last_error(code, &format!("push_data_to failed: {e}"));
+                code as i32
+            }
+        }
+    })
+}
+
 // ---------------------------------------------------------------------------
 // Lifecycle helpers — flush, cancel, reset_stats
 // ---------------------------------------------------------------------------
@@ -1041,6 +1139,20 @@ mod tests {
         assert_eq!(rc, crate::error::TstError::InvalidConfig as i32);
     }
 
+    #[test]
+    fn push_data_null_handle_returns_invalid_config() {
+        let rc = unsafe { tst_rtsp_mount_push_data(std::ptr::null_mut(), [0u8; 4].as_ptr(), 4, 0) };
+        assert_eq!(rc, crate::error::TstError::InvalidConfig as i32);
+    }
+
+    #[test]
+    fn push_data_to_null_handle_returns_invalid_config() {
+        let rc = unsafe {
+            tst_rtsp_mount_push_data_to(std::ptr::null_mut(), 0, [0u8; 4].as_ptr(), 4, 0)
+        };
+        assert_eq!(rc, crate::error::TstError::InvalidConfig as i32);
+    }
+
     #[test]
     fn flush_null_handle_returns_invalid_config() {
         let rc = unsafe { tst_rtsp_mount_flush(std::ptr::null_mut()) };
diff --git a/bindings/c/core/src/sender/mux_sender.rs b/bindings/c/core/src/sender/mux_sender.rs
index 9c2576c8b..27cd1125e 100644
--- a/bindings/c/core/src/sender/mux_sender.rs
+++ b/bindings/c/core/src/sender/mux_sender.rs
@@ -1237,6 +1237,92 @@ pub unsafe extern "C" fn tst_managed_mux_sender_send_subtitle_to(
         })
 }
 
+/// Send one data payload through the managed mux sender's single data
+/// stream and out the underlying reconnecting transport.
+///
+/// Pass-through contract identical to `tst_mux_sender_send_data`: `data`
+/// lands verbatim as one PES packet on `stream_id` 0xBD; PTS is written
+/// only for `carries_pts = true` streams.
+///
+/// Single-stream form: see `tst_managed_mux_sender_send_data_to` for the
+/// multi-stream variant.
+///
+/// # Errors
+///
+/// Routed through `tst_get_last_error()`. Same code set as
+/// `tst_mux_sender_send_data` plus `TST_E_NOT_AVAILABLE` (transport
+/// mid-reconnect; transient).
+///
+/// # C ABI
+///
+/// `tst_managed_mux_sender_send_data` — see `bindings/c/include/tstrans.h`.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn tst_managed_mux_sender_send_data(
+    p: *mut TstManagedMuxSender,
+    data: *const u8,
+    len: usize,
+    pts_90khz: i64,
+) -> libc::c_int {
+    let Some(handle) = (unsafe { p.as_ref() }) else {
+        set_last_error(TstError::InvalidConfig, "null sender pointer");
+        return TstError::InvalidConfig as i32;
+    };
+    let slice = match unsafe { crate::ffi_slice::ffi_slice(data, len, "data") } {
+        Ok(s) => s,
+        Err(code) => return code,
+    };
+    let pts = Pts90khz::new(pts_90khz);
+    handle
+        .inner
+        .with_inner_ref(|s| match s.send_data(slice, pts) {
+            Ok(()) => 0,
+            Err(e) => {
+                record_shell_error(&e);
+                unsafe { tst_get_last_error() }
+            }
+        })
+}
+
+/// Send one data payload targeting a specific data elementary stream on a
+/// managed (auto-reconnecting) sender. `stream_handle` is stable across
+/// reconnects. Out-of-range handles surface as `TST_E_INVALID_USAGE`.
+///
+/// On a single-stream sender, prefer `tst_managed_mux_sender_send_data`.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn tst_managed_mux_sender_send_data_to(
+    p: *mut TstManagedMuxSender,
+    stream_handle: TstDataStreamHandle,
+    data: *const u8,
+    len: usize,
+    pts_90khz: i64,
+) -> libc::c_int {
+    let Some(wrapper) = (unsafe { p.as_ref() }) else {
+        set_last_error(TstError::InvalidConfig, "null sender pointer");
+        return TstError::InvalidConfig as i32;
+    };
+    let slice = match unsafe { crate::ffi_slice::ffi_slice(data, len, "data") } {
+        Ok(s) => s,
+        Err(code) => return code,
+    };
+    let stream = match DataStreamHandle::try_from_raw(stream_handle) {
+        Ok(h) => h,
+        Err(e) => {
+            crate::error::record_mux_error(&e);
+            return unsafe { tst_get_last_error() };
+        }
+    };
+    let pts = Pts90khz::new(pts_90khz);
+    wrapper
+        .inner
+        .with_inner_ref(|s| match s.send_data_to(stream, slice, pts) {
+            Ok(()) => 0,
+            Err(e) => {
+                record_shell_error(&e);
+                unsafe { tst_get_last_error() }
+            }
+        })
+}
+
 /// Snapshot stats for a `tst_managed_mux_sender_t` into `*out`.
 ///
 /// Returns 0 on success, `TST_E_INVALID_CONFIG` if either pointer is
diff --git a/bindings/c/include/tstrans.h b/bindings/c/include/tstrans.h
index 11e8d52d2..9a6d7c876 100644
--- a/bindings/c/include/tstrans.h
+++ b/bindings/c/include/tstrans.h
@@ -56,7 +56,7 @@
  * Minor version of the C ABI contract. See [`TST_ABI_VERSION_MAJOR`]
  * for the bump policy.
  *
- * Cbindgen emits this as `#define TST_ABI_VERSION_MINOR 12` in the
+ * Cbindgen emits this as `#define TST_ABI_VERSION_MINOR 13` in the
  * generated header. Runtime accessor: [`tst_get_abi_version_minor`].
  *
  * History (additive bumps only — major stays at 0 pre-1.0):
@@ -123,8 +123,15 @@
  *   alongside video/KLV, mirroring the Rust `push_data` family.
  *   Additive — no symbol removed, no signature or struct layout
  *   changed.
+ * - `13` — private-data push through the managed-sender and RTSP-mount
+ *   pipeline shells: the srt-gated pair `tst_managed_mux_sender_send_data`
+ *   / `tst_managed_mux_sender_send_data_to` (behind `TST_HAS_SRT`) and the
+ *   rtp-gated pair `tst_rtsp_mount_push_data` / `tst_rtsp_mount_push_data_to`
+ *   (behind `TST_HAS_RTP`). Completes the data-stream surface parity with the
+ *   video/klv/audio/subtitle push families on both shells. Additive — no
+ *   symbol removed, no signature or struct layout changed.
  */
-#define TST_ABI_VERSION_MINOR 12
+#define TST_ABI_VERSION_MINOR 13
 
 #define TST_CODEC_KIND_AUDIO 3
 
@@ -1756,6 +1763,12 @@ typedef struct tst_mux_sender_stats_t {
  */
 typedef uint32_t tst_audio_stream_handle_t;
 
+/**
+ * Opaque per-program ordinal for a private/application data elementary
+ * stream. Same packed encoding as [`TstVideoStreamHandle`].
+ */
+typedef uint32_t tst_data_stream_handle_t;
+
 /**
  * Opaque per-program ordinal for a KLV elementary stream. Same packed
  * encoding as [`TstVideoStreamHandle`].
@@ -1840,12 +1853,6 @@ typedef struct tst_sender_stats_t {
  */
 typedef uint32_t tst_program_handle_t;
 
-/**
- * Opaque per-program ordinal for a private/application data elementary
- * stream. Same packed encoding as [`TstVideoStreamHandle`].
- */
-typedef uint32_t tst_data_stream_handle_t;
-
 /**
  * `repr(C)` mirror of `tst_core::publisher::PublisherStats` — the
  * universal cross-publisher stats subset. Returned by
@@ -2283,6 +2290,51 @@ int tst_managed_mux_sender_send_audio_to(struct tst_managed_mux_sender_t *p,
                                          int64_t pts_90khz);
 #endif
 
+#if defined(TST_HAS_SRT)
+/**
+ * Send one data payload through the managed mux sender's single data
+ * stream and out the underlying reconnecting transport.
+ *
+ * Pass-through contract identical to `tst_mux_sender_send_data`: `data`
+ * lands verbatim as one PES packet on `stream_id` 0xBD; PTS is written
+ * only for `carries_pts = true` streams.
+ *
+ * Single-stream form: see `tst_managed_mux_sender_send_data_to` for the
+ * multi-stream variant.
+ *
+ * # Errors
+ *
+ * Routed through `tst_get_last_error()`. Same code set as
+ * `tst_mux_sender_send_data` plus `TST_E_NOT_AVAILABLE` (transport
+ * mid-reconnect; transient).
+ *
+ * # C ABI
+ *
+ * `tst_managed_mux_sender_send_data` — see `bindings/c/include/tstrans.h`.
+ */
+
+int tst_managed_mux_sender_send_data(struct tst_managed_mux_sender_t *p,
+                                     const uint8_t *data,
+                                     size_t len,
+                                     int64_t pts_90khz);
+#endif
+
+#if defined(TST_HAS_SRT)
+/**
+ * Send one data payload targeting a specific data elementary stream on a
+ * managed (auto-reconnecting) sender. `stream_handle` is stable across
+ * reconnects. Out-of-range handles surface as `TST_E_INVALID_USAGE`.
+ *
+ * On a single-stream sender, prefer `tst_managed_mux_sender_send_data`.
+ */
+
+int tst_managed_mux_sender_send_data_to(struct tst_managed_mux_sender_t *p,
+                                        tst_data_stream_handle_t stream_handle,
+                                        const uint8_t *data,
+                                        size_t len,
+                                        int64_t pts_90khz);
+#endif
+
 #if defined(TST_HAS_SRT)
 /**
  * Send one KLV blob through the managed mux sender's single KLV stream
@@ -7554,6 +7606,49 @@ int tst_rtsp_mount_push_audio_to(struct TstRtspMountHandle *handle,
                                  int64_t pts_90khz);
 #endif
 
+#if (defined(TST_HAS_RTP) && defined(TST_HAS_RTP))
+/**
+ * Push one data payload through the mount's single data stream and out the
+ * RTSP broadcast fanout (single-stream shorthand).
+ *
+ * Pass-through: `data` lands verbatim as one PES packet on `stream_id`
+ * 0xBD. PTS is written only for `carries_pts = true` streams.
+ *
+ * Returns `0` on success, `TST_E_CLOSED` after `tst_rtsp_mount_cancel`,
+ * `TST_E_RTSP_MOUNT` on muxer or mount errors, `TST_E_INVALID_CONFIG` if
+ * `handle` is null.
+ *
+ * # Safety
+ *
+ * `handle` must be a valid non-freed `*mut tst_rtsp_mount_handle_t`.
+ * `data` must be readable for `len` bytes.
+ */
+
+int tst_rtsp_mount_push_data(struct TstRtspMountHandle *handle,
+                             const uint8_t *data,
+                             size_t len,
+                             int64_t pts_90khz);
+#endif
+
+#if (defined(TST_HAS_RTP) && defined(TST_HAS_RTP))
+/**
+ * Push one data payload targeting a specific data elementary stream.
+ *
+ * On a single-stream mount, prefer `tst_rtsp_mount_push_data`.
+ *
+ * # Safety
+ *
+ * `handle` must be a valid non-freed `*mut tst_rtsp_mount_handle_t`.
+ * `data` must be readable for `len` bytes.
+ */
+
+int tst_rtsp_mount_push_data_to(struct TstRtspMountHandle *handle,
+                                tst_data_stream_handle_t stream_handle,
+                                const uint8_t *data,
+                                size_t len,
+                                int64_t pts_90khz);
+#endif
+
 #if (defined(TST_HAS_RTP) && defined(TST_HAS_RTP))
 /**
  * Push one raw KLV blob through the mount's single KLV stream (single-stream
diff --git a/bindings/c/tests/receiving/live_pair.rs b/bindings/c/tests/receiving/live_pair.rs
index 863eba80c..c100e95af 100644
--- a/bindings/c/tests/receiving/live_pair.rs
+++ b/bindings/c/tests/receiving/live_pair.rs
@@ -27,9 +27,10 @@ use tstrans::config::{
 };
 use tstrans::error::tst_get_last_error_str;
 use tstrans::sender::mux_sender::{
-    tst_managed_mux_sender_close, tst_managed_mux_sender_open, tst_managed_mux_sender_send_klv_to,
-    tst_managed_mux_sender_send_video_to, tst_mux_sender_close, tst_mux_sender_open,
-    tst_mux_sender_send_data, tst_mux_sender_send_data_to, tst_mux_sender_send_video,
+    tst_managed_mux_sender_close, tst_managed_mux_sender_open, tst_managed_mux_sender_send_data_to,
+    tst_managed_mux_sender_send_klv_to, tst_managed_mux_sender_send_video_to, tst_mux_sender_close,
+    tst_mux_sender_open, tst_mux_sender_send_data, tst_mux_sender_send_data_to,
+    tst_mux_sender_send_video,
 };
 
 fn last_error_msg() -> String {
@@ -383,6 +384,134 @@ fn mux_sender_data_stream_loopback() {
     listener_thread.join().expect("listener thread panicked");
 }
 
+// ---------------------------------------------------------------------------
+// Managed data stream: managed_mux_sender with 1 video + 1 data stream
+// ---------------------------------------------------------------------------
+
+#[test]
+fn managed_mux_sender_data_stream_loopback() {
+    let (port_tx, port_rx) = mpsc::channel::<u16>();
+    let (bytes_tx, bytes_rx) = mpsc::channel::<Vec<u8>>();
+
+    let listener_thread = thread::spawn(move || {
+        let mut listener = ListenerBuilder::new()
+            .recv_timeout(Duration::from_secs(5))
+            .bind("127.0.0.1:0")
+            .expect("bind");
+        let port = listener.local_addr().expect("local_addr").port();
+        port_tx.send(port).expect("send port");
+        let (mut accepted, _peer) = listener.accept().expect("accept");
+
+        // Accumulate until both PIDs (video 0x1011, data 0x1041) have been
+        // seen or the deadline expires.
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        let mut accumulated = Vec::with_capacity(64 * 1024);
+        let mut buf = vec![0u8; 4096];
+        while std::time::Instant::now() < deadline {
+            match accepted.recv(&mut buf) {
+                Ok(n) if n > 0 => accumulated.extend_from_slice(&buf[..n]),
+                Ok(_) => break,
+                Err(_) => break,
+            }
+            let seen = pids_seen(&accumulated);
+            if seen.contains(&0x1011) && seen.contains(&0x1041) {
+                break;
+            }
+        }
+        bytes_tx.send(accumulated).expect("send bytes");
+    });
+
+    let port = port_rx
+        .recv_timeout(Duration::from_secs(5))
+        .expect("listener didn't bind in time");
+    let url = CString::new(format!("srt://127.0.0.1:{port}")).unwrap();
+
+    unsafe {
+        let cfg: *mut TstMuxConfig = tst_mux_config_new();
+        let prog = tst_mux_config_add_program(cfg, 1, 0x1000);
+        // Capture both handles before the config is freed — they are stable
+        // integer indices (not pointers) and survive config teardown.
+        let h_video = tst_mux_config_add_video_stream(cfg, prog, 0x1011, TstVideoCodec::H264);
+        // One data stream (stream_type 0xF0, carries_pts = true).
+        let h_data = tst_mux_config_add_data_stream(cfg, prog, 0x1041, 0xF0, true);
+
+        let policy = tst_reconnect_policy_new();
+        let s = tst_managed_mux_sender_open(url.as_ptr(), cfg, policy);
+        assert!(!s.is_null(), "open failed: {}", last_error_msg());
+        tst_mux_config_free(cfg);
+        tst_reconnect_policy_free(policy);
+
+        let nal: [u8; 9] = [0x00, 0x00, 0x00, 0x01, 0x65, 0xAA, 0xAA, 0xAA, 0xAA];
+        // Arbitrary opaque payload — data streams are a PES pass-through, so
+        // the muxer applies no framing and no payload inspection.
+        let payload: [u8; 12] = [
+            0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07,
+        ];
+
+        // Three rounds; PTS strictly increasing. Uses the explicit-handle
+        // tst_managed_mux_sender_send_data_to entry point.
+        for round in 0u64..3 {
+            let base_pts = (round * 3003) as i64;
+            let rc_video = tst_managed_mux_sender_send_video_to(
+                s,
+                h_video,
+                nal.as_ptr(),
+                nal.len(),
+                base_pts,
+                true,
+            );
+            assert_eq!(
+                rc_video,
+                0,
+                "send_video_to round {round} failed: {}",
+                last_error_msg()
+            );
+
+            let rc_data = tst_managed_mux_sender_send_data_to(
+                s,
+                h_data,
+                payload.as_ptr(),
+                payload.len(),
+                base_pts + 1,
+            );
+            assert_eq!(
+                rc_data,
+                0,
+                "send_data_to round {round} failed: {}",
+                last_error_msg()
+            );
+        }
+
+        let received = bytes_rx
+            .recv_timeout(Duration::from_secs(15))
+            .expect("listener thread didn't deliver bytes within 15s");
+
+        assert!(!received.is_empty(), "received empty buffer");
+        assert_eq!(
+            received[0], 0x47,
+            "expected TS sync byte 0x47 at offset 0; got 0x{:02x}",
+            received[0]
+        );
+
+        let pids = pids_seen(&received);
+        eprintln!("pids_seen = {:?}", pids);
+        assert!(
+            pids.contains(&0x1011),
+            "missing video PID 0x1011; pids_seen = {:?}",
+            pids
+        );
+        assert!(
+            pids.contains(&0x1041),
+            "missing data PID 0x1041; pids_seen = {:?}",
+            pids
+        );
+
+        tst_managed_mux_sender_close(s);
+    }
+
+    listener_thread.join().expect("listener thread panicked");
+}
+
 /// Walk a byte buffer treating every 188-byte aligned run as a TS packet and
 /// collect all 13-bit PIDs encountered.  Byte 0 of each packet must be the
 /// TS sync byte 0x47; if it isn't the walker advances one byte at a time to
diff --git a/bindings/jvm/src/main/java/org/tstrans/rtp/MountHandle.java b/bindings/jvm/src/main/java/org/tstrans/rtp/MountHandle.java
index 786178ec0..82d222755 100644
--- a/bindings/jvm/src/main/java/org/tstrans/rtp/MountHandle.java
+++ b/bindings/jvm/src/main/java/org/tstrans/rtp/MountHandle.java
@@ -6,6 +6,7 @@
 import org.tstrans.NativeLoader;
 import org.tstrans.RtspException;
 import org.tstrans.mpegts.AudioStreamHandle;
+import org.tstrans.mpegts.DataStreamHandle;
 import org.tstrans.mpegts.KlvStreamHandle;
 import org.tstrans.mpegts.SubtitleStreamHandle;
 import org.tstrans.mpegts.VideoStreamHandle;
@@ -61,6 +62,15 @@ public void pushAudio(byte[] frames, long pts) throws RtspException {
     public void pushSubtitle(byte[] payload, long pts) throws RtspException {
         ensureOpen(); nPushSubtitle(handle.get(), pts, payload);
     }
+    /**
+     * Push one private-data payload onto the lone configured data stream
+     * (pass-through: no AU-cell wrap, no framing — {@code data} lands verbatim as
+     * the PES payload). {@code pts} drives PSI/PCR pacing and is written into the
+     * PES header only when the stream was configured with {@code carriesPts = true}.
+     */
+    public void pushData(byte[] data, long pts) throws RtspException {
+        ensureOpen(); nPushData(handle.get(), data, pts);
+    }
 
     // ── Push family — handle-targeted ─────────────────────────────────────
     public void pushVideoTo(VideoStreamHandle h, byte[] nal, long pts, boolean keyFrame)
@@ -78,6 +88,14 @@ public void pushSubtitleTo(SubtitleStreamHandle h, byte[] payload, long pts)
             throws RtspException {
         ensureOpen(); nPushSubtitleTo(handle.get(), h.raw(), pts, payload);
     }
+    /**
+     * Push one private-data payload to a specific configured data stream. Same
+     * pass-through and PTS semantics as {@link #pushData}. An invalid handle
+     * raises {@link RtspException} of kind {@code MOUNT}.
+     */
+    public void pushDataTo(DataStreamHandle h, byte[] data, long pts) throws RtspException {
+        ensureOpen(); nPushDataTo(handle.get(), h.raw(), data, pts);
+    }
 
     // ── Stream-handle accessors (first-of-kind + all-of-kind) ─────────────
     public Optional<VideoStreamHandle> videoHandle() {
@@ -96,6 +114,10 @@ public Optional<SubtitleStreamHandle> subtitleHandle() {
         ensureOpen(); long r = nSubtitleHandle(handle.get());
         return r < 0 ? Optional.empty() : Optional.of(SubtitleStreamHandle.fromRaw(r));
     }
+    public Optional<DataStreamHandle> dataHandle() {
+        ensureOpen(); long r = nDataHandle(handle.get());
+        return r < 0 ? Optional.empty() : Optional.of(DataStreamHandle.fromRaw(r));
+    }
     public List<VideoStreamHandle> videoHandles() {
         ensureOpen();
         List<VideoStreamHandle> out = new ArrayList<>();
@@ -120,6 +142,12 @@ public List<SubtitleStreamHandle> subtitleHandles() {
         for (long r : nSubtitleHandles(handle.get())) out.add(SubtitleStreamHandle.fromRaw(r));
         return out;
     }
+    public List<DataStreamHandle> dataHandles() {
+        ensureOpen();
+        List<DataStreamHandle> out = new ArrayList<>();
+        for (long r : nDataHandles(handle.get())) out.add(DataStreamHandle.fromRaw(r));
+        return out;
+    }
 
     // ── Lifecycle ─────────────────────────────────────────────────────────
     /** Drain buffered TS and broadcast to subscribers. Always safe. */
@@ -150,6 +178,7 @@ private static native void nPushKlv(long handle, byte[] klv, long pts, int metad
     private static native void nPushAudio(long handle, byte[] frames, long pts) throws RtspException;
     private static native void nPushSubtitle(long handle, long pts, byte[] payload)
         throws RtspException;
+    private static native void nPushData(long handle, byte[] data, long pts) throws RtspException;
     private static native void nPushVideoTo(long handle, long streamHandleRaw, byte[] nal, long pts,
         boolean keyFrame) throws RtspException;
     private static native void nPushKlvTo(long handle, long streamHandleRaw, byte[] klv, long pts,
@@ -158,14 +187,18 @@ private static native void nPushAudioTo(long handle, long streamHandleRaw, byte[
         long pts) throws RtspException;
     private static native void nPushSubtitleTo(long handle, long streamHandleRaw, long pts,
         byte[] payload) throws RtspException;
+    private static native void nPushDataTo(long handle, long streamHandleRaw, byte[] data,
+        long pts) throws RtspException;
     private static native long nVideoHandle(long handle);
     private static native long nKlvHandle(long handle);
     private static native long nAudioHandle(long handle);
     private static native long nSubtitleHandle(long handle);
+    private static native long nDataHandle(long handle);
     private static native long[] nVideoHandles(long handle);
     private static native long[] nKlvHandles(long handle);
     private static native long[] nAudioHandles(long handle);
     private static native long[] nSubtitleHandles(long handle);
+    private static native long[] nDataHandles(long handle);
     private static native void nFlush(long handle);
     private static native void nResetStats(long handle);
     private static native void nClose(long handle);
diff --git a/bindings/jvm/src/main/java/org/tstrans/srt/ManagedMuxSender.java b/bindings/jvm/src/main/java/org/tstrans/srt/ManagedMuxSender.java
index 0776e8671..5e0d9ad2e 100644
--- a/bindings/jvm/src/main/java/org/tstrans/srt/ManagedMuxSender.java
+++ b/bindings/jvm/src/main/java/org/tstrans/srt/ManagedMuxSender.java
@@ -5,6 +5,7 @@
 import org.tstrans.NativeLoader;
 import org.tstrans.SrtException;
 import org.tstrans.mpegts.AudioStreamHandle;
+import org.tstrans.mpegts.DataStreamHandle;
 import org.tstrans.mpegts.KlvStreamHandle;
 import org.tstrans.mpegts.MuxerConfig;
 import org.tstrans.mpegts.SubtitleStreamHandle;
@@ -179,6 +180,29 @@ public void pushSubtitle(byte[] payload, long pts) throws MuxException, SrtExcep
         nPushSubtitle(handle.get(), pts, payload);
     }
 
+    /**
+     * Push one private-data payload onto the lone configured data stream.
+     * Pass-through: the muxer applies no AU-cell wrap and no framing (UNLIKE
+     * {@link #pushKlv}) — {@code data} lands verbatim as the PES payload, and
+     * one push produces exactly one PES packet on stream_id {@code 0xBD}
+     * ({@code private_stream_1}). {@code pts} is written into the PES header
+     * only when the stream was configured with {@code carriesPts = true}, but
+     * it ALWAYS drives PSI/PCR pacing.
+     *
+     * @param data raw payload bytes (caller's framing convention; at most
+     *             65527 bytes with PTS, 65532 without)
+     * @param pts  90&nbsp;kHz presentation timestamp
+     * @throws IllegalStateException if the sender is closed
+     * @throws MuxException {@code INPUT_MALFORMED} (payload over the PES
+     *     ceiling), or {@code INVALID_USAGE} (zero data streams, or &gt;1 —
+     *     ambiguous, use {@link #pushDataTo})
+     * @throws SrtException on transport failure
+     */
+    public void pushData(byte[] data, long pts) throws MuxException, SrtException {
+        ensureOpen();
+        nPushData(handle.get(), data, pts);
+    }
+
     // ── Push family — handle-targeted variants ────────────────────────────
 
     /**
@@ -250,6 +274,24 @@ public void pushSubtitleTo(SubtitleStreamHandle h, byte[] payload, long pts)
         nPushSubtitleTo(handle.get(), h.raw(), pts, payload);
     }
 
+    /**
+     * Push one private-data payload to a specific configured data stream. Same
+     * pass-through and PTS semantics as {@link #pushData}.
+     *
+     * @param h    the target stream handle (from {@link #dataHandle()})
+     * @param data raw payload bytes (caller's framing convention; at most
+     *             65527 bytes with PTS, 65532 without)
+     * @param pts  90&nbsp;kHz presentation timestamp
+     * @throws IllegalStateException if the sender is closed
+     * @throws SrtException {@code CONFIG_INVALID} if the handle is invalid
+     * @throws MuxException on muxer failure
+     */
+    public void pushDataTo(DataStreamHandle h, byte[] data, long pts)
+            throws MuxException, SrtException {
+        ensureOpen();
+        nPushDataTo(handle.get(), h.raw(), data, pts);
+    }
+
     // ── Handle getters ────────────────────────────────────────────────────
 
     /**
@@ -296,6 +338,17 @@ public Optional<SubtitleStreamHandle> subtitleHandle() {
         return raw < 0 ? Optional.empty() : Optional.of(SubtitleStreamHandle.fromRaw(raw));
     }
 
+    /**
+     * First configured data stream handle, or {@link Optional#empty()}.
+     *
+     * @return the first data handle, if any
+     */
+    public Optional<DataStreamHandle> dataHandle() {
+        ensureOpen();
+        long raw = nDataHandle(handle.get());
+        return raw < 0 ? Optional.empty() : Optional.of(DataStreamHandle.fromRaw(raw));
+    }
+
     // ── Stats + lifecycle ─────────────────────────────────────────────────
 
     /**
@@ -368,6 +421,8 @@ private static native void nPushAudio(long handle, byte[] frames, long pts)
         throws MuxException, SrtException;
     private static native void nPushSubtitle(long handle, long pts, byte[] payload)
         throws MuxException, SrtException;
+    private static native void nPushData(long handle, byte[] data, long pts)
+        throws MuxException, SrtException;
 
     private static native void nPushVideoTo(long handle, long streamHandleRaw, byte[] nal,
         long pts, boolean keyFrame) throws MuxException, SrtException;
@@ -377,11 +432,14 @@ private static native void nPushAudioTo(long handle, long streamHandleRaw, byte[
         long pts) throws MuxException, SrtException;
     private static native void nPushSubtitleTo(long handle, long streamHandleRaw, long pts,
         byte[] payload) throws MuxException, SrtException;
+    private static native void nPushDataTo(long handle, long streamHandleRaw, byte[] data,
+        long pts) throws MuxException, SrtException;
 
     private static native long nVideoHandle(long handle);
     private static native long nKlvHandle(long handle);
     private static native long nAudioHandle(long handle);
     private static native long nSubtitleHandle(long handle);
+    private static native long nDataHandle(long handle);
 
     private static native TransportStats nStats(long handle);
     private static native long nReconnectAttempts(long handle);
diff --git a/bindings/jvm/src/mpegts/muxer.rs b/bindings/jvm/src/mpegts/muxer.rs
index 1bbf847fb..cd9b227f2 100644
--- a/bindings/jvm/src/mpegts/muxer.rs
+++ b/bindings/jvm/src/mpegts/muxer.rs
@@ -160,6 +160,10 @@ pub(crate) fn build_muxer_config_from_arrays<'local>(
         }
     };
     let desc_lens = read_int_array(env, data_desc_lens).ok_or(())?;
+    if kinds.len() != n || codecs.len() != n || type_codes.len() != n || carries.len() != n {
+        throw_mux(env, "INTERNAL", "stream sibling-array length mismatch");
+        return Err(());
+    }
     if desc_lens.len() != n {
         throw_mux(env, "INTERNAL", "dataDescLens length mismatch");
         return Err(());
@@ -501,7 +505,7 @@ pub extern "system" fn Java_org_tstrans_mpegts_Muxer_nPull<'local>(
         let out_len = match env.get_array_length(&out) {
             Ok(l) => l as usize,
             Err(_) => {
-                throw_mux(env, "INTERNAL", "failed to read byte[] length");
+                let _ = env.throw_new("java/lang/RuntimeException", "failed to read byte[] length");
                 return 0;
             }
         };
@@ -520,7 +524,10 @@ pub extern "system" fn Java_org_tstrans_mpegts_Muxer_nPull<'local>(
         // read by `set_byte_array_region`.
         let i8_view = unsafe { core::slice::from_raw_parts(scratch.as_ptr() as *const i8, n) };
         if env.set_byte_array_region(&out, 0, i8_view).is_err() {
-            throw_mux(env, "INTERNAL", "failed to write byte[] result");
+            let _ = env.throw_new(
+                "java/lang/RuntimeException",
+                "failed to write byte[] result",
+            );
             return 0;
         }
         n as jint
diff --git a/bindings/jvm/src/rtp/mux_sender.rs b/bindings/jvm/src/rtp/mux_sender.rs
index 1076b5799..da0255b84 100644
--- a/bindings/jvm/src/rtp/mux_sender.rs
+++ b/bindings/jvm/src/rtp/mux_sender.rs
@@ -312,12 +312,12 @@ pub extern "system" fn Java_org_tstrans_rtp_MuxSender_nPushVideoTo<'local>(
     key_frame: jboolean,
 ) {
     crate::panic::jni_catch(&mut env, (), |env| {
-        let h = match VideoStreamHandle::try_from_raw(stream_handle_raw as u32) {
-            Ok(h) => h,
-            Err(_) => {
-                throw_rtp(env, "TRANSPORT", "invalid stream handle");
-                return;
-            }
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| VideoStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_rtp(env, "TRANSPORT", "invalid stream handle");
+            return;
         };
         let Some(buf) = read_bytes(env, &nal) else {
             return;
@@ -340,12 +340,12 @@ pub extern "system" fn Java_org_tstrans_rtp_MuxSender_nPushKlvTo<'local>(
     metadata_service_id: jint,
 ) {
     crate::panic::jni_catch(&mut env, (), |env| {
-        let h = match KlvStreamHandle::try_from_raw(stream_handle_raw as u32) {
-            Ok(h) => h,
-            Err(_) => {
-                throw_rtp(env, "TRANSPORT", "invalid stream handle");
-                return;
-            }
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| KlvStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_rtp(env, "TRANSPORT", "invalid stream handle");
+            return;
         };
         let Ok(service_id) = checked_u8(env, i64::from(metadata_service_id), "metadataServiceId")
         else {
@@ -371,12 +371,12 @@ pub extern "system" fn Java_org_tstrans_rtp_MuxSender_nPushAudioTo<'local>(
     pts: jlong,
 ) {
     crate::panic::jni_catch(&mut env, (), |env| {
-        let h = match AudioStreamHandle::try_from_raw(stream_handle_raw as u32) {
-            Ok(h) => h,
-            Err(_) => {
-                throw_rtp(env, "TRANSPORT", "invalid stream handle");
-                return;
-            }
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| AudioStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_rtp(env, "TRANSPORT", "invalid stream handle");
+            return;
         };
         let Some(buf) = read_bytes(env, &frames) else {
             return;
@@ -398,12 +398,12 @@ pub extern "system" fn Java_org_tstrans_rtp_MuxSender_nPushSubtitleTo<'local>(
     payload: JByteArray<'local>,
 ) {
     crate::panic::jni_catch(&mut env, (), |env| {
-        let h = match SubtitleStreamHandle::try_from_raw(stream_handle_raw as u32) {
-            Ok(h) => h,
-            Err(_) => {
-                throw_rtp(env, "TRANSPORT", "invalid stream handle");
-                return;
-            }
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| SubtitleStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_rtp(env, "TRANSPORT", "invalid stream handle");
+            return;
         };
         let Some(buf) = read_bytes(env, &payload) else {
             return;
@@ -417,9 +417,7 @@ pub extern "system" fn Java_org_tstrans_rtp_MuxSender_nPushSubtitleTo<'local>(
 /// `nPushDataTo(handle, streamHandleRaw, data, pts)`. The raw handle is
 /// validated via the strict `u32::try_from` + `DataStreamHandle::try_from_raw`
 /// chain (rejecting negative / out-of-u32 values up front rather than
-/// truncating, mirroring `Muxer::nPushDataTo`). The older `*To` siblings in
-/// this file still decode their handles with the truncating `as u32` cast;
-/// hardening them is deliberately deferred.
+/// truncating, mirroring `Muxer::nPushDataTo`).
 #[unsafe(no_mangle)]
 pub extern "system" fn Java_org_tstrans_rtp_MuxSender_nPushDataTo<'local>(
     mut env: JNIEnv<'local>,
diff --git a/bindings/jvm/src/rtp/server.rs b/bindings/jvm/src/rtp/server.rs
index 0e2f61513..0cffe901c 100644
--- a/bindings/jvm/src/rtp/server.rs
+++ b/bindings/jvm/src/rtp/server.rs
@@ -16,7 +16,7 @@ use secrecy::SecretString;
 
 use tst_core::mpegts::common::Pts90khz;
 use tst_core::mpegts::mux::{
-    AudioStreamHandle, KlvStreamHandle, SubtitleStreamHandle, VideoStreamHandle,
+    AudioStreamHandle, DataStreamHandle, KlvStreamHandle, SubtitleStreamHandle, VideoStreamHandle,
 };
 use tst_rtp::RtspServer as RustRtspServer;
 use tst_rtp::RtspServerCancelHandle as RustServerCancel;
@@ -749,6 +749,30 @@ pub extern "system" fn Java_org_tstrans_rtp_MountHandle_nPushSubtitle<'local>(
     })
 }
 
+/// `MountHandle.nPushData(handle, data, pts)`.
+#[unsafe(no_mangle)]
+pub extern "system" fn Java_org_tstrans_rtp_MountHandle_nPushData<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    data: JByteArray<'local>,
+    pts: jlong,
+) {
+    crate::panic::jni_catch(&mut env, (), |env| {
+        let Some(buf) = read_bytes(env, &data) else {
+            return;
+        };
+        let Some(res) = with_mount(env, handle, |inner| {
+            inner.push_data(&buf, Pts90khz::new(pts))
+        }) else {
+            return;
+        };
+        if let Err(e) = res {
+            mount_error_to_jvm(env, &e);
+        }
+    })
+}
+
 // ── MountHandle: push family — handle-targeted variants ─────────────────────
 
 /// `MountHandle.nPushVideoTo(handle, streamHandleRaw, nal, pts, keyFrame)`.
@@ -763,14 +787,14 @@ pub extern "system" fn Java_org_tstrans_rtp_MountHandle_nPushVideoTo<'local>(
     key_frame: jboolean,
 ) {
     crate::panic::jni_catch(&mut env, (), |env| {
-        let h = match VideoStreamHandle::try_from_raw(stream_handle_raw as u32) {
-            Ok(h) => h,
-            Err(_) => {
-                // MountHandle has no transport concept — a forged handle is a MOUNT error
-                // (DIFFERS from MuxSender, which maps forged handles to RtpException(TRANSPORT)).
-                throw_rtsp(env, "MOUNT", "invalid stream handle");
-                return;
-            }
+        // MountHandle has no transport concept — a forged handle is a MOUNT error
+        // (DIFFERS from MuxSender, which maps forged handles to RtpException(TRANSPORT)).
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| VideoStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_rtsp(env, "MOUNT", "invalid stream handle");
+            return;
         };
         let Some(buf) = read_bytes(env, &nal) else {
             return;
@@ -798,12 +822,12 @@ pub extern "system" fn Java_org_tstrans_rtp_MountHandle_nPushKlvTo<'local>(
     metadata_service_id: jint,
 ) {
     crate::panic::jni_catch(&mut env, (), |env| {
-        let h = match KlvStreamHandle::try_from_raw(stream_handle_raw as u32) {
-            Ok(h) => h,
-            Err(_) => {
-                throw_rtsp(env, "MOUNT", "invalid stream handle");
-                return;
-            }
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| KlvStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_rtsp(env, "MOUNT", "invalid stream handle");
+            return;
         };
         let Ok(service_id) = checked_u8(env, i64::from(metadata_service_id), "metadataServiceId")
         else {
@@ -834,12 +858,12 @@ pub extern "system" fn Java_org_tstrans_rtp_MountHandle_nPushAudioTo<'local>(
     pts: jlong,
 ) {
     crate::panic::jni_catch(&mut env, (), |env| {
-        let h = match AudioStreamHandle::try_from_raw(stream_handle_raw as u32) {
-            Ok(h) => h,
-            Err(_) => {
-                throw_rtsp(env, "MOUNT", "invalid stream handle");
-                return;
-            }
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| AudioStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_rtsp(env, "MOUNT", "invalid stream handle");
+            return;
         };
         let Some(buf) = read_bytes(env, &frames) else {
             return;
@@ -866,12 +890,12 @@ pub extern "system" fn Java_org_tstrans_rtp_MountHandle_nPushSubtitleTo<'local>(
     payload: JByteArray<'local>,
 ) {
     crate::panic::jni_catch(&mut env, (), |env| {
-        let h = match SubtitleStreamHandle::try_from_raw(stream_handle_raw as u32) {
-            Ok(h) => h,
-            Err(_) => {
-                throw_rtsp(env, "MOUNT", "invalid stream handle");
-                return;
-            }
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| SubtitleStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_rtsp(env, "MOUNT", "invalid stream handle");
+            return;
         };
         let Some(buf) = read_bytes(env, &payload) else {
             return;
@@ -887,6 +911,42 @@ pub extern "system" fn Java_org_tstrans_rtp_MountHandle_nPushSubtitleTo<'local>(
     })
 }
 
+/// `MountHandle.nPushDataTo(handle, streamHandleRaw, data, pts)`. The raw handle
+/// is validated via the strict `u32::try_from` + `DataStreamHandle::try_from_raw`
+/// chain (rejecting negative / out-of-u32 values up front rather than
+/// truncating). A forged handle is a `MOUNT` error (DIFFERS from MuxSender, which
+/// maps forged handles to RtpException(TRANSPORT)).
+#[unsafe(no_mangle)]
+pub extern "system" fn Java_org_tstrans_rtp_MountHandle_nPushDataTo<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    stream_handle_raw: jlong,
+    data: JByteArray<'local>,
+    pts: jlong,
+) {
+    crate::panic::jni_catch(&mut env, (), |env| {
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| DataStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_rtsp(env, "MOUNT", "invalid stream handle");
+            return;
+        };
+        let Some(buf) = read_bytes(env, &data) else {
+            return;
+        };
+        let Some(res) = with_mount(env, handle, |inner| {
+            inner.push_data_to(h, &buf, Pts90khz::new(pts))
+        }) else {
+            return;
+        };
+        if let Err(e) = res {
+            mount_error_to_jvm(env, &e);
+        }
+    })
+}
+
 // ── MountHandle: stream-handle accessors (first-of-kind; -1 = none) ──────────
 
 /// `MountHandle.nVideoHandle(handle)` — first configured video stream handle, or `-1`.
@@ -969,6 +1029,26 @@ pub extern "system" fn Java_org_tstrans_rtp_MountHandle_nSubtitleHandle(
     })
 }
 
+/// `MountHandle.nDataHandle(handle)` — first configured data stream handle, or `-1`.
+#[unsafe(no_mangle)]
+pub extern "system" fn Java_org_tstrans_rtp_MountHandle_nDataHandle(
+    mut env: JNIEnv<'_>,
+    _class: JClass<'_>,
+    handle: jlong,
+) -> jlong {
+    crate::panic::jni_catch(&mut env, 0, |env| {
+        with_mount(env, handle, |inner| {
+            inner
+                .data_handles()
+                .into_iter()
+                .next()
+                .map(|h| i64::from(h.raw()))
+                .unwrap_or(-1)
+        })
+        .unwrap_or(-1)
+    })
+}
+
 // ── MountHandle: stream-handle accessors (all-of-kind; long[]) ──────────────
 
 /// `MountHandle.nVideoHandles(handle)` → `long[]` of all video stream handles.
@@ -1075,6 +1155,32 @@ pub extern "system" fn Java_org_tstrans_rtp_MountHandle_nSubtitleHandles<'local>
     })
 }
 
+/// `MountHandle.nDataHandles(handle)` → `long[]` of all data stream handles.
+#[unsafe(no_mangle)]
+pub extern "system" fn Java_org_tstrans_rtp_MountHandle_nDataHandles<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+) -> JLongArray<'local> {
+    crate::panic::jni_catch(&mut env, JObject::null().into(), |env| {
+        let Some(raws) = with_mount(env, handle, |inner| {
+            inner
+                .data_handles()
+                .into_iter()
+                .map(|h| i64::from(h.raw()))
+                .collect::<Vec<i64>>()
+        }) else {
+            return JObject::null().into();
+        };
+        let arr = match env.new_long_array(raws.len() as i32) {
+            Ok(a) => a,
+            Err(_) => return JObject::null().into(),
+        };
+        let _ = env.set_long_array_region(&arr, 0, &raws);
+        arr
+    })
+}
+
 // ── MountHandle: lifecycle ──────────────────────────────────────────────────
 
 /// `MountHandle.nFlush(handle)` — drain + broadcast buffered TS.
diff --git a/bindings/jvm/src/srt/managed_convenience.rs b/bindings/jvm/src/srt/managed_convenience.rs
index 56c1fd45b..9b0491c17 100644
--- a/bindings/jvm/src/srt/managed_convenience.rs
+++ b/bindings/jvm/src/srt/managed_convenience.rs
@@ -48,7 +48,7 @@ use jni::sys::{jboolean, jint, jlong, jobject};
 
 use tst_core::mpegts::common::Pts90khz;
 use tst_core::mpegts::mux::{
-    AudioStreamHandle, KlvStreamHandle, SubtitleStreamHandle, VideoStreamHandle,
+    AudioStreamHandle, DataStreamHandle, KlvStreamHandle, SubtitleStreamHandle, VideoStreamHandle,
 };
 use tst_core::transport::TransportError;
 use tst_pipeline::{
@@ -413,6 +413,25 @@ pub extern "system" fn Java_org_tstrans_srt_ManagedMuxSender_nPushSubtitle<'loca
     })
 }
 
+/// `nPushData(handle, data, pts)`.
+#[unsafe(no_mangle)]
+pub extern "system" fn Java_org_tstrans_srt_ManagedMuxSender_nPushData<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    data: JByteArray<'local>,
+    pts: jlong,
+) {
+    crate::panic::jni_catch(&mut env, (), |env| {
+        let Some(buf) = read_bytes(env, &data) else {
+            return;
+        };
+        with_mux_push(env, handle, |inner| {
+            inner.send_data(&buf, Pts90khz::new(pts))
+        });
+    })
+}
+
 // ── Push family — handle-targeted variants ─────────────────────────────────
 
 /// `nPushVideoTo(handle, streamHandleRaw, nal, pts, keyFrame)`.
@@ -427,12 +446,12 @@ pub extern "system" fn Java_org_tstrans_srt_ManagedMuxSender_nPushVideoTo<'local
     key_frame: jboolean,
 ) {
     crate::panic::jni_catch(&mut env, (), |env| {
-        let h = match VideoStreamHandle::try_from_raw(stream_handle_raw as u32) {
-            Ok(h) => h,
-            Err(_) => {
-                throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
-                return;
-            }
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| VideoStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
+            return;
         };
         let Some(buf) = read_bytes(env, &nal) else {
             return;
@@ -455,12 +474,12 @@ pub extern "system" fn Java_org_tstrans_srt_ManagedMuxSender_nPushKlvTo<'local>(
     metadata_service_id: jint,
 ) {
     crate::panic::jni_catch(&mut env, (), |env| {
-        let h = match KlvStreamHandle::try_from_raw(stream_handle_raw as u32) {
-            Ok(h) => h,
-            Err(_) => {
-                throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
-                return;
-            }
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| KlvStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
+            return;
         };
         let Ok(service_id) = checked_u8(env, i64::from(metadata_service_id), "metadataServiceId")
         else {
@@ -486,12 +505,12 @@ pub extern "system" fn Java_org_tstrans_srt_ManagedMuxSender_nPushAudioTo<'local
     pts: jlong,
 ) {
     crate::panic::jni_catch(&mut env, (), |env| {
-        let h = match AudioStreamHandle::try_from_raw(stream_handle_raw as u32) {
-            Ok(h) => h,
-            Err(_) => {
-                throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
-                return;
-            }
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| AudioStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
+            return;
         };
         let Some(buf) = read_bytes(env, &frames) else {
             return;
@@ -513,12 +532,12 @@ pub extern "system" fn Java_org_tstrans_srt_ManagedMuxSender_nPushSubtitleTo<'lo
     payload: JByteArray<'local>,
 ) {
     crate::panic::jni_catch(&mut env, (), |env| {
-        let h = match SubtitleStreamHandle::try_from_raw(stream_handle_raw as u32) {
-            Ok(h) => h,
-            Err(_) => {
-                throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
-                return;
-            }
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| SubtitleStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
+            return;
         };
         let Some(buf) = read_bytes(env, &payload) else {
             return;
@@ -529,6 +548,36 @@ pub extern "system" fn Java_org_tstrans_srt_ManagedMuxSender_nPushSubtitleTo<'lo
     })
 }
 
+/// `nPushDataTo(handle, streamHandleRaw, data, pts)`. The raw handle is
+/// validated via the strict `u32::try_from` + `DataStreamHandle::try_from_raw`
+/// chain (rejecting negative / out-of-u32 values up front rather than
+/// truncating, mirroring `MuxSender::nPushDataTo`).
+#[unsafe(no_mangle)]
+pub extern "system" fn Java_org_tstrans_srt_ManagedMuxSender_nPushDataTo<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    stream_handle_raw: jlong,
+    data: JByteArray<'local>,
+    pts: jlong,
+) {
+    crate::panic::jni_catch(&mut env, (), |env| {
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| DataStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
+            return;
+        };
+        let Some(buf) = read_bytes(env, &data) else {
+            return;
+        };
+        with_mux_push(env, handle, |inner| {
+            inner.send_data_to(h, &buf, Pts90khz::new(pts))
+        });
+    })
+}
+
 // ── Handle getters ─────────────────────────────────────────────────────────
 
 /// `nVideoHandle(handle)` — first configured video stream handle, or `-1`.
@@ -587,6 +636,20 @@ pub extern "system" fn Java_org_tstrans_srt_ManagedMuxSender_nSubtitleHandle(
     })
 }
 
+/// `nDataHandle(handle)` — first configured data stream handle, or `-1`.
+#[unsafe(no_mangle)]
+pub extern "system" fn Java_org_tstrans_srt_ManagedMuxSender_nDataHandle(
+    mut env: JNIEnv<'_>,
+    _class: JClass<'_>,
+    handle: jlong,
+) -> jlong {
+    crate::panic::jni_catch(&mut env, 0, |env| {
+        mux_first_handle(env, handle, |inner| {
+            inner.data_handles().into_iter().next().map(|h| h.raw())
+        })
+    })
+}
+
 // ── Stats + lifecycle ──────────────────────────────────────────────────────
 
 /// `nStats(handle)` — `TransportStats` projecting the SRT socket counters + the
diff --git a/bindings/jvm/src/srt/mux_sender.rs b/bindings/jvm/src/srt/mux_sender.rs
index bb6464a43..f1d1d5dac 100644
--- a/bindings/jvm/src/srt/mux_sender.rs
+++ b/bindings/jvm/src/srt/mux_sender.rs
@@ -342,12 +342,12 @@ pub extern "system" fn Java_org_tstrans_srt_MuxSender_nPushVideoTo<'local>(
     key_frame: jboolean,
 ) {
     crate::panic::jni_catch(&mut env, (), |env| {
-        let h = match VideoStreamHandle::try_from_raw(stream_handle_raw as u32) {
-            Ok(h) => h,
-            Err(_) => {
-                throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
-                return;
-            }
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| VideoStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
+            return;
         };
         let Some(buf) = read_bytes(env, &nal) else {
             return;
@@ -370,12 +370,12 @@ pub extern "system" fn Java_org_tstrans_srt_MuxSender_nPushKlvTo<'local>(
     metadata_service_id: jint,
 ) {
     crate::panic::jni_catch(&mut env, (), |env| {
-        let h = match KlvStreamHandle::try_from_raw(stream_handle_raw as u32) {
-            Ok(h) => h,
-            Err(_) => {
-                throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
-                return;
-            }
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| KlvStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
+            return;
         };
         let Ok(service_id) = checked_u8(env, i64::from(metadata_service_id), "metadataServiceId")
         else {
@@ -401,12 +401,12 @@ pub extern "system" fn Java_org_tstrans_srt_MuxSender_nPushAudioTo<'local>(
     pts: jlong,
 ) {
     crate::panic::jni_catch(&mut env, (), |env| {
-        let h = match AudioStreamHandle::try_from_raw(stream_handle_raw as u32) {
-            Ok(h) => h,
-            Err(_) => {
-                throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
-                return;
-            }
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| AudioStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
+            return;
         };
         let Some(buf) = read_bytes(env, &frames) else {
             return;
@@ -428,12 +428,12 @@ pub extern "system" fn Java_org_tstrans_srt_MuxSender_nPushSubtitleTo<'local>(
     payload: JByteArray<'local>,
 ) {
     crate::panic::jni_catch(&mut env, (), |env| {
-        let h = match SubtitleStreamHandle::try_from_raw(stream_handle_raw as u32) {
-            Ok(h) => h,
-            Err(_) => {
-                throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
-                return;
-            }
+        let Some(h) = u32::try_from(stream_handle_raw)
+            .ok()
+            .and_then(|r| SubtitleStreamHandle::try_from_raw(r).ok())
+        else {
+            throw_srt(env, "CONFIG_INVALID", "invalid stream handle");
+            return;
         };
         let Some(buf) = read_bytes(env, &payload) else {
             return;
@@ -447,9 +447,7 @@ pub extern "system" fn Java_org_tstrans_srt_MuxSender_nPushSubtitleTo<'local>(
 /// `nPushDataTo(handle, streamHandleRaw, data, pts)`. The raw handle is
 /// validated via the strict `u32::try_from` + `DataStreamHandle::try_from_raw`
 /// chain (rejecting negative / out-of-u32 values up front rather than
-/// truncating, mirroring `Muxer::nPushDataTo`). The older `*To` siblings in
-/// this file still decode their handles with the truncating `as u32` cast;
-/// hardening them is deliberately deferred.
+/// truncating, mirroring `Muxer::nPushDataTo`).
 #[unsafe(no_mangle)]
 pub extern "system" fn Java_org_tstrans_srt_MuxSender_nPushDataTo<'local>(
     mut env: JNIEnv<'local>,
diff --git a/bindings/jvm/src/test/java/org/tstrans/rtp/RtspServerTest.java b/bindings/jvm/src/test/java/org/tstrans/rtp/RtspServerTest.java
index 98a262516..7ca135856 100644
--- a/bindings/jvm/src/test/java/org/tstrans/rtp/RtspServerTest.java
+++ b/bindings/jvm/src/test/java/org/tstrans/rtp/RtspServerTest.java
@@ -5,8 +5,10 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 import org.tstrans.RtspException;
+import org.tstrans.mpegts.DataStreamHandle;
 import org.tstrans.mpegts.MuxerConfig;
 import org.tstrans.mpegts.VideoCodec;
+import org.tstrans.mpegts.VideoStreamHandle;
 
 class RtspServerTest {
 
@@ -124,6 +126,33 @@ void addUnicastMountPushAndStats() throws Exception {
         }
     }
 
+    @Test @Timeout(15)
+    void addUnicastMountDataPushAndStats() throws Exception {
+        try (RtspServer s = RtspServer.start(RtspServerConfig.of("127.0.0.1:0"))) {
+            MuxerConfig cfg = MuxerConfig.builder()
+                .programNumber(1).pmtPid(0x1000)
+                .addVideo(0x1011, VideoCodec.H264)
+                .addData(0x0100, 0xF0, true).build();
+            try (MountHandle m = s.addUnicastMount("/data", cfg)) {
+                // The config declares one data stream → both accessors surface it.
+                assertTrue(m.dataHandle().isPresent());
+                assertEquals(1, m.dataHandles().size());
+                MountStats before = m.stats();
+                // pushData (lone-data-stream shorthand) advances the flow counters.
+                m.pushData(new byte[] {(byte) 0xD0, 'D', 'A', 'T', 'A'}, 0L);
+                m.flush();
+                assertTrue(m.stats().bytesPushed() > before.bytesPushed());
+                // pushDataTo with the configured handle also succeeds.
+                m.pushDataTo(m.dataHandle().get(), new byte[] {1, 2, 3}, 90_000L);
+                // Strict handle decode: a forged/negative handle is rejected with
+                // RtspException(MOUNT) in the JNI shim before reaching the mount.
+                RtspException forged = assertThrows(RtspException.class,
+                    () -> m.pushDataTo(DataStreamHandle.fromRaw(-1L), new byte[] {1}, 0L));
+                assertEquals(RtspException.Kind.MOUNT, forged.kind());
+            }
+        }
+    }
+
     @Test @Timeout(15)
     void addMulticastMountKind() throws Exception {
         try (RtspServer s = RtspServer.start(RtspServerConfig.of("127.0.0.1:0"))) {
@@ -159,6 +188,35 @@ void duplicateMountThrowsMount() throws Exception {
         }
     }
 
+    /**
+     * Strict handle decode in {@code pushVideoTo}: a negative jlong or a value
+     * exceeding {@code u32::MAX} must be rejected with
+     * {@code RtspException(MOUNT)} without truncating into a plausible handle.
+     * Regression for the pre-3.4 {@code as u32} cast that would silently wrap
+     * {@code -1L} to {@code 0xFFFF_FFFF} and {@code 0x1_0000_0000L} to {@code 0}.
+     */
+    @Test @Timeout(15)
+    void pushVideoToRejectsForgedStreamHandle() throws Exception {
+        try (RtspServer s = RtspServer.start(RtspServerConfig.of("127.0.0.1:0"))) {
+            MuxerConfig cfg = MuxerConfig.builder()
+                .programNumber(1).pmtPid(0x1000)
+                .addVideo(0x1011, VideoCodec.H264).build();
+            try (MountHandle m = s.addUnicastMount("/v", cfg)) {
+                byte[] nal = idr();
+                // Negative jlong: rejected by the u32::try_from leg before try_from_raw.
+                RtspException neg = assertThrows(RtspException.class,
+                    () -> m.pushVideoTo(VideoStreamHandle.fromRaw(-1L), nal, 0L, true));
+                assertEquals(RtspException.Kind.MOUNT, neg.kind(),
+                    "negative VideoStreamHandle must raise RtspException(MOUNT)");
+                // Out-of-u32 jlong: also rejected by the u32::try_from leg.
+                RtspException over = assertThrows(RtspException.class,
+                    () -> m.pushVideoTo(VideoStreamHandle.fromRaw(0x1_0000_0000L), nal, 0L, true));
+                assertEquals(RtspException.Kind.MOUNT, over.kind(),
+                    "out-of-u32 VideoStreamHandle must raise RtspException(MOUNT)");
+            }
+        }
+    }
+
     private static byte[] idr() {
         byte[] b = new byte[20];
         b[0]=0; b[1]=0; b[2]=0; b[3]=1; b[4]=0x65;
diff --git a/bindings/jvm/src/test/java/org/tstrans/srt/SrtManagedLiveTest.java b/bindings/jvm/src/test/java/org/tstrans/srt/SrtManagedLiveTest.java
index 699659162..f5ff6f60e 100644
--- a/bindings/jvm/src/test/java/org/tstrans/srt/SrtManagedLiveTest.java
+++ b/bindings/jvm/src/test/java/org/tstrans/srt/SrtManagedLiveTest.java
@@ -17,6 +17,7 @@
 import org.tstrans.SrtException;
 import org.tstrans.codec.NalUnit;
 import org.tstrans.codec.VideoUnit;
+import org.tstrans.mpegts.DataStreamHandle;
 import org.tstrans.mpegts.DemuxEvent;
 import org.tstrans.mpegts.Demuxer;
 import org.tstrans.mpegts.Muxer;
@@ -108,6 +109,7 @@ void managedMuxSenderToDemuxReceiverByteFaithful() throws Exception {
 
         CompletableFuture<Integer> portFuture = new CompletableFuture<>();
         CompletableFuture<String> shaFuture = new CompletableFuture<>();
+        CompletableFuture<DemuxEvent.UnknownSample> dataFuture = new CompletableFuture<>();
         // Counted down by main once it has STOPPED pushing and closed the sender;
         // the receiver holds its teardown on it so a final in-flight mini-batch can
         // never land on a closing peer (which would surface spurious sender-side
@@ -130,16 +132,20 @@ void managedMuxSenderToDemuxReceiverByteFaithful() throws Exception {
                 rx = sock.intoDemuxReceiver();
 
                 String sha = null;
+                DemuxEvent.UnknownSample data = null;
                 try {
                     for (DemuxEvent e : rx) {
-                        if (e instanceof DemuxEvent.Video v && !v.payload().isEmpty()) {
+                        if (sha == null && e instanceof DemuxEvent.Video v
+                                && !v.payload().isEmpty()) {
                             sha = sha256Units(v.payload());
-                            break;
+                        } else if (data == null && e instanceof DemuxEvent.UnknownSample u) {
+                            data = u;
                         }
+                        if (sha != null && data != null) break;
                     }
                 } catch (RuntimeException re) {
                     if (!isCleanEndOfStream(re)) throw re;
-                    // else: fall through; sha may still be null → fail below
+                    // else: fall through; sha/data may still be null → fail below
                 }
 
                 if (sha == null) {
@@ -148,9 +154,16 @@ void managedMuxSenderToDemuxReceiverByteFaithful() throws Exception {
                 } else {
                     shaFuture.complete(sha);
                 }
+                if (data == null) {
+                    dataFuture.completeExceptionally(new AssertionError(
+                        "no UnknownSample (private-data) event arrived before end-of-stream"));
+                } else {
+                    dataFuture.complete(data);
+                }
             } catch (Exception ex) {
                 portFuture.completeExceptionally(ex);
                 shaFuture.completeExceptionally(ex);
+                dataFuture.completeExceptionally(ex);
             } finally {
                 // Hold teardown until main has stopped pushing and closed its sender
                 // (bounded, so a failed main can never park this daemon forever).
@@ -194,12 +207,32 @@ void managedMuxSenderToDemuxReceiverByteFaithful() throws Exception {
             // on the failure path. (Daemon-side runaway caps may use 400; main-side
             // loops must fit the budget.)
             long pts = 0;
-            for (int round = 0; round < 200 && !shaFuture.isDone(); round++) {
+            for (int round = 0;
+                    round < 200 && (!shaFuture.isDone() || !dataFuture.isDone());
+                    round++) {
                 for (int i = 0; i < 6; i++, pts += 3000L) {
                     s.pushVideo(syntheticH264Idr(), pts, true);
                 }
+                // One distinctive private-data record per round (lone-data-stream
+                // pushData shorthand) so it deterministically flows under continuous
+                // streaming — identical bytes each push; the receiver captures the
+                // first. The following round's video pushes flush its PES.
+                s.pushData(DATA_PAYLOAD, pts);
                 Thread.sleep(50);
             }
+            // Convenience accessor: the config declares one data stream, so the
+            // handle must surface (exercises the native + sentinel path on a live
+            // managed sender).
+            assertTrue(s.dataHandle().isPresent(),
+                "config declares one data stream → dataHandle() must surface it");
+            // Strict handle decode: a forged/negative handle is rejected with
+            // SrtException(CONFIG_INVALID) in the JNI shim before reaching the
+            // muxer (DIFFERS from Muxer.pushDataTo, which maps it to MuxException).
+            // Thrown before any push, so the live sender's state is untouched.
+            SrtException forged = assertThrows(SrtException.class,
+                () -> s.pushDataTo(DataStreamHandle.fromRaw(-1L), DATA_PAYLOAD, 0L));
+            assertEquals(SrtException.Kind.CONFIG_INVALID, forged.kind(),
+                "forged DataStreamHandle must raise SrtException(CONFIG_INVALID)");
             long attempts = s.reconnectAttempts();
             TransportStats st = s.stats();
             assertEquals(0L, attempts, "no reconnect should have occurred on the happy path");
@@ -221,6 +254,22 @@ void managedMuxSenderToDemuxReceiverByteFaithful() throws Exception {
                 + "payload SHA as the offline Muxer→Demuxer path (cross-binding parity, "
                 + "self-validating)");
 
+        // The pushData record must round-trip byte-faithfully as an UnknownSample
+        // on the configured 0xF0 data stream.
+        DemuxEvent.UnknownSample dataSample;
+        try {
+            dataSample = dataFuture.get(TIMEOUT_SEC, TimeUnit.SECONDS);
+        } catch (Exception e) {
+            throw new AssertionError("receiver thread did not surface the private-data sample", e);
+        }
+        assertEquals(0xF0, dataSample.streamType(),
+            "private-data sample must carry the configured raw stream_type");
+        ByteBuffer dataView = dataSample.payload().duplicate();
+        byte[] dataBytes = new byte[dataView.remaining()];
+        dataView.get(dataBytes);
+        assertArrayEquals(DATA_PAYLOAD, dataBytes,
+            "private-data payload must arrive verbatim (pass-through, no framing)");
+
         receiverThread.join(TimeUnit.SECONDS.toMillis(TIMEOUT_SEC));
     }
 
@@ -542,11 +591,16 @@ private static byte[] syntheticH264Idr() {
         return buf;
     }
 
-    /** The single-program H.264 config shared by the live and offline paths. */
+    /** Distinctive private-data record pushed alongside the video stream (test 1). */
+    private static final byte[] DATA_PAYLOAD =
+        {(byte) 0xD0, 'D', 'A', 'T', 'A', (byte) 0xBE, (byte) 0xEF, 0x01};
+
+    /** The single-program H.264 + private-data config shared by the live and offline paths. */
     private static MuxerConfig roundtripConfig() {
         return MuxerConfig.builder()
             .programNumber(1).pmtPid(0x1000)
             .addVideo(0x1011, VideoCodec.H264)
+            .addData(0x0100, 0xF0, true)
             .build();
     }
 
diff --git a/bindings/python/python/tstrans/rtp.pyi b/bindings/python/python/tstrans/rtp.pyi
index 8b0a3736b..9e2cba704 100644
--- a/bindings/python/python/tstrans/rtp.pyi
+++ b/bindings/python/python/tstrans/rtp.pyi
@@ -409,6 +409,7 @@ class MountHandle:
     ) -> None: ...
     def push_audio(self, frames: BytesLike, *, pts: Pts90khz) -> None: ...
     def push_subtitle(self, payload: BytesLike, *, pts: Pts90khz) -> None: ...
+    def push_data(self, data: BytesLike, *, pts: Pts90khz) -> None: ...
 
     # Push surface — multi-stream (explicit handle).
     def push_video_to(
@@ -441,16 +442,25 @@ class MountHandle:
         *,
         pts: Pts90khz,
     ) -> None: ...
+    def push_data_to(
+        self,
+        handle: DataStreamHandle,
+        data: BytesLike,
+        *,
+        pts: Pts90khz,
+    ) -> None: ...
 
     # Stream-handle accessors (first-of-kind + all-of-kind).
     def video_handle(self) -> Optional[VideoStreamHandle]: ...
     def klv_handle(self) -> Optional[KlvStreamHandle]: ...
     def audio_handle(self) -> Optional[AudioStreamHandle]: ...
     def subtitle_handle(self) -> Optional[SubtitleStreamHandle]: ...
+    def data_handle(self) -> Optional[DataStreamHandle]: ...
     def video_handles(self) -> List[VideoStreamHandle]: ...
     def klv_handles(self) -> List[KlvStreamHandle]: ...
     def audio_handles(self) -> List[AudioStreamHandle]: ...
     def subtitle_handles(self) -> List[SubtitleStreamHandle]: ...
+    def data_handles(self) -> List[DataStreamHandle]: ...
 
     # Lifecycle.
     def flush(self) -> None: ...
diff --git a/bindings/python/python/tstrans/srt.pyi b/bindings/python/python/tstrans/srt.pyi
index 73593de50..c5f53711b 100644
--- a/bindings/python/python/tstrans/srt.pyi
+++ b/bindings/python/python/tstrans/srt.pyi
@@ -662,6 +662,7 @@ class ManagedMuxSender:
     ) -> None: ...
     def push_audio(self, adts: BytesLike, *, pts: Pts90khz) -> None: ...
     def push_subtitle(self, payload: BytesLike, *, pts: Pts90khz) -> None: ...
+    def push_data(self, data: BytesLike, *, pts: Pts90khz) -> None: ...
 
     # Push surface — multi-stream variants.
     def push_video_to(
@@ -694,12 +695,20 @@ class ManagedMuxSender:
         *,
         pts: Pts90khz,
     ) -> None: ...
+    def push_data_to(
+        self,
+        handle: DataStreamHandle,
+        data: BytesLike,
+        *,
+        pts: Pts90khz,
+    ) -> None: ...
 
     # Stream handle accessors.
     def video_handle(self) -> Optional[VideoStreamHandle]: ...
     def klv_handle(self) -> Optional[KlvStreamHandle]: ...
     def audio_handle(self) -> Optional[AudioStreamHandle]: ...
     def subtitle_handle(self) -> Optional[SubtitleStreamHandle]: ...
+    def data_handle(self) -> Optional[DataStreamHandle]: ...
 
     def stats(self) -> Tuple[SocketStats, MuxerStats]: ...
     def reconnect_attempts(self) -> int: ...
diff --git a/bindings/python/src/rtp/server.rs b/bindings/python/src/rtp/server.rs
index 9dc2cf8e1..a7e80106c 100644
--- a/bindings/python/src/rtp/server.rs
+++ b/bindings/python/src/rtp/server.rs
@@ -29,8 +29,8 @@ use tst_rtp::rtsp::server::{RtspServer as RustRtspServer, ServerStats as RustSer
 
 use crate::errors::{make_rtp_error, make_rtsp_error};
 use crate::mux::{
-    PyAudioStreamHandle, PyKlvStreamHandle, PyMuxerProgramConfig, PySubtitleStreamHandle,
-    PyVideoStreamHandle, py_pts90khz,
+    PyAudioStreamHandle, PyDataStreamHandle, PyKlvStreamHandle, PyMuxerProgramConfig,
+    PySubtitleStreamHandle, PyVideoStreamHandle, py_pts90khz,
 };
 
 // ---------------------------------------------------------------------------
@@ -382,6 +382,22 @@ impl PyMountHandle {
         res.map_err(|e| mount_error_to_pyerr(py, e))
     }
 
+    /// Push one data payload onto the lone configured data stream.
+    /// Pass-through: lands verbatim as one PES packet on stream_id 0xBD.
+    #[pyo3(signature = (data, *, pts))]
+    pub fn push_data(
+        &self,
+        py: Python<'_>,
+        data: &Bound<'_, PyAny>,
+        pts: &Bound<'_, PyAny>,
+    ) -> PyResult<()> {
+        let rust_pts = py_pts90khz(pts)?;
+        let coerced = coerce_bytes_like(py, data)?;
+        let slice = coerced.as_bytes();
+        let res = py.allow_threads(|| self.inner.push_data(slice, rust_pts));
+        res.map_err(|e| mount_error_to_pyerr(py, e))
+    }
+
     // ── Push surface — multi-stream variants ───────────────────────────────
     //
     // The `_to` variants take an explicit stream handle (obtained from
@@ -466,6 +482,22 @@ impl PyMountHandle {
         res.map_err(|e| mount_error_to_pyerr(py, e))
     }
 
+    #[pyo3(signature = (handle, data, *, pts))]
+    pub fn push_data_to(
+        &self,
+        py: Python<'_>,
+        handle: PyRef<'_, PyDataStreamHandle>,
+        data: &Bound<'_, PyAny>,
+        pts: &Bound<'_, PyAny>,
+    ) -> PyResult<()> {
+        let rust_pts = py_pts90khz(pts)?;
+        let handle_inner = handle.0;
+        let coerced = coerce_bytes_like(py, data)?;
+        let slice = coerced.as_bytes();
+        let res = py.allow_threads(|| self.inner.push_data_to(handle_inner, slice, rust_pts));
+        res.map_err(|e| mount_error_to_pyerr(py, e))
+    }
+
     // ── Stream-handle accessors ────────────────────────────────────────────
     //
     // Return the first configured handle of each kind (single-stream
@@ -513,6 +545,16 @@ impl PyMountHandle {
             .map(PySubtitleStreamHandle)
     }
 
+    /// Get the first configured data stream handle, or `None` if none
+    /// declared.
+    pub fn data_handle(&self) -> Option<PyDataStreamHandle> {
+        self.inner
+            .data_handles()
+            .into_iter()
+            .next()
+            .map(PyDataStreamHandle)
+    }
+
     /// All configured video stream handles.
     pub fn video_handles(&self) -> Vec<PyVideoStreamHandle> {
         self.inner
@@ -549,6 +591,15 @@ impl PyMountHandle {
             .collect()
     }
 
+    /// All configured data stream handles.
+    pub fn data_handles(&self) -> Vec<PyDataStreamHandle> {
+        self.inner
+            .data_handles()
+            .into_iter()
+            .map(PyDataStreamHandle)
+            .collect()
+    }
+
     // ── Lifecycle helpers ──────────────────────────────────────────────────
 
     /// Drain any TS packets queued in the inner muxer and broadcast them
diff --git a/bindings/python/src/srt/managed_convenience.rs b/bindings/python/src/srt/managed_convenience.rs
index 65607da9d..b8a9c228b 100644
--- a/bindings/python/src/srt/managed_convenience.rs
+++ b/bindings/python/src/srt/managed_convenience.rs
@@ -53,7 +53,7 @@ use tst_srt::{Listener, ListenerConfig, Socket, SocketConfig, SrtTransport, SrtU
 
 use crate::errors::{make_demux_error, make_srt_error, mux_error_to_pyerr};
 use crate::mux::{
-    PyAudioStreamHandle, PyKlvStreamHandle, PyMuxerProgramConfig, PyMuxerStats,
+    PyAudioStreamHandle, PyDataStreamHandle, PyKlvStreamHandle, PyMuxerProgramConfig, PyMuxerStats,
     PySubtitleStreamHandle, PyVideoStreamHandle, py_pts90khz,
 };
 use crate::srt::errors::{
@@ -286,8 +286,6 @@ impl PyManagedMuxSender {
     }
 
     // ── Push family — single-stream variants ──────────────────────────────
-    // No `push_data` here deliberately (W3 private-data arc) — same gap as
-    // rtp's `PyMountHandle`; recorded internal-consistency follow-up, W5.
 
     /// Push one video access unit onto the lone configured video stream.
     /// Annex-B framing for H.264/H.265/H.266; raw OBU stream for AV1.
@@ -368,6 +366,26 @@ impl PyManagedMuxSender {
         res.map_err(|e| mux_sender_error_to_pyerr(py, e))
     }
 
+    /// Push one data payload onto the lone configured data stream.
+    /// Pass-through: lands verbatim as one PES packet on stream_id 0xBD.
+    #[pyo3(signature = (data, *, pts))]
+    fn push_data(
+        &self,
+        py: Python<'_>,
+        data: &Bound<'_, PyAny>,
+        pts: &Bound<'_, PyAny>,
+    ) -> PyResult<()> {
+        let inner = self
+            .inner
+            .as_ref()
+            .ok_or_else(|| make_srt_error(py, "CLOSED", "ManagedMuxSender is closed"))?;
+        let rust_pts = py_pts90khz(pts)?;
+        let coerced = coerce_bytes_like(py, data)?;
+        let slice = coerced.as_bytes();
+        let res = py.allow_threads(|| inner.send_data(slice, rust_pts));
+        res.map_err(|e| mux_sender_error_to_pyerr(py, e))
+    }
+
     // ── Push family — handle-targeted variants ────────────────────────────
 
     #[pyo3(signature = (handle, nal, *, pts, key_frame = false))]
@@ -455,6 +473,26 @@ impl PyManagedMuxSender {
         res.map_err(|e| mux_sender_error_to_pyerr(py, e))
     }
 
+    #[pyo3(signature = (handle, data, *, pts))]
+    fn push_data_to(
+        &self,
+        py: Python<'_>,
+        handle: PyRef<'_, PyDataStreamHandle>,
+        data: &Bound<'_, PyAny>,
+        pts: &Bound<'_, PyAny>,
+    ) -> PyResult<()> {
+        let inner = self
+            .inner
+            .as_ref()
+            .ok_or_else(|| make_srt_error(py, "CLOSED", "ManagedMuxSender is closed"))?;
+        let rust_pts = py_pts90khz(pts)?;
+        let handle_inner = handle.0;
+        let coerced = coerce_bytes_like(py, data)?;
+        let slice = coerced.as_bytes();
+        let res = py.allow_threads(|| inner.send_data_to(handle_inner, slice, rust_pts));
+        res.map_err(|e| mux_sender_error_to_pyerr(py, e))
+    }
+
     // ── Handle getters ────────────────────────────────────────────────────
 
     fn video_handle(&self) -> Option<PyVideoStreamHandle> {
@@ -493,6 +531,15 @@ impl PyManagedMuxSender {
             .map(PySubtitleStreamHandle)
     }
 
+    fn data_handle(&self) -> Option<PyDataStreamHandle> {
+        let inner = self.inner.as_ref()?;
+        inner
+            .data_handles()
+            .into_iter()
+            .next()
+            .map(PyDataStreamHandle)
+    }
+
     // ── Stats ──────────────────────────────────────────────────────────────
 
     /// `(SocketStats, MuxerStats)` snapshot. Same shape as T5's
diff --git a/bindings/python/tests/test_rtsp_server.py b/bindings/python/tests/test_rtsp_server.py
index 3bd9aa758..a04906086 100644
--- a/bindings/python/tests/test_rtsp_server.py
+++ b/bindings/python/tests/test_rtsp_server.py
@@ -20,6 +20,7 @@
 from tstrans.exceptions import RtspError, RtspErrorKind
 from tstrans.mpegts import (
     AudioCodec,
+    DataStreamHandle,
     KlvStreamType,
     MuxerProgramConfigBuilder,
     Pts90khz,
@@ -522,3 +523,83 @@ def test_reset_stats_zeros_counters():
         mount.reset_stats()
         assert mount.stats().bytes_pushed == 0
         assert mount.stats().packets_pushed == 0
+
+
+# ---------------------------------------------------------------------------
+# push_data / push_data_to / data_handle on MountHandle.
+# ---------------------------------------------------------------------------
+
+
+def _video_data_program():
+    """Video + one private data stream (user-private stream_type 0xF0)."""
+    return (
+        MuxerProgramConfigBuilder(1, 0x1000)
+        .add_video(0x1011, VideoCodec.H264)
+        .add_data(0x1F0, 0xF0, carries_pts=True)
+        .build()
+    )
+
+
+_DATA_RECORD = b"\x01\x02\x03\x04private-mount-data"
+
+
+def test_data_handle_returns_none_for_video_only_program():
+    """data_handle() returns None when no data stream is configured."""
+    cfg = RtspServerConfig(bind_addr="127.0.0.1:0", graceful_shutdown_drain_ms=50)
+    with RtspServer.start(cfg) as server:
+        mount = server.add_unicast_mount("/live", _single_video_program())
+        assert mount.data_handle() is None
+
+
+def test_data_handle_returns_handle_for_data_program():
+    """data_handle() returns a DataStreamHandle when a data stream is
+    configured in the program."""
+    cfg = RtspServerConfig(bind_addr="127.0.0.1:0", graceful_shutdown_drain_ms=50)
+    with RtspServer.start(cfg) as server:
+        mount = server.add_unicast_mount("/live", _video_data_program())
+        h = mount.data_handle()
+        assert h is not None
+        assert isinstance(h, DataStreamHandle)
+        # data_handles() (list form) mirrors the other *_handles() accessors.
+        handles = mount.data_handles()
+        assert len(handles) == 1
+        assert all(isinstance(x, DataStreamHandle) for x in handles)
+
+
+def test_push_data_succeeds_with_no_peers():
+    """push_data to a mount with no connected peers succeeds (broadcast
+    no-receivers path; muxer consumes the payload, fanout is suppressed)."""
+    cfg = RtspServerConfig(bind_addr="127.0.0.1:0", graceful_shutdown_drain_ms=50)
+    with RtspServer.start(cfg) as server:
+        mount = server.add_unicast_mount("/live", _video_data_program())
+        mount.push_data(_DATA_RECORD, pts=Pts90khz(0))
+
+
+def test_push_data_to_with_explicit_handle():
+    """push_data_to with an explicit DataStreamHandle succeeds."""
+    cfg = RtspServerConfig(bind_addr="127.0.0.1:0", graceful_shutdown_drain_ms=50)
+    with RtspServer.start(cfg) as server:
+        mount = server.add_unicast_mount("/live", _video_data_program())
+        h = mount.data_handle()
+        assert h is not None
+        mount.push_data_to(h, _DATA_RECORD, pts=Pts90khz(0))
+
+
+def test_push_data_updates_mount_stats():
+    """push_data contributes to bytes_pushed (same accounting as push_video)."""
+    cfg = RtspServerConfig(bind_addr="127.0.0.1:0", graceful_shutdown_drain_ms=50)
+    with RtspServer.start(cfg) as server:
+        mount = server.add_unicast_mount("/live", _video_data_program())
+        # Push a key-frame so the muxer can emit its first TS packet set
+        # (muxer holds back until it can emit a PCR-bearing adaptation).
+        mount.push_video(_IDR_NAL, pts=Pts90khz(0), key_frame=True)
+        mount.push_data(_DATA_RECORD, pts=Pts90khz(3000))
+        assert mount.stats().bytes_pushed > 0
+
+
+def test_push_data_accepts_bytearray():
+    """push_data accepts bytearray (bytes-like coercion path)."""
+    cfg = RtspServerConfig(bind_addr="127.0.0.1:0", graceful_shutdown_drain_ms=50)
+    with RtspServer.start(cfg) as server:
+        mount = server.add_unicast_mount("/live", _video_data_program())
+        mount.push_data(bytearray(_DATA_RECORD), pts=Pts90khz(0))
diff --git a/bindings/python/tests/test_srt_managed_convenience.py b/bindings/python/tests/test_srt_managed_convenience.py
index 85470915f..5bf010fff 100644
--- a/bindings/python/tests/test_srt_managed_convenience.py
+++ b/bindings/python/tests/test_srt_managed_convenience.py
@@ -29,6 +29,7 @@
 import tstrans.srt
 from tstrans.exceptions import SrtError, SrtErrorKind
 from tstrans.mpegts import (
+    DataStreamHandle,
     DemuxEvent,
     DemuxerConfig,
     MuxerProgramConfigBuilder,
@@ -66,6 +67,16 @@ def _video_only_program() -> object:
     )
 
 
+def _video_data_program() -> object:
+    """Video + one private data stream (user-private stream_type 0xF0)."""
+    return (
+        MuxerProgramConfigBuilder(1, 0x100)
+        .add_video(0x101, VideoCodec.H264)
+        .add_data(0x1F0, 0xF0, carries_pts=True)
+        .build()
+    )
+
+
 def _fast_policy() -> ReconnectPolicy:
     """ReconnectPolicy with zero-wait constant backoff so tests don't sleep."""
     return ReconnectPolicy(
@@ -322,3 +333,208 @@ def test_managed_demux_receiver_iter_returns_self() -> None:
     finally:
         sender.close()
         receiver.close()
+
+
+# --------------------------------------------------------------------------- #
+# push_data / push_data_to / data_handle                                      #
+# --------------------------------------------------------------------------- #
+
+
+def test_data_handle_none_for_video_only_program() -> None:
+    """data_handle() returns None when no data stream is configured."""
+    port = _free_tcp_port()
+    sender, receiver = _make_managed_pair(port)
+    try:
+        assert sender.data_handle() is None
+    finally:
+        sender.close()
+        receiver.close()
+
+
+def test_data_handle_returns_handle_for_data_program() -> None:
+    """data_handle() returns a DataStreamHandle when a data stream is
+    configured in the program."""
+    port = _free_tcp_port()
+    listener_url = f"srt://:{port}?mode=listener"
+    caller_url = f"srt://127.0.0.1:{port}?mode=caller"
+
+    rx_box: list[ManagedDemuxReceiver] = []
+    rx_err: list[BaseException] = []
+
+    def accept_worker() -> None:
+        try:
+            r = ManagedDemuxReceiver.from_url(listener_url, policy=_fast_policy())
+            rx_box.append(r)
+        except BaseException as exc:  # noqa: BLE001
+            rx_err.append(exc)
+
+    t = threading.Thread(target=accept_worker, daemon=True)
+    t.start()
+    time.sleep(0.1)
+    sender = ManagedMuxSender.from_url(
+        caller_url, _video_data_program(), policy=_fast_policy()
+    )
+    t.join(timeout=5.0)
+    if rx_err:
+        sender.close()
+        raise rx_err[0]
+    receiver = rx_box[0]
+    try:
+        h = sender.data_handle()
+        assert h is not None
+        assert isinstance(h, DataStreamHandle)
+    finally:
+        sender.close()
+        receiver.close()
+
+
+def test_push_data_on_closed_sender_raises_closed() -> None:
+    """push_data on a closed ManagedMuxSender raises SrtError(CLOSED)."""
+    port = _free_tcp_port()
+    listener_url = f"srt://:{port}?mode=listener"
+    caller_url = f"srt://127.0.0.1:{port}?mode=caller"
+
+    rx_box: list[ManagedDemuxReceiver] = []
+
+    def accept_worker() -> None:
+        try:
+            r = ManagedDemuxReceiver.from_url(listener_url, policy=_fast_policy())
+            rx_box.append(r)
+        except BaseException:  # noqa: BLE001
+            pass
+
+    t = threading.Thread(target=accept_worker, daemon=True)
+    t.start()
+    time.sleep(0.1)
+    sender = ManagedMuxSender.from_url(
+        caller_url, _video_data_program(), policy=_fast_policy()
+    )
+    t.join(timeout=5.0)
+    if rx_box:
+        rx_box[0].close()
+    sender.close()
+    with pytest.raises(SrtError) as exc_info:
+        sender.push_data(b"\x01\x02\x03", pts=Pts90khz.from_raw(0))
+    assert exc_info.value.kind == SrtErrorKind.CLOSED
+
+
+def test_push_data_round_trips_payload_fidelity() -> None:
+    """End-to-end over a managed SRT loopback: push distinct payloads via
+    both `push_data` (single-stream shorthand) and `push_data_to`
+    (explicit handle); the receiver-side demuxer must surface them
+    byte-faithfully as `UnknownSample` events on the data PID (0x1F0,
+    stream_type 0xF0), with the pushed PTS preserved (carries_pts=True).
+
+    The consumer thread only *collects* events — all fidelity assertions
+    run on the main thread after draining, so a genuine payload/pts
+    mismatch surfaces as a real test failure instead of being swallowed
+    into a background-thread exception list."""
+    port = _free_tcp_port()
+    listener_url = f"srt://:{port}?mode=listener"
+    caller_url = f"srt://127.0.0.1:{port}?mode=caller"
+
+    DATA_PID = 0x1F0
+    DATA_STREAM_TYPE = 0xF0
+    PAYLOAD_SHORTHAND = b"\x01\x02\x03\x04shorthand-record"
+    PAYLOAD_HANDLE = b"\x05\x06\x07\x08handle-record"
+    # carries_pts=True → the pushed PTS round-trips verbatim (non-zero).
+    PTS_BASE = 900_000
+    PTS_STEP = 3000
+    N = 24
+    pushed_pts = {PTS_BASE + i * PTS_STEP for i in range(N)}
+    # Key-frame video NAL so the demuxer reaches a ProgramMap and starts
+    # surfacing samples promptly.
+    NAL_IDR_DATA = b"\x00\x00\x00\x01\x65\xBB"
+
+    rx_box: list[ManagedDemuxReceiver] = []
+    rx_err: list[BaseException] = []
+
+    def accept_worker() -> None:
+        try:
+            r = ManagedDemuxReceiver.from_url(listener_url, policy=_fast_policy())
+            rx_box.append(r)
+        except BaseException as exc:  # noqa: BLE001
+            rx_err.append(exc)
+
+    t = threading.Thread(target=accept_worker, daemon=True)
+    t.start()
+    time.sleep(0.1)
+    sender = ManagedMuxSender.from_url(
+        caller_url, _video_data_program(), policy=_fast_policy()
+    )
+    t.join(timeout=5.0)
+    if rx_err:
+        sender.close()
+        raise rx_err[0]
+    receiver = rx_box[0]
+
+    data_samples: list[object] = []
+    consumer_err: list[BaseException] = []
+
+    def consumer() -> None:
+        # Collect-only: append every UnknownSample, break once both
+        # target payloads have been observed. No assertions here.
+        try:
+            seen_payloads: set[bytes] = set()
+            for ev in receiver:
+                if isinstance(ev, DemuxEvent.UnknownSample):
+                    data_samples.append(ev)
+                    seen_payloads.add(bytes(ev.payload))
+                    if {PAYLOAD_SHORTHAND, PAYLOAD_HANDLE} <= seen_payloads:
+                        break
+        except BaseException as exc:  # noqa: BLE001
+            consumer_err.append(exc)
+
+    ct = threading.Thread(target=consumer, daemon=True)
+    ct.start()
+    # Park the receiver inside recv_event before pushing.
+    time.sleep(0.2)
+    try:
+        data_h = sender.data_handle()
+        assert data_h is not None
+        for i in range(N):
+            pts = Pts90khz.from_raw(PTS_BASE + i * PTS_STEP)
+            sender.push_video(NAL_IDR_DATA, pts=pts, key_frame=(i % 4 == 0))
+            sender.push_data(PAYLOAD_SHORTHAND, pts=pts)
+            sender.push_data_to(data_h, PAYLOAD_HANDLE, pts=pts)
+        # Give TSBPD time to release the buffered packets to the consumer.
+        time.sleep(0.5)
+    finally:
+        # Close the RECEIVER first: its cancel handle fires while the
+        # consumer is parked in recv_event (after it has drained the
+        # buffered samples), unblocking it cleanly. Closing the sender
+        # first would break the SRT link and make the managed receiver
+        # attempt a reconnect, parking in a blocking re-accept the cancel
+        # handle can't interrupt — a dropped/mismatched record would then
+        # HANG the test instead of failing loud.
+        receiver.close()
+        ct.join(timeout=5.0)
+        sender.close()
+
+    if not data_samples and consumer_err:
+        pytest.fail(f"consumer raised before any data sample: {consumer_err}")
+
+    # ── Fidelity assertions (main thread) ──────────────────────────────
+    assert data_samples, (
+        "expected at least one UnknownSample on the data stream; got none "
+        "(a regression dropping every data record would land here)"
+    )
+    payloads = {bytes(s.payload) for s in data_samples}
+    assert PAYLOAD_SHORTHAND in payloads, (
+        f"push_data payload not round-tripped; observed payloads={payloads!r}"
+    )
+    assert PAYLOAD_HANDLE in payloads, (
+        f"push_data_to payload not round-tripped; observed payloads={payloads!r}"
+    )
+    # No corruption: every surfaced data payload is exactly one of the two
+    # we pushed, on the right PID / stream_type, with a preserved PTS.
+    for s in data_samples:
+        assert bytes(s.payload) in (PAYLOAD_SHORTHAND, PAYLOAD_HANDLE), (
+            f"unexpected/corrupt data payload: {bytes(s.payload)!r}"
+        )
+        assert s.stream.pid == DATA_PID
+        assert s.stream_type == DATA_STREAM_TYPE
+        # carries_pts=True ⇒ pts is one of the pushed (non-zero) values,
+        # not the demuxer's no-PTS substitute of 0.
+        assert s.pts.raw != 0
+        assert s.pts.raw in pushed_pts
diff --git a/crates/tst-rtp/public-api.txt b/crates/tst-rtp/public-api.txt
index 44fd3b0e0..693441ad1 100644
--- a/crates/tst-rtp/public-api.txt
+++ b/crates/tst-rtp/public-api.txt
@@ -866,6 +866,7 @@ impl core::panic::unwind_safe::UnwindSafe for tst_rtp::rtsp::server::mount::Moun
 pub struct tst_rtp::rtsp::server::mount::MountHandle
 impl tst_rtp::rtsp::server::mount::MountHandle
 pub fn tst_rtp::rtsp::server::mount::MountHandle::audio_handles(&self) -> alloc::vec::Vec<tst_core::mpegts::mux::types::AudioStreamHandle>
+pub fn tst_rtp::rtsp::server::mount::MountHandle::data_handles(&self) -> alloc::vec::Vec<tst_core::mpegts::mux::types::DataStreamHandle>
 pub fn tst_rtp::rtsp::server::mount::MountHandle::flush(&self)
 pub fn tst_rtp::rtsp::server::mount::MountHandle::klv_handles(&self) -> alloc::vec::Vec<tst_core::mpegts::mux::types::KlvStreamHandle>
 pub fn tst_rtp::rtsp::server::mount::MountHandle::mount_kind(&self) -> &tst_rtp::rtsp::server::mount::MountKind
@@ -873,6 +874,8 @@ pub fn tst_rtp::rtsp::server::mount::MountHandle::mount_path(&self) -> &str
 pub fn tst_rtp::rtsp::server::mount::MountHandle::peer_count(&self) -> usize
 pub fn tst_rtp::rtsp::server::mount::MountHandle::push_audio(&self, &[u8], tst_core::mpegts::common::Pts90khz) -> core::result::Result<(), tst_rtp::error::MountError>
 pub fn tst_rtp::rtsp::server::mount::MountHandle::push_audio_to(&self, tst_core::mpegts::mux::types::AudioStreamHandle, &[u8], tst_core::mpegts::common::Pts90khz) -> core::result::Result<(), tst_rtp::error::MountError>
+pub fn tst_rtp::rtsp::server::mount::MountHandle::push_data(&self, &[u8], tst_core::mpegts::common::Pts90khz) -> core::result::Result<(), tst_rtp::error::MountError>
+pub fn tst_rtp::rtsp::server::mount::MountHandle::push_data_to(&self, tst_core::mpegts::mux::types::DataStreamHandle, &[u8], tst_core::mpegts::common::Pts90khz) -> core::result::Result<(), tst_rtp::error::MountError>
 pub fn tst_rtp::rtsp::server::mount::MountHandle::push_klv(&self, &[u8], tst_core::mpegts::common::Pts90khz, u8) -> core::result::Result<(), tst_rtp::error::MountError>
 pub fn tst_rtp::rtsp::server::mount::MountHandle::push_klv_to(&self, tst_core::mpegts::mux::types::KlvStreamHandle, &[u8], tst_core::mpegts::common::Pts90khz, u8) -> core::result::Result<(), tst_rtp::error::MountError>
 pub fn tst_rtp::rtsp::server::mount::MountHandle::push_subtitle(&self, &[u8], tst_core::mpegts::common::Pts90khz) -> core::result::Result<(), tst_rtp::error::MountError>
@@ -1779,6 +1782,7 @@ impl<W> core::panic::unwind_safe::UnwindSafe for tst_rtp::rtsp::interleaved::Int
 pub struct tst_rtp::MountHandle
 impl tst_rtp::rtsp::server::mount::MountHandle
 pub fn tst_rtp::rtsp::server::mount::MountHandle::audio_handles(&self) -> alloc::vec::Vec<tst_core::mpegts::mux::types::AudioStreamHandle>
+pub fn tst_rtp::rtsp::server::mount::MountHandle::data_handles(&self) -> alloc::vec::Vec<tst_core::mpegts::mux::types::DataStreamHandle>
 pub fn tst_rtp::rtsp::server::mount::MountHandle::flush(&self)
 pub fn tst_rtp::rtsp::server::mount::MountHandle::klv_handles(&self) -> alloc::vec::Vec<tst_core::mpegts::mux::types::KlvStreamHandle>
 pub fn tst_rtp::rtsp::server::mount::MountHandle::mount_kind(&self) -> &tst_rtp::rtsp::server::mount::MountKind
@@ -1786,6 +1790,8 @@ pub fn tst_rtp::rtsp::server::mount::MountHandle::mount_path(&self) -> &str
 pub fn tst_rtp::rtsp::server::mount::MountHandle::peer_count(&self) -> usize
 pub fn tst_rtp::rtsp::server::mount::MountHandle::push_audio(&self, &[u8], tst_core::mpegts::common::Pts90khz) -> core::result::Result<(), tst_rtp::error::MountError>
 pub fn tst_rtp::rtsp::server::mount::MountHandle::push_audio_to(&self, tst_core::mpegts::mux::types::AudioStreamHandle, &[u8], tst_core::mpegts::common::Pts90khz) -> core::result::Result<(), tst_rtp::error::MountError>
+pub fn tst_rtp::rtsp::server::mount::MountHandle::push_data(&self, &[u8], tst_core::mpegts::common::Pts90khz) -> core::result::Result<(), tst_rtp::error::MountError>
+pub fn tst_rtp::rtsp::server::mount::MountHandle::push_data_to(&self, tst_core::mpegts::mux::types::DataStreamHandle, &[u8], tst_core::mpegts::common::Pts90khz) -> core::result::Result<(), tst_rtp::error::MountError>
 pub fn tst_rtp::rtsp::server::mount::MountHandle::push_klv(&self, &[u8], tst_core::mpegts::common::Pts90khz, u8) -> core::result::Result<(), tst_rtp::error::MountError>
 pub fn tst_rtp::rtsp::server::mount::MountHandle::push_klv_to(&self, tst_core::mpegts::mux::types::KlvStreamHandle, &[u8], tst_core::mpegts::common::Pts90khz, u8) -> core::result::Result<(), tst_rtp::error::MountError>
 pub fn tst_rtp::rtsp::server::mount::MountHandle::push_subtitle(&self, &[u8], tst_core::mpegts::common::Pts90khz) -> core::result::Result<(), tst_rtp::error::MountError>
diff --git a/crates/tst-rtp/src/rtsp/server/mount.rs b/crates/tst-rtp/src/rtsp/server/mount.rs
index 6d6743320..5c441cd24 100644
--- a/crates/tst-rtp/src/rtsp/server/mount.rs
+++ b/crates/tst-rtp/src/rtsp/server/mount.rs
@@ -298,6 +298,28 @@ impl MountHandle {
         Ok(())
     }
 
+    /// Push one data payload. Mirror of
+    /// `tst_pipeline::MuxSender::send_data`.
+    ///
+    /// Pass-through: no AU-cell wrap, no framing, no inspection — `data`
+    /// lands verbatim as one PES packet on `stream_id` 0xBD. See
+    /// [`Self::push_video`] for the drain + broadcast contract and error
+    /// mapping.
+    pub fn push_data(
+        &self,
+        data: &[u8],
+        pts: tst_core::mpegts::common::Pts90khz,
+    ) -> Result<(), crate::error::MountError> {
+        let mut muxer = self
+            .state
+            .muxer
+            .lock()
+            .map_err(|_| crate::error::MountError::Closed)?;
+        muxer.push_data(data, pts)?;
+        drain_and_broadcast(&mut muxer, &self.state);
+        Ok(())
+    }
+
     // ── Multi-stream / multi-program variants ────────────────────────────
     //
     // Use these when the mount's `MuxerConfig` declares more than one
@@ -388,6 +410,24 @@ impl MountHandle {
         Ok(())
     }
 
+    /// Push to a specific data stream handle. Mirror of
+    /// `MuxSender::send_data_to`.
+    pub fn push_data_to(
+        &self,
+        handle: tst_core::mpegts::mux::DataStreamHandle,
+        data: &[u8],
+        pts: tst_core::mpegts::common::Pts90khz,
+    ) -> Result<(), crate::error::MountError> {
+        let mut muxer = self
+            .state
+            .muxer
+            .lock()
+            .map_err(|_| crate::error::MountError::Closed)?;
+        muxer.push_data_to(handle, data, pts)?;
+        drain_and_broadcast(&mut muxer, &self.state);
+        Ok(())
+    }
+
     // ── Lifecycle helpers ────────────────────────────────────────────────
 
     /// Drain any TS packets queued in the inner muxer and broadcast them
@@ -464,6 +504,14 @@ impl MountHandle {
             Err(_) => Vec::new(),
         }
     }
+
+    /// Configured data stream handles. Mirror of `Muxer::data_handles`.
+    pub fn data_handles(&self) -> Vec<tst_core::mpegts::mux::DataStreamHandle> {
+        match self.state.muxer.lock() {
+            Ok(m) => m.data_handles(),
+            Err(_) => Vec::new(),
+        }
+    }
 }
 
 /// Default RTP MPEG-TS payload size (7 × 188-byte TS packets). Matches
@@ -632,4 +680,38 @@ mod tests {
         let handles = handle.video_handles();
         assert_eq!(handles.len(), 1);
     }
+
+    fn mock_unicast_mount_state_with_data() -> Arc<MountState> {
+        // Program with H.264 video (for PCR) + one data stream (PCR-ineligible).
+        let mut prog = MuxerProgramConfigBuilder::new(1, 0x1000);
+        prog.add_video(0x1011, VideoCodec::H264);
+        prog.add_data(0x1012, 0x06, false);
+        let mut b = MuxerConfig::builder();
+        b.add_program(prog.build());
+        let cfg = b.build().unwrap();
+        MountState::new("/test", MountKind::Unicast, cfg, 256).unwrap()
+    }
+
+    #[test]
+    fn push_data_without_subscribers_succeeds() {
+        let state = mock_unicast_mount_state_with_data();
+        let handle = MountHandle { state };
+        handle
+            .push_data(&[0xDE, 0xAD, 0xBE, 0xEF], Pts90khz::new(0))
+            .expect("push_data succeeds even with no peers");
+    }
+
+    #[test]
+    fn push_data_to_specific_handle_succeeds() {
+        let state = mock_unicast_mount_state_with_data();
+        let handle = MountHandle { state };
+        let h = handle
+            .data_handles()
+            .into_iter()
+            .next()
+            .expect("one data handle configured");
+        handle
+            .push_data_to(h, &[0x01, 0x02], Pts90khz::new(900))
+            .expect("push_data_to succeeds");
+    }
 }
diff --git a/scripts/check/c/abi-rustdoc-coverage.sh b/scripts/check/c/abi-rustdoc-coverage.sh
index 9c823c379..0e6f19cfb 100755
--- a/scripts/check/c/abi-rustdoc-coverage.sh
+++ b/scripts/check/c/abi-rustdoc-coverage.sh
@@ -95,6 +95,8 @@ ALLOWLIST=(
     "tst_managed_mux_sender_send_audio_to"
     "tst_managed_mux_sender_send_subtitle"
     "tst_managed_mux_sender_send_subtitle_to"
+    "tst_managed_mux_sender_send_data"
+    "tst_managed_mux_sender_send_data_to"
     "tst_managed_raw_sender_close"
     "tst_managed_raw_sender_get_stats"
     "tst_managed_raw_sender_reset_stats"
@@ -518,6 +520,8 @@ ALLOWLIST=(
     "tst_rtsp_mount_push_klv_to"
     "tst_rtsp_mount_push_audio_to"
     "tst_rtsp_mount_push_subtitle_to"
+    "tst_rtsp_mount_push_data"
+    "tst_rtsp_mount_push_data_to"
     "tst_rtsp_mount_flush"
     "tst_rtsp_mount_cancel"
     "tst_rtsp_mount_reset_stats"
diff --git a/tests/coverage/surface-manifest.toml b/tests/coverage/surface-manifest.toml
index 8684f6700..85ca699c5 100644
--- a/tests/coverage/surface-manifest.toml
+++ b/tests/coverage/surface-manifest.toml
@@ -7516,6 +7516,10 @@ reason = "bulk bootstrap 2026-06-03; graduate to [[surface]] as coverage is asse
 item = "tst_rtp::rtsp::server::mount::MountHandle::fmt"
 reason = "bulk bootstrap 2026-06-03; graduate to [[surface]] as coverage is asserted"
 
+[[exempt]]
+item = "tst_rtp::rtsp::server::mount::MountHandle::data_handles"
+reason = "bulk bootstrap 2026-06-03; graduate to [[surface]] as coverage is asserted"
+
 [[exempt]]
 item = "tst_rtp::rtsp::server::mount::MountHandle::klv_handles"
 reason = "bulk bootstrap 2026-06-03; graduate to [[surface]] as coverage is asserted"
@@ -7548,6 +7552,14 @@ reason = "bulk bootstrap 2026-06-03; graduate to [[surface]] as coverage is asse
 item = "tst_rtp::rtsp::server::mount::MountHandle::push_klv_to"
 reason = "bulk bootstrap 2026-06-03; graduate to [[surface]] as coverage is asserted"
 
+[[exempt]]
+item = "tst_rtp::rtsp::server::mount::MountHandle::push_data"
+reason = "bulk bootstrap 2026-06-03; graduate to [[surface]] as coverage is asserted"
+
+[[exempt]]
+item = "tst_rtp::rtsp::server::mount::MountHandle::push_data_to"
+reason = "bulk bootstrap 2026-06-03; graduate to [[surface]] as coverage is asserted"
+
 [[exempt]]
 item = "tst_rtp::rtsp::server::mount::MountHandle::push_subtitle"
 reason = "bulk bootstrap 2026-06-03; graduate to [[surface]] as coverage is asserted"