mux: sink and reader audit fixes; DTS in MP4; halt-aware network receive

Fixes from the mkv, mp4, network and ts sink audit; track timing and
DiscardPadding ride the FMKV wire. MP4 writes DTS per ETSI TS 102 114
(ddts rates and timescales). The MKV PCM depth probe is bounded and
evidence-based. Network receives use keepalive and wait on readiness,
so a Stop or a dead peer ends them.

CHANGELOG.md

@@ -14,6 +14,8 @@
 - `list_drives` on Linux without sysfs keeps an sg node unless INQUIRY says it is not optical or open() fails with no device (ENOENT/ENXIO/ENODEV) or no permission (EACCES/EPERM), so a wedged or busy drive is still listed.
 - Dropping a transport or `Drive` unlocks the tray only if it locked it.
 - `Error::ImageTruncated` displays as `E<code>: have/want`.
+- `json://` audio `sample_rate` is now a number in Hz (`null` when unknown) instead of a display string such as `"48kHz"`; a new `sample_rates` array lists every rate a stream carries (e.g. `[48000, 96000]` for a 48/96 kHz combo).
+- `network://` / `stdio://` FMKV headers use version 2 when a track carries decoder timing (Opus CodecDelay/SeekPreRoll), adding per-frame DiscardPadding; streams without timing stay version 1.
 
 ### Removed
 
@@ -28,6 +30,9 @@
 ### Fixed
 
 - TrueHD channel labels keep the LFE split (3.0 was labelled "2.1", 7.0 "6.1"), and a 6-channel presentation counts its Lvh/Rvh pair as 2. An older network:// receiver parses the new layout strings as unknown and omits the MKV Channels element for those tracks; keep both ends on the same version.
+- `mp4://` DTS tracks write the stream's maximum rate in `ddts` and read back as 96 kHz where the source is 96 kHz.
+- `mkv://` PCM tracks without BitDepth: bounded read-ahead and a multi-block depth estimate.
+- `network://` receive arms TCP keepalive on every accepted socket; a rejected keepalive option only warns.
 
 ## [1.7.7] — 2026-09-26
 

Cargo.lock

@@ -328,9 +328,11 @@ dependencies = [
  "memchr",
  "rayon",
  "roxmltree",
+ "rustix",
  "serde",
  "serde_json",
  "sha1",
+ "socket2",
  "tempfile",
  "tracing",
  "zip",
@@ -626,6 +628,16 @@ version = "1.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b9be42f50aa861c555654aa3a37f52f4b1074bacf4e48fe0ef7fa584e80f1f0f"
 
+[[package]]
+name = "socket2"
+version = "0.6.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4"
+dependencies = [
+ "libc",
+ "windows-sys",
+]
+
 [[package]]
 name = "syn"
 version = "2.0.119"

Cargo.toml

@@ -48,6 +48,8 @@ rip = [
     "dep:crossbeam-channel",
     "dep:rayon",
     "dep:memchr",
+    "dep:socket2",
+    "dep:rustix",
 ]
 
 [dependencies]
@@ -62,6 +64,12 @@ aes = { version = "0.9", optional = true }
 freemkv-unlock = { path = "../freemkv-unlock", optional = true }
 zip = { version = "8", default-features = false, features = ["deflate"], optional = true }
 base64 = { version = "0.23", optional = true }
+# Safe TCP keepalive setup for `network://` receive (dead-peer detection);
+# `all` provides TcpKeepalive::with_retries.
+socket2 = { version = "0.6", features = ["all"], optional = true }
+# Safe readiness poll (poll/WSAPoll) so a halt-aware `network://` receive never
+# arms SO_RCVTIMEO, which on Windows can drop data when it cancels a recv.
+rustix = { version = "1", default-features = false, features = ["std", "event"], optional = true }
 # Read-only XML DOM parser (pure Rust, forbid(unsafe_code), entity-expansion
 # bounded). Parses the HD-DVD Advanced-Content playlist `ADV_OBJ/VPLST000.XPL`
 # — untrusted disc bytes — into authoritative titles/clips/chapters. A real

src/error.rs

@@ -198,6 +198,10 @@ pub const E_DIR_IMAGE_FANOUT: u16 = 9068;
 pub const E_SEAM_PLAN_DROPPED_MOST: u16 = 9069;
 /// A sink finished having written no frames at all.
 pub const E_SINK_WROTE_NOTHING: u16 = 9070;
+/// A write or finish reached a stream that is already finished or whose output failed.
+pub const E_STREAM_CLOSED: u16 = 9071;
+/// Per-track metadata was set after the stream header was already written.
+pub const E_STREAM_HEADER_WRITTEN: u16 = 9072;
 pub const E_M2TS_PACKET_MALFORMED: u16 = 9021;
 /// A `network://` output target resolved to no address that is safe to
 /// connect to (every resolved IP was loopback / private / link-local /
@@ -841,6 +845,10 @@ pub enum Error {
     },
     /// A sink finished having written no frames at all.
     SinkWroteNothing,
+    /// See [`E_STREAM_CLOSED`].
+    StreamClosed,
+    /// See [`E_STREAM_HEADER_WRITTEN`].
+    StreamHeaderWritten,
 }
 
 impl Error {
@@ -972,6 +980,8 @@ impl Error {
             Error::DirImageFanout { .. } => E_DIR_IMAGE_FANOUT,
             Error::SeamPlanDroppedMost { .. } => E_SEAM_PLAN_DROPPED_MOST,
             Error::SinkWroteNothing => E_SINK_WROTE_NOTHING,
+            Error::StreamClosed => E_STREAM_CLOSED,
+            Error::StreamHeaderWritten => E_STREAM_HEADER_WRITTEN,
             Error::DirImageFileChanged { .. } => E_DIR_IMAGE_FILE_CHANGED,
             Error::DirImageTooLarge => E_DIR_IMAGE_TOO_LARGE,
         }
@@ -1572,6 +1582,8 @@ mod tests {
             }
             .code(),
             Error::SinkWroteNothing.code(),
+            Error::StreamClosed.code(),
+            Error::StreamHeaderWritten.code(),
             Error::ShortImageRead {
                 lba: 0,
                 expected: 1,
@@ -2006,6 +2018,8 @@ mod tests {
             (Error::MkvLacingInvalid, E_MKV_LACING_INVALID),
             (Error::MkvSourceInvalid, E_MKV_SOURCE_INVALID),
             (Error::MkvUnencodable, E_MKV_UNENCODABLE),
+            (Error::StreamClosed, E_STREAM_CLOSED),
+            (Error::StreamHeaderWritten, E_STREAM_HEADER_WRITTEN),
             (Error::Mp4NoVideoTrack, E_MP4_NO_VIDEO_TRACK),
             (Error::Mp4Invalid, E_MP4_INVALID),
             (Error::Mp4MissingCodecPrivate, E_MP4_MISSING_CODEC_PRIVATE),

src/mux/demux_sink.rs

@@ -286,6 +286,9 @@ struct PgsSupWriter {
     // Delay a duration-derived clear until the next frame, so an original
     // clear (or replacement PCS) at that timestamp takes precedence.
     pending_clear: Option<(i64, u16, u16)>,
+    // Bytes of truncated trailing segments not written (a partial segment would
+    // desync the `.sup` framing); reported at finish.
+    truncated_bytes: u64,
 }
 
 // ── PGS / HDMV segment framing constants ─────────────────────────────────────
@@ -320,13 +323,14 @@ fn ns_to_90k(pts_ns: i64) -> u32 {
 
 impl PgsSupWriter {
     /// Walk the concatenated segments in `data`, emitting each with a `PG`
-    /// header carrying `pts90k`/`dts90k`. Returns bytes written.
+    /// header carrying `pts90k`/`dts90k`. Returns `(bytes written, trailing
+    /// bytes of a truncated segment left unwritten)`.
     fn emit_segments(
         data: &[u8],
         pts90k: u32,
         dts90k: u32,
         w: &mut dyn Write,
-    ) -> io::Result<usize> {
+    ) -> io::Result<(usize, usize)> {
         let mut pos = 0;
         let mut written = 0;
         // Each PGS segment in the payload is: type(1) + size(2 BE) + size bytes.
@@ -343,7 +347,7 @@ impl PgsSupWriter {
             written += SUP_HEADER_LEN + PGS_SEG_HEADER_LEN + size;
             pos = seg_end;
         }
-        Ok(written)
+        Ok((written, data.len() - pos))
     }
 
     // Synthetic "clear" display set (empty PCS + END) re-emitted at `display_pts + duration`;
@@ -402,14 +406,16 @@ impl EsWriter for PgsSupWriter {
             if end < pts_ns || (end == pts_ns && !is_pcs) {
                 let clear = Self::synthetic_clear_display_set(width, height);
                 let pts = ns_to_90k(end);
-                written += Self::emit_segments(&clear, pts, pts, w)?;
+                written += Self::emit_segments(&clear, pts, pts, w)?.0;
                 self.pending_clear = None;
             } else if is_pcs {
                 // A real clear/replacement at or before the computed end wins.
                 self.pending_clear = None;
             }
         }
-        written += Self::emit_segments(&f.data, pts90, pts90, w)?;
+        let (n, truncated) = Self::emit_segments(&f.data, pts90, pts90, w)?;
+        written += n;
+        self.truncated_bytes += truncated as u64;
         // Retain fallback support for old MKVs that only carry durations, and
         // for a final display without a clear. Never synthesize a clear OF a
         // clear (nor of a standalone WDS/END segment).
@@ -427,6 +433,13 @@ impl EsWriter for PgsSupWriter {
             let pts = ns_to_90k(end);
             Self::emit_segments(&clear, pts, pts, w)?;
         }
+        if self.truncated_bytes > 0 {
+            tracing::warn!(
+                target: "mux",
+                bytes = self.truncated_bytes,
+                "PGS .sup: truncated trailing segments were not written"
+            );
+        }
         Ok(())
     }
 }
@@ -1347,7 +1360,7 @@ mod tests {
         // One segment: type=0x16, size=2, payload=[0xDE,0xAD].
         let payload = [SEG_PCS, 0x00, 0x02, 0xDE, 0xAD];
         let mut out = Vec::new();
-        let written = PgsSupWriter::emit_segments(&payload, 0x10, 0x10, &mut out).unwrap();
+        let (written, _) = PgsSupWriter::emit_segments(&payload, 0x10, 0x10, &mut out).unwrap();
         assert_eq!(&out[0..2], &SUP_MAGIC);
         assert_eq!(&out[2..6], &0x10u32.to_be_bytes()); // PTS
         assert_eq!(&out[6..10], &0x10u32.to_be_bytes()); // DTS
@@ -1581,10 +1594,18 @@ mod tests {
             let mut bytes = Vec::new();
             assert_eq!(
                 PgsSupWriter::emit_segments(data, 0, 0, &mut bytes).unwrap(),
-                0
+                (0, data.len())
             );
             assert!(bytes.is_empty());
         }
+        // A frame ending in a truncated segment is accounted, not silently lost.
+        let mut writer = PgsSupWriter::default();
+        let mut frame = sup_frame(0, None, true);
+        frame.data.extend_from_slice(&[SEG_PCS, 0, 9, 1]);
+        writer
+            .write_frame(&mut Vec::new(), &frame, frame.pts)
+            .unwrap();
+        assert_eq!(writer.truncated_bytes, 4);
     }
 
     #[test]
@@ -2065,6 +2086,56 @@ mod tests {
         let _ = std::fs::remove_dir_all(&dir);
     }
 
+    // `audio://` where every AUDIO frame falls outside the marks but video frames
+    // map: filtered video must not count as written, so the export fails loudly.
+    #[test]
+    fn a_filtered_export_whose_persisted_track_was_emptied_fails() {
+        let dir = tempdir();
+        let mut title = title_with(
+            vec![video_stream(Codec::H264), audio_stream(Codec::Ac3, "eng")],
+            vec![None, None],
+        );
+        title.clips = (0..2u32)
+            .map(|i| crate::disc::Clip {
+                feed_span: None,
+                clip_id: format!("0000{i}"),
+                in_time: (100 + 100 * i) * 45_000,
+                out_time: (200 + 100 * i) * 45_000,
+                duration_secs: 100.0,
+                source_packets: 0,
+            })
+            .collect();
+        let opts = DemuxOptions {
+            base: "Empty".to_string(),
+            kind_filter: Some(TrackKind::Audio),
+            export_chapters: false,
+            ..Default::default()
+        };
+        let mut sink = DemuxSink::create(&dir, &title, &opts).unwrap();
+        let frame = |track: usize, pts: i64| PesFrame {
+            discard_padding_ns: 0,
+            coding: None,
+            source: None,
+            track,
+            pts,
+            keyframe: true,
+            data: vec![0x0B, 0x77],
+            duration_ns: None,
+        };
+        for s in [150i64, 151, 152] {
+            let _ = Stream::write(&mut sink, &frame(0, s * 1_000_000_000));
+        }
+        for s in [0i64, 1] {
+            let _ = Stream::write(&mut sink, &frame(1, s * 1_000_000_000));
+        }
+        let err = Stream::finish(&mut sink).expect_err("no audio was written");
+        assert_eq!(
+            crate::error::error_code(&err),
+            Some(crate::error::E_SINK_WROTE_NOTHING)
+        );
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
     #[test]
     fn sink_read_returns_write_only() {
         let dir = tempdir();

src/mux/demux_thread.rs

@@ -4,8 +4,9 @@
 //! consumer thread so feed and codec parse pipeline instead of serialise.
 //!
 //! [`DemuxThread::spawn_zero_copy`] returns a handle plus a
-//! `Receiver<DemuxBatch>`; dropping the handle closes the channel and
+//! `Receiver<DemuxBatch>`. `Drop::drop` joins the worker, which exits only once
-//! `Drop::drop` joins the worker.
+//! that receiver is gone: drop the receiver FIRST (an owner declares it before
+//! the handle), or a worker blocked on a full channel deadlocks the join.
 
 use crate::halt::Halt;
 use crossbeam_channel::{Receiver, Sender, bounded};
@@ -586,4 +587,38 @@ mod tests {
             "recycle_rx recv must fail after recycle_tx drop"
         );
     }
+
+    // PipelinedPesStream must drop its receiver before this handle (field order):
+    // a worker blocked on the full channel then exits instead of deadlocking.
+    #[test]
+    fn dropping_a_stream_with_a_blocked_worker_does_not_hang() {
+        let (pf_tx, pf_rx) = bounded::<std::io::Result<Vec<u8>>>(16);
+        let (rc_tx, _rc_rx) = bounded::<Vec<u8>>(16);
+        let pid = 0x1011;
+        let ts = super::super::ts::TsDemuxer::new(&[pid]);
+        let (dt, rx) =
+            DemuxThread::spawn_zero_copy(pf_rx, rc_tx, (), None, Some(ts), None).unwrap();
+        for i in 0..8u8 {
+            let mut chunk = bdts_pes_packet(pid, &[i]);
+            chunk.extend(bdts_pes_packet(pid, &[i, i]));
+            pf_tx.send(Ok(chunk)).unwrap();
+        }
+        drop(pf_tx);
+        std::thread::sleep(Duration::from_millis(200));
+        let stream = super::super::pipelined_stream::PipelinedPesStream::new(
+            dt,
+            rx,
+            crate::disc::DiscTitle::empty(),
+            Vec::new(),
+            Vec::new(),
+        );
+        let (done_tx, done_rx) = std::sync::mpsc::channel();
+        std::thread::spawn(move || {
+            drop(stream);
+            let _ = done_tx.send(());
+        });
+        done_rx
+            .recv_timeout(Duration::from_secs(10))
+            .expect("drop must not deadlock on the blocked demux worker");
+    }
 }

src/mux/driver.rs

@@ -22,7 +22,8 @@ use crate::sector::{FileSectorSource, KeyFetch, SectorSource};
 use crate::session::DiscSession;
 
 use super::resolve::{
-    InputOptions, StreamUrl, build_iso_pipeline, input, output, parse_url, resolve_mux_key_map,
+    InputOptions, StreamUrl, build_iso_pipeline, input_with_halt, output, parse_url,
+    resolve_mux_key_map,
 };
 use super::videomap::{Medium, SourceInfo};
 
@@ -112,11 +113,11 @@ pub enum MuxInput<'a> {
         key_map: Option<Arc<AacsKeyMap>>,
     },
     /// Any URL-addressed source (`iso://`, `mkv://`, `m2ts://`, `network://`,
-    /// stdio) opened via [`input`].
+    /// stdio) opened via [`input`](super::resolve::input).
     Url {
         /// The source URL.
         url: &'a str,
-        /// Input options forwarded to [`input`].
+        /// Input options forwarded to [`input`](super::resolve::input).
         opts: InputOptions,
     },
 }
@@ -245,7 +246,22 @@ pub fn mux_stream(
                     title: in_opts.title_index.unwrap_or(0),
                     ..SourceInfo::default()
                 };
-                let stream = input(url, &in_opts)?;
+                // A Stop while `network://` waits for its sender is a clean interrupt.
+                let stream = match input_with_halt(url, &in_opts, Some(halt)) {
+                    Ok(s) => s,
+                    Err(e) if crate::error::is_halt(&e) => {
+                        return Ok(MuxOutcome {
+                            completed: false,
+                            output_opened: false,
+                            bytes_written: 0,
+                            errors: 0,
+                            lost_bytes: 0,
+                            streams: 0,
+                            undelivered_streams: Vec::new(),
+                        });
+                    }
+                    Err(e) => return Err(e),
+                };
                 let source = SourceInfo {
                     playlist: stream.info().playlist.clone(),
                     ..source
@@ -507,6 +523,21 @@ fn reader_event_fn(events: Arc<dyn MuxEvents>) -> crate::sector::prefetched::Eve
     })
 }
 
+// Join the write consumer after the pump. A send that hit its deadline means the
+// consumer is wedged, so the join gets only the short grace, not JOIN_TIMEOUT.
+fn finish_pumped<I: Send + 'static, R: Send + 'static>(
+    pipe: Pipeline<I, R>,
+    halt: &Halt,
+    send_timed_out: bool,
+) -> Result<R, Error> {
+    if send_timed_out {
+        let wedged = Halt::new();
+        wedged.cancel();
+        return pipe.finish_with_halt(Some(&wedged));
+    }
+    pipe.finish_with_halt(Some(halt))
+}
+
 // Whether a finished mux counts as COMPLETED: interrupted, finalize_failed, or halt_cancelled
 // each force `false`. Pure fn so this mapping is unit-tested directly.
 fn mux_run_completed(interrupted: bool, finalize_failed: bool, halt_cancelled: bool) -> bool {
@@ -704,11 +735,14 @@ fn drive_mux(
     // every `POLL_INTERVAL`, so Ctrl-C / `/api/stop` stays responsive.
     let deadline = send_deadline;
     let mut interrupted = false;
+    // A send refused with no halt and a healthy consumer ran out its deadline.
+    let mut send_timed_out = false;
 
     // Buffered header frames first, in order.
     for frame in buffered {
         if pipe.send_with_halt(frame, halt, deadline).is_err() {
             interrupted = true;
+            send_timed_out = !halt.is_cancelled() && !pipe.consumer_failed();
             break;
         }
         // Feed write-side progress during the drain exactly as the steady-state
@@ -732,6 +766,7 @@ fn drive_mux(
                     }
                     if pipe.send_with_halt(frame, halt, deadline).is_err() {
                         interrupted = true;
+                        send_timed_out = !halt.is_cancelled() && !pipe.consumer_failed();
                         break;
                     }
                     events.on_write_progress(bytes.load(Ordering::Relaxed), total_bytes);
@@ -785,7 +820,7 @@ fn drive_mux(
     // the container. On halt/wedge this returns an error variant, translated
     // to `completed = false` rather than a hard failure.
     let (bytes_written, undelivered_streams, finalize_failed) =
-        match pipe.finish_with_halt(Some(halt)) {
+        match finish_pumped(pipe, halt, send_timed_out) {
             Ok((b, undelivered)) => (b, undelivered, false),
             Err(Error::Halted | Error::PipelineJoinTimeout) => {
                 (bytes.load(Ordering::Relaxed), Vec::new(), true)
@@ -796,7 +831,8 @@ fn drive_mux(
         tracing::warn!(
             target: "mux",
             streams = ?undelivered_streams,
-            "output sink could not deliver every planned stream; the file does not match              the pre-mux plan (surfaced as MuxOutcome::undelivered_streams)"
+            "output sink could not deliver every planned stream; the file does not match \
+             the pre-mux plan (surfaced as MuxOutcome::undelivered_streams)"
         );
     }
 
@@ -2715,4 +2751,136 @@ mod tests {
             "a non-DVD passes decrypt_keys() through unchanged"
         );
     }
+
+    // A consumer wedged past the send deadline must not hold the finish for the
+    // full JOIN_TIMEOUT (600 s).
+    #[test]
+    fn a_send_timeout_bounds_the_final_join() {
+        // Only the first write wedges, returning after finish has given up; the
+        // leaked consumer must then neither apply the queued item nor close.
+        struct Stuck {
+            closed: Arc<std::sync::atomic::AtomicBool>,
+            woke: std::sync::mpsc::Sender<()>,
+            first: bool,
+            applied: Arc<std::sync::atomic::AtomicUsize>,
+        }
+        impl Sink<u32> for Stuck {
+            type Output = ();
+            fn apply(&mut self, _: u32) -> Result<Flow, Error> {
+                self.applied.fetch_add(1, Ordering::SeqCst);
+                if std::mem::take(&mut self.first) {
+                    std::thread::sleep(Duration::from_secs(7));
+                    let _ = self.woke.send(());
+                }
+                Ok(Flow::Continue)
+            }
+            fn close(self) -> Result<(), Error> {
+                self.closed.store(true, Ordering::SeqCst);
+                Ok(())
+            }
+        }
+        let closed = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let applied = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let (woke_tx, woke) = std::sync::mpsc::channel();
+        let stuck = Stuck {
+            closed: closed.clone(),
+            woke: woke_tx,
+            first: true,
+            applied: applied.clone(),
+        };
+        let pipe = Pipeline::spawn(1, stuck).unwrap();
+        let halt = Halt::new();
+        let mut timed_out = false;
+        for i in 0..4 {
+            if pipe
+                .send_with_halt(i, &halt, Duration::from_millis(50))
+                .is_err()
+            {
+                timed_out = true;
+                break;
+            }
+        }
+        assert!(timed_out, "the stuck consumer trips the send deadline");
+        let (tx, rx) = std::sync::mpsc::channel();
+        std::thread::spawn(move || {
+            let _ = tx.send(finish_pumped(pipe, &halt, true).is_err());
+        });
+        let failed = rx
+            .recv_timeout(Duration::from_secs(30))
+            .expect("the join must be bounded by the grace, not JOIN_TIMEOUT");
+        assert!(failed);
+        woke.recv_timeout(Duration::from_secs(30))
+            .expect("the wedged write returns");
+        std::thread::sleep(Duration::from_millis(500));
+        assert!(
+            !closed.load(Ordering::SeqCst),
+            "an abandoned consumer must not finalise the output late"
+        );
+        assert_eq!(
+            applied.load(Ordering::SeqCst),
+            1,
+            "nor apply the queued item"
+        );
+    }
+
+    // An Opus-style CodecDelay/SeekPreRoll on an mkv:// source must reach the
+    // mkv:// output through the driver (the sink's set_track_timing wiring).
+    #[test]
+    fn track_timing_propagates_from_source_to_mkv_output() {
+        use crate::disc::{Codec, ColorSpace, FrameRate, HdrFormat, Resolution, VideoStream};
+        use crate::mux::mkvstream::MkvStream;
+        let dir = tempfile::tempdir().expect("tempdir");
+        let src = dir.path().join("src.mkv");
+        let dst = dir.path().join("dst.mkv");
+        let mut title = DiscTitle {
+            streams: vec![crate::disc::Stream::Video(VideoStream {
+                pid: 0x1011,
+                codec: Codec::H264,
+                resolution: Resolution::R1080p,
+                frame_rate: FrameRate::F24,
+                hdr: HdrFormat::Sdr,
+                color_space: ColorSpace::Bt709,
+                display_aspect: None,
+                secondary: false,
+                label: String::new(),
+                measured_cicp: None,
+            })],
+            ..DiscTitle::empty()
+        };
+        title.codec_privates = vec![Some(vec![0x01, 0x64, 0x00, 0x1F, 0xFF, 0xE1])];
+        let timing = crate::pes::TrackTiming {
+            codec_delay_ns: 6_500_000,
+            seek_preroll_ns: 80_000_000,
+        };
+        let mut w = MkvStream::create(Box::new(std::fs::File::create(&src).unwrap()), &title, None)
+            .unwrap();
+        w.set_track_timing(0, timing).unwrap();
+        w.write(&PesFrame {
+            discard_padding_ns: 0,
+            coding: None,
+            source: None,
+            track: 0,
+            pts: 0,
+            keyframe: true,
+            data: vec![0xA1; 16],
+            duration_ns: None,
+        })
+        .unwrap();
+        w.finish().unwrap();
+
+        let stream = Box::new(MkvStream::open(std::fs::File::open(&src).unwrap()).unwrap());
+        let out = drive_mux(
+            stream,
+            &format!("mkv://{}", dst.display()),
+            &Halt::new(),
+            &SpyEvents::new(),
+            None,
+            Duration::from_secs(60),
+            None,
+        )
+        .expect("mkv remux runs");
+        assert!(out.completed);
+        let back = MkvStream::open(std::fs::File::open(&dst).unwrap()).unwrap();
+        assert_eq!(back.track_timing(0), timing);
+    }
 }

src/mux/ebml.rs

@@ -563,6 +563,11 @@ pub const BLOCK_ADD_ID: u32 = 0xEE;
 /// (a keyframe Block in a BlockGroup carries none). Written for non-keyframe
 /// video frames that must live in a BlockGroup to carry an MVC BlockAdditional.
 pub const REFERENCE_BLOCK: u32 = 0xFB;
+/// DiscardPadding (RFC 9559 §5.1.3.5.7): signed ns of decoded audio to drop.
+pub const DISCARD_PADDING: u32 = 0x75A2;
+/// CodecDelay / SeekPreRoll (RFC 9559 §5.1.4.1.18/19), TrackEntry children, ns.
+pub const CODEC_DELAY: u32 = 0x56AA;
+pub const SEEK_PRE_ROLL: u32 = 0x56BB;
 
 // Audio
 pub const AUDIO: u32 = 0xE1;
@@ -629,6 +634,8 @@ pub const CODEC_EAC3: &str = "A_EAC3";
 pub const CODEC_TRUEHD: &str = "A_TRUEHD";
 pub const CODEC_DTS: &str = "A_DTS";
 pub const CODEC_PCM_BE: &str = "A_PCM/INT/BIG";
+/// Little-endian integer PCM (read side only; converted to big-endian).
+pub const CODEC_PCM_LE: &str = "A_PCM/INT/LIT";
 /// AAC. The generic registered ID; the AudioSpecificConfig travels in
 /// CodecPrivate, so no profile suffix is needed (and the `A_AAC/MPEG4/*`
 /// suffixed forms are legacy).

src/mux/fvi_sink.rs

@@ -4,7 +4,7 @@
 //! *video-index* record per coded picture of the title's primary video
 //! track, instead of muxing frames into a container.
 //!
-//! On-disk shape: the freemkv FVI format  — JSON Lines, a header object on line 1, then one
+//! On-disk shape: the freemkv FVI format — JSON Lines, a header object on line 1, then one
 //! record per picture. The sink is purely additive: it does NOT touch the MKV mux path.
 
 use crate::disc::{DiscTitle, Stream as DiscStream};
@@ -17,7 +17,7 @@ use std::fs::File;
 use std::io::{self, BufWriter, Write};
 use std::path::Path;
 
-/// Write the FVI header row  into `w`.
+/// Write the FVI header row into `w`.
 fn write_fvi_header(w: &mut dyn Write, h: &MapHeader) -> io::Result<()> {
     let mut source = serde_json::json!({
         "medium": h.source.medium.as_str(),

src/mux/interop_tests.rs

@@ -66,6 +66,30 @@ fn generate(
     vfr: bool,
     bframes: bool,
     multi_audio: bool,
+) -> PathBuf {
+    generate_with(
+        dir,
+        video,
+        audio,
+        transport,
+        vfr,
+        bframes,
+        multi_audio,
+        false,
+    )
+}
+
+// `video_late`: video starts 0.3 s in, so the 0.1 s audio precedes the first IDR.
+#[allow(clippy::too_many_arguments)]
+fn generate_with(
+    dir: &Path,
+    video: &str,
+    audio: Option<&str>,
+    transport: bool,
+    vfr: bool,
+    bframes: bool,
+    multi_audio: bool,
+    video_late: bool,
 ) -> PathBuf {
     let path = dir.join(if transport { "input.m2ts" } else { "input.mkv" });
     let mut command = Command::new("ffmpeg");
@@ -145,6 +169,9 @@ fn generate(
             "vfr",
         ]);
     }
+    if video_late {
+        command.args(["-vf", "setpts=PTS+0.3/TB"]);
+    }
     if transport {
         command.args(["-f", "mpegts", "-mpegts_m2ts_mode", "0"]);
     }
@@ -477,8 +504,7 @@ fn ffmpeg_mkv_roundtrip_preserves_vfr_and_multiple_audio_tracks() {
 }
 
 // Discovered while building the gate; retain executable, strict reproducers.
-// These deliberately do NOT have the ffmpeg_ prefix: run ffmpeg_ explicitly
+// They carry the ffmpeg_ prefix, so the QA gate runs them with the rest.
-// to investigate them. Promote each into the QA gate when its fix lands.
 
 #[test]
 #[ignore = "requires ffmpeg and ffprobe"]
@@ -539,3 +565,22 @@ fn ffmpeg_generic_transport_audio_discovery() {
         validate(dir.path(), &input, &output, 2, false);
     }
 }
+
+#[test]
+#[ignore = "requires ffmpeg and ffprobe"]
+fn ffmpeg_audio_before_the_first_video_frame_keeps_its_offset() {
+    let dir = fixture_dir("audio-first");
+    let input = generate_with(
+        dir.path(),
+        "libx264",
+        Some("ac3"),
+        true,
+        false,
+        false,
+        false,
+        true,
+    );
+    let output = dir.path().join("output.mkv");
+    remux(&input, &output, true);
+    validate(dir.path(), &input, &output, 2, false);
+}

src/mux/m2ts_mux/mod.rs

@@ -119,6 +119,8 @@ pub struct M2tsMux<W: Write> {
     /// onto the very first video PES so a receiver tuning at stream start
     /// has a clock reference (the PMT advertises the video PID as PCR_PID).
     first_video_written: bool,
+    /// Last PCR stamped (90 kHz); the PCR is held monotonic against it.
+    last_pcr_90k: u64,
 }
 
 impl<W: Write> M2tsMux<W> {
@@ -140,6 +142,7 @@ impl<W: Write> M2tsMux<W> {
             packets_written: 0,
             video_packets_since_pcr: 0,
             first_video_written: false,
+            last_pcr_90k: 0,
         }
     }
 
@@ -173,11 +176,16 @@ impl<W: Write> M2tsMux<W> {
     /// drives the random_access_indicator bit on the first packet of this
     /// PES (and gates codec_private NAL prepending — those only attach to
     /// the first keyframe).
+    ///
+    /// Frames carry no DTS, so the PCR leads the PTS by `PCR_LEAD_90KHZ` and is
+    /// held monotonic; reordering deeper than that lead can still let it pass a
+    /// B-frame's PTS.
     pub fn write_video(&mut self, pts_ns: i64, keyframe: bool, data: &[u8]) -> io::Result<()> {
         let pts_90k = self.base_relative_pts(pts_ns, /* may_seed_base */ true);
-        // PCR comes "before" the PTS it timestamps; clamp at 0 for the
+        let pcr = pts_90k
-        // first frame so we don't underflow.
+            .saturating_sub(PCR_LEAD_90KHZ)
-        let pcr = pts_90k.saturating_sub(PCR_LEAD_90KHZ);
+            .max(self.last_pcr_90k);
+        self.last_pcr_90k = pcr;
 
         // Annex-B-ify the frame and prepend VPS/SPS/PPS once, on the
         // FIRST keyframe (not first frame — non-key frames before the
@@ -256,7 +264,11 @@ impl<W: Write> M2tsMux<W> {
             0
         };
         if may_seed_base {
-            self.base_pts_90k.get_or_insert(raw_90k);
+            // Seeded one headroom early (see tsmux) so a slightly earlier frame
+            // arriving second keeps its offset.
+            let headroom = (super::tsmux::ORIGIN_HEADROOM_NS as u64) * 9 / 100_000;
+            self.base_pts_90k
+                .get_or_insert(raw_90k.wrapping_sub(headroom) & 0x1_FFFF_FFFF);
         }
         let base = self.base_pts_90k.unwrap_or(raw_90k);
         // Modular 33-bit subtraction: the PTS clock wraps every 2^33 ticks
@@ -569,6 +581,9 @@ fn mpegts_crc32(data: &[u8]) -> u32 {
 mod tests {
     use super::*;
 
+    // The origin headroom in 90 kHz ticks: the seeding frame's encoded PTS.
+    const H: u64 = super::super::tsmux::ORIGIN_HEADROOM_NS as u64 * 9 / 100_000;
+
     // Golden vector for an audio PES header (ISO/IEC 13818-1 §2.4.3.7).
     #[test]
     fn audio_pes_header_is_byte_exact_and_bounded_unlike_video() {
@@ -1280,9 +1295,7 @@ mod tests {
             let mut frame = Vec::new();
             frame.extend_from_slice(&4u32.to_be_bytes());
             frame.extend_from_slice(&[0x40, 0x01, 0x0C, 0x01]);
-            // 1s → 90000 ticks; base is this same frame, so relative PTS=0
+            // A single frame: it seeds the origin one headroom below itself.
-            // and PCR clamps to 0. Use a single frame: PTS rebases to 0,
-            // so PCR = 0.saturating_sub(lead) = 0.
             mux.write_video(1_000_000_000, true, &frame).unwrap();
             mux.finish().unwrap();
         }
@@ -1295,16 +1308,16 @@ mod tests {
             | ((af[3] as u64) << 9)
             | ((af[4] as u64) << 1)
             | ((af[5] as u64 >> 7) & 0x01);
-        // Single frame rebases its own PTS to 0; PCR = 0 - lead clamped to 0.
+        // The seeding frame sits one headroom past the origin; PCR leads it.
-        assert_eq!(base, 0, "first frame PCR clamps to 0 (no underflow)");
+        assert_eq!(base, H - PCR_LEAD_90KHZ, "first frame PCR leads its PTS");
     }
 
     // ── base_relative_pts overflow / saturation ───────────────────────────
 
     #[test]
     fn extreme_pts_does_not_overflow_and_clamps_to_33bit() {
-        // Two frames are required: a single frame always rebases to PTS 0, making
+        // Two frames are required: a single frame always encodes the headroom, making
-        // the only assertion the vacuous `0 < 2^33`. Bit 32 is unreachable (the
+        // the only assertion vacuous. Bit 32 is unreachable (the
         // signed-33-bit delta rule floors the upper half to 0), so this pins encoding.
         let decode_pts = |pkt: &[u8]| -> u64 {
             // Payload after the AF: AF area = 1 (length byte) + af_len.
@@ -1328,7 +1341,8 @@ mod tests {
         {
             let mut mux = M2tsMux::new(&mut sink);
             mux.write_video(0, true, &frame).unwrap();
-            mux.write_video(477_218_477 * 100_000, true, &frame)
+            // Less one headroom (10_000 x 100 us), which the origin already adds.
+            mux.write_video((477_218_477 - 10_000) * 100_000, true, &frame)
                 .unwrap();
             mux.finish().unwrap();
         }
@@ -1398,7 +1412,7 @@ mod tests {
     #[test]
     fn negative_pts_ns_encodes_zero() {
         // base_relative_pts treats pts_ns <= 0 as raw 0. A negative input
-        // must encode PTS 0, not a wrapped value.
+        // must encode as t=0 (the headroom), not a wrapped value.
         let mut sink: Vec<u8> = Vec::new();
         {
             let mut mux = M2tsMux::new(&mut sink);
@@ -1416,7 +1430,7 @@ mod tests {
             | (((pes[11] >> 1) as u64) << 15)
             | ((pes[12] as u64) << 7)
             | ((pes[13] >> 1) as u64);
-        assert_eq!(pts, 0, "negative pts_ns encodes PTS 0");
+        assert_eq!(pts, H, "negative pts_ns is treated as 0, the seeding frame");
     }
 
     #[test]
@@ -1454,12 +1468,12 @@ mod tests {
         assert_eq!(pkts.len(), 2, "one PUSI packet per small audio frame");
         assert_eq!(
             decode_audio_pts(pkts[0]),
-            0,
+            H,
-            "the first (audio) frame seeds the origin → 0"
+            "the first (audio) frame seeds the origin one headroom below it"
         );
         assert_eq!(
             decode_audio_pts(pkts[1]),
-            90_000,
+            H + 90_000,
             "the second audio frame (1s later) must keep its spacing, not collapse to 0"
         );
     }
@@ -1518,4 +1532,49 @@ mod tests {
             "first packet is PAT on PID 0"
         );
     }
+
+    // Same for the 188-byte muxer: a frame shortly before the seeding one keeps
+    // its offset instead of flooring to the origin.
+    #[test]
+    fn a_frame_shortly_before_the_seeding_frame_keeps_its_offset() {
+        let mut sink: Vec<u8> = Vec::new();
+        let mut mux = M2tsMux::new(&mut sink);
+        let audio = mux.base_relative_pts(100_000_000, true);
+        let video = mux.base_relative_pts(50_000_000, true);
+        assert_eq!(audio - video, 4_500);
+    }
+
+    fn all_pcrs(buf: &[u8]) -> Vec<u64> {
+        find_all_pkts(buf, PID_VIDEO, false)
+            .into_iter()
+            .filter_map(af_body)
+            .filter(|af| af.first().is_some_and(|f| f & 0x10 != 0))
+            .map(|af| {
+                ((af[1] as u64) << 25)
+                    | ((af[2] as u64) << 17)
+                    | ((af[3] as u64) << 9)
+                    | ((af[4] as u64) << 1)
+                    | ((af[5] as u64 >> 7) & 0x01)
+            })
+            .collect()
+    }
+
+    // B-frames arrive with PTS below an earlier P-frame's: the PCR must still
+    // never step backwards (ISO/IEC 13818-1 §2.4.2.2).
+    #[test]
+    fn pcr_never_goes_backwards_with_b_frames() {
+        let mut frame = Vec::new();
+        frame.extend_from_slice(&8000u32.to_be_bytes());
+        frame.extend(std::iter::repeat_n(0x41u8, 8000));
+        let mut sink: Vec<u8> = Vec::new();
+        {
+            let mut mux = M2tsMux::new(&mut sink);
+            for (i, ms) in [80i64, 200, 120, 160].into_iter().enumerate() {
+                mux.write_video(ms * 1_000_000, i == 0, &frame).unwrap();
+            }
+        }
+        let pcrs = all_pcrs(&sink);
+        assert!(pcrs.len() >= 4, "{pcrs:?}");
+        assert!(pcrs.windows(2).all(|w| w[1] >= w[0]), "{pcrs:?}");
+    }
 }

src/mux/meta.rs

@@ -26,8 +26,10 @@ const MAGIC: [u8; 8] = [b'F', b'M', b'K', b'V', 0x00, 0x01, 0x00, 0x00];
 
 /// Highest header format version this build understands. A header tagged with
 /// a newer version is rejected so older readers cleanly refuse incompatible
-/// formats instead of silently mis-parsing them as v1.
+/// formats instead of silently mis-parsing them as v1. Version 2 = every PES
-const SUPPORTED_VERSION: u8 = 1;
+/// frame after the header carries an 8-byte DiscardPadding extension; it is
+/// written only when a track has decoder timing, so other streams stay v1.
+const SUPPORTED_VERSION: u8 = 2;
 
 /// Index of the version byte within [`MAGIC`].
 const VERSION_BYTE: usize = 5;
@@ -47,6 +49,24 @@ pub struct M2tsMeta {
     pub duration: f64,
     /// Stream descriptors.
     pub streams: Vec<MetaStream>,
+    /// Per-track decoder timing (Opus CodecDelay/SeekPreRoll); non-default only.
+    /// Readers that predate it ignore the field.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub timings: Vec<MetaTiming>,
+    /// Frames on the wire carry the DiscardPadding extension (header v2). Taken
+    /// from the version byte, not the JSON.
+    #[serde(skip)]
+    pub frame_padding: bool,
+}
+
+/// Decoder timing for one stream index, carried in the FMKV header.
+#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
+pub struct MetaTiming {
+    pub track: usize,
+    #[serde(default)]
+    pub codec_delay_ns: u64,
+    #[serde(default)]
+    pub seek_preroll_ns: u64,
 }
 
 /// A single stream descriptor in the metadata.
@@ -165,9 +185,40 @@ impl M2tsMeta {
             title: title.playlist.clone(),
             duration: title.duration_secs,
             streams,
+            timings: Vec::new(),
+            frame_padding: false,
         }
     }
 
+    /// Attach per-track decoder timing; any non-default timing switches the
+    /// stream to header v2 so DiscardPadding travels with each frame.
+    pub fn with_timings(mut self, timings: &[crate::pes::TrackTiming]) -> Self {
+        self.timings = timings
+            .iter()
+            .enumerate()
+            .filter(|(_, t)| **t != crate::pes::TrackTiming::default())
+            .map(|(track, t)| MetaTiming {
+                track,
+                codec_delay_ns: t.codec_delay_ns,
+                seek_preroll_ns: t.seek_preroll_ns,
+            })
+            .collect();
+        self.frame_padding = !self.timings.is_empty();
+        self
+    }
+
+    /// Decoder timing of stream `track` (default when the header has none).
+    pub fn timing(&self, track: usize) -> crate::pes::TrackTiming {
+        self.timings
+            .iter()
+            .find(|t| t.track == track)
+            .map(|t| crate::pes::TrackTiming {
+                codec_delay_ns: t.codec_delay_ns,
+                seek_preroll_ns: t.seek_preroll_ns,
+            })
+            .unwrap_or_default()
+    }
+
     /// Convert back to a library Title (for remux).
     pub fn to_title(&self) -> DiscTitle {
         let streams = self
@@ -310,7 +361,11 @@ pub fn write_header(w: &mut impl Write, meta: &M2tsMeta) -> io::Result<()> {
     let padded_len = raw_len.div_ceil(BD_SOURCE_PACKET_BYTES) * BD_SOURCE_PACKET_BYTES;
     let padding = padded_len - raw_len;
 
-    w.write_all(&MAGIC)?;
+    let mut magic = MAGIC;
+    if meta.frame_padding {
+        magic[VERSION_BYTE] = 2;
+    }
+    w.write_all(&magic)?;
     w.write_all(&json_len.to_be_bytes())?;
     w.write_all(&json)?;
     if padding > 0 {
@@ -367,8 +422,9 @@ pub fn read_header(r: &mut impl Read) -> io::Result<Option<M2tsMeta>> {
     let mut json_buf = vec![0u8; json_len];
     r.read_exact(&mut json_buf)?;
 
-    let meta: M2tsMeta =
+    let mut meta: M2tsMeta =
         serde_json::from_slice(&json_buf).map_err(|_| crate::error::Error::NoMetadata)?;
+    meta.frame_padding = magic[VERSION_BYTE] >= 2;
 
     // Skip padding to next 192-byte boundary (at most BD_SOURCE_PACKET_BYTES-1 bytes →
     // a stack buffer, no heap allocation).
@@ -625,7 +681,7 @@ mod tests {
         // The version byte lives at index 5. A regression that shifted the
         // version byte would make every header read the wrong version.
         assert_eq!(&MAGIC[0..4], b"FMKV");
-        assert_eq!(MAGIC[VERSION_BYTE], SUPPORTED_VERSION);
+        assert_eq!(MAGIC[VERSION_BYTE], 1, "the base (v1) magic");
         assert_eq!(VERSION_BYTE, 5);
         assert_eq!(MAGIC.len(), 8);
     }
@@ -813,4 +869,23 @@ mod tests {
         assert_eq!(back.playlist, "The Movie");
         assert_eq!(back.duration_secs, 7384.5);
     }
+
+    // Timing switches the header to v2 (older readers refuse it cleanly); no
+    // timing keeps the v1 wire byte-identical.
+    #[test]
+    fn timing_round_trips_and_selects_the_header_version() {
+        let t = video_title(HdrFormat::Sdr, ColorSpace::Bt709);
+        let timing = crate::pes::TrackTiming {
+            codec_delay_ns: 1,
+            seek_preroll_ns: 2,
+        };
+        for (timings, version) in [(vec![], 1u8), (vec![timing], 2)] {
+            let mut buf = Vec::new();
+            write_header(&mut buf, &M2tsMeta::from_title(&t).with_timings(&timings)).unwrap();
+            assert_eq!(buf[VERSION_BYTE], version);
+            let back = read_header(&mut io::Cursor::new(&buf)).unwrap().unwrap();
+            assert_eq!(back.frame_padding, version == 2);
+            assert_eq!(back.timing(0), timings.first().copied().unwrap_or_default());
+        }
+    }
 }

src/mux/meta_sink.rs

@@ -144,6 +144,18 @@ pub(crate) fn title_json(title: &DiscTitle) -> serde_json::Value {
     doc
 }
 
+// Every rate a stream carries, in Hz: a BD combo (48/96, 48/192) lists both,
+// Unknown is empty.
+fn sample_rates(rate: crate::disc::SampleRate) -> Vec<u32> {
+    use crate::disc::SampleRate::*;
+    match rate {
+        S48_96 => vec![48_000, 96_000],
+        S48_192 => vec![48_000, 192_000],
+        Unknown => Vec::new(),
+        r => vec![r.hz() as u32],
+    }
+}
+
 /// Compact id for an audio stream's editorial purpose (no localized prose —
 /// the app maps it to display text). Mirrors the `Codec::id()` convention.
 fn purpose_id(p: crate::disc::LabelPurpose) -> &'static str {
@@ -157,6 +169,22 @@ fn purpose_id(p: crate::disc::LabelPurpose) -> &'static str {
     }
 }
 
+// `doc[key][i]` as a mutable object, or `dummy` (reset to an object) if absent.
+fn entry<'a>(
+    doc: &'a mut serde_json::Value,
+    key: &str,
+    i: usize,
+    dummy: &'a mut serde_json::Value,
+) -> &'a mut serde_json::Value {
+    match doc.get_mut(key).and_then(|v| v.get_mut(i)) {
+        Some(e) if e.is_object() => e,
+        _ => {
+            *dummy = serde_json::json!({});
+            dummy
+        }
+    }
+}
+
 /// Merge the per-stream fields `TitleProfile` drops back into the `json://`
 /// document, straight from the source Title. Walks `title.streams` in declared
 /// order with a per-kind cursor so each raw stream lines up with the profile's
@@ -165,10 +193,12 @@ fn enrich_streams(doc: &mut serde_json::Value, title: &DiscTitle) {
     use crate::disc::Stream;
     use serde_json::json;
     let (mut vi, mut ai, mut si) = (0usize, 0usize, 0usize);
+    // `doc[key][i]` panics on a missing array; skip an entry the profile lacks.
+    let mut dummy = serde_json::Value::Null;
     for s in &title.streams {
         match s {
             Stream::Video(v) => {
-                let e = &mut doc["video"][vi];
+                let e = entry(doc, "video", vi, &mut dummy);
                 e["pid"] = json!(v.pid);
                 e["color_space"] = json!(v.color_space.id());
                 e["display_aspect"] = match v.display_aspect {
@@ -199,15 +229,18 @@ fn enrich_streams(doc: &mut serde_json::Value, title: &DiscTitle) {
                 vi += 1;
             }
             Stream::Audio(a) => {
-                let e = &mut doc["audio"][ai];
+                let e = entry(doc, "audio", ai, &mut dummy);
                 e["pid"] = json!(a.pid);
                 e["secondary"] = json!(a.secondary);
-                e["sample_rate"] = json!(a.sample_rate.to_string());
+                // Numeric Hz (null when unknown), not the "48kHz" display form.
+                let hz = a.sample_rate.hz() as u32;
+                e["sample_rate"] = if hz == 0 { json!(null) } else { json!(hz) };
+                e["sample_rates"] = json!(sample_rates(a.sample_rate));
                 e["purpose"] = json!(purpose_id(a.purpose));
                 ai += 1;
             }
             Stream::Subtitle(sub) => {
-                let e = &mut doc["subtitles"][si];
+                let e = entry(doc, "subtitles", si, &mut dummy);
                 e["pid"] = json!(sub.pid);
                 e["descriptive_service"] =
                     json!(sub.qualifier == crate::disc::LabelQualifier::DescriptiveService);
@@ -446,7 +479,8 @@ mod tests {
         let a = &v["audio"][0];
         assert_eq!(a["pid"], 0x1100);
         assert_eq!(a["secondary"], true);
-        assert_eq!(a["sample_rate"], "48kHz");
+        assert_eq!(a["sample_rate"], 48_000);
+        assert_eq!(a["sample_rates"], serde_json::json!([48_000]));
         assert_eq!(a["purpose"], "commentary");
 
         let s = &v["subtitles"][0];
@@ -497,4 +531,34 @@ mod tests {
         );
         let _ = std::fs::remove_file(&jpath);
     }
+
+    // A document missing the per-kind arrays must not panic the enrichment.
+    #[test]
+    fn enriching_a_document_without_stream_arrays_does_not_panic() {
+        let mut t = DiscTitle::empty();
+        t.streams = vec![crate::disc::Stream::Audio(crate::disc::AudioStream {
+            pid: 0x1100,
+            codec: crate::disc::Codec::Ac3,
+            channels: crate::disc::AudioChannels::Stereo,
+            language: "eng".into(),
+            sample_rate: crate::disc::SampleRate::Unknown,
+            secondary: false,
+            purpose: crate::disc::LabelPurpose::Normal,
+            label: String::new(),
+        })];
+        let mut doc = serde_json::json!({});
+        enrich_streams(&mut doc, &t);
+        assert_eq!(
+            title_json(&t)["audio"][0]["sample_rate"],
+            serde_json::Value::Null
+        );
+    }
+
+    #[test]
+    fn combo_sample_rates_keep_both_rates() {
+        use crate::disc::SampleRate;
+        assert_eq!(sample_rates(SampleRate::S48_96), vec![48_000, 96_000]);
+        assert_eq!(sample_rates(SampleRate::S48_192), vec![48_000, 192_000]);
+        assert!(sample_rates(SampleRate::Unknown).is_empty());
+    }
 }

src/mux/mkv.rs

@@ -548,6 +548,9 @@ pub struct MkvMuxer<W: Write + Seek> {
     /// track type (mirroring `track_is_video`) so the invariant holds for every
     /// subtitle track — including a sparse forced-narrative second track.
     track_is_subtitle: Vec<bool>,
+    /// Per-track-index flag: true for an `S_VOBSUB` track, whose SPU carries its
+    /// own display duration.
+    track_is_vobsub: Vec<bool>,
     /// Index of the PRIMARY video track — the first track whose type is video.
     /// This (not the literal index 0) is the clip-boundary epoch driver: the
     /// M2TS/PMT path orders streams by PMT declaration order and may list an
@@ -570,6 +573,11 @@ pub struct MkvMuxer<W: Write + Seek> {
     /// `finish()` via `frame_count == 0`, not this counter; a PARTIAL drop is normal and is
     /// only logged, not an error — `finish()` is this field's only reader, so keep that log.
     dropped_pre_cluster: u64,
+    /// Ticks the timeline origin sits before the first video keyframe.
+    origin_lead_ticks: i64,
+    /// Per VobSub track: file offset of an open-ended block's 3-byte
+    /// BlockDuration value, and that block's timestamp in ticks.
+    vobsub_open_end: std::collections::HashMap<usize, (u64, i64)>,
     seek_fixups: Vec<SeekPositionFixup>,
     /// Absolute file offset of the CUES SeekHead entry (a fixed 21-byte Seek
     /// element). When `finish()` writes no Cues element (zero cue points), this
@@ -705,10 +713,48 @@ struct InterlacedRewrite {
 // AUs.
 const TIMESTAMP_SCALE_NS: i64 = 100_000;
 
-// Last-resort BlockDuration (1 s) for a SUBTITLE frame that reaches the muxer with
+// BlockDuration for a durationless PGS block (issue #52): a PGS cue ends at the
-// no duration. Parsers (PGS) already synthesize one; this only fires if another
+// next display set, so a long bound only caps a cue that is already replaced.
-// subtitle path forgets, keeping every subtitle block durated (issue #52).
+const PGS_FALLBACK_BLOCK_DURATION_NS: i64 = 30_000_000_000;
-const MIN_SUBTITLE_BLOCK_DURATION_NS: i64 = 1_000_000_000;
+
+// A VobSub SPU with no stop command ends at the next SPU on its track (patched
+// in place), or after this cap. 100_000 ticks keeps the field 3 bytes wide.
+const VOBSUB_OPEN_END_TICKS: u64 = 100_000;
+
+// VobSub SP_DCSQ delays count 1024 ticks of the 90 kHz clock.
+const SPU_DELAY_NS: u64 = 1024 * 1_000_000_000 / 90_000;
+
+// Display time of a VobSub SPU: the delay of its first STP_DSP (0x02) command,
+// walking the SP_DCSQ chain. `None` when the SPU has no stop or is malformed.
+fn vobsub_display_ns(spu: &[u8]) -> Option<u64> {
+    let be16 = |at: usize| -> Option<usize> {
+        Some(u16::from_be_bytes([*spu.get(at)?, *spu.get(at + 1)?]) as usize)
+    };
+    let mut dcsq = be16(2)?;
+    // Each DCSQ is at least 5 bytes, so a well-formed chain has bounded length.
+    for _ in 0..spu.len() / 5 + 1 {
+        let delay = be16(dcsq)? as u64;
+        let next = be16(dcsq + 2)?;
+        let mut at = dcsq + 4;
+        loop {
+            let args = match *spu.get(at)? {
+                0x02 => return Some(delay * SPU_DELAY_NS),
+                0x00 | 0x01 => 0,
+                0x03 | 0x04 => 2,
+                0x05 => 6,
+                0x06 => 4,
+                0x07 => be16(at + 1)?,
+                _ => break,
+            };
+            at += 1 + args;
+        }
+        if next == dcsq {
+            return None;
+        }
+        dcsq = next;
+    }
+    None
+}
 
 // Nominal new-cluster interval (2 s) in TimestampScale ticks; a keyframe only opens a new
 // cluster once this much has elapsed.
@@ -898,10 +944,10 @@ impl<W: Write + Seek> MkvMuxer<W> {
             ebml::write_uint(&mut writer, ebml::TRACK_TYPE, track.track_type)?;
             if let Some(timing) = timings.get(i) {
                 if timing.codec_delay_ns > 0 {
-                    ebml::write_uint(&mut writer, 0x56AA, timing.codec_delay_ns)?;
+                    ebml::write_uint(&mut writer, ebml::CODEC_DELAY, timing.codec_delay_ns)?;
                 }
                 if timing.seek_preroll_ns > 0 {
-                    ebml::write_uint(&mut writer, 0x56BB, timing.seek_preroll_ns)?;
+                    ebml::write_uint(&mut writer, ebml::SEEK_PRE_ROLL, timing.seek_preroll_ns)?;
                 }
             }
 
@@ -1167,6 +1213,10 @@ impl<W: Write + Seek> MkvMuxer<W> {
                 .iter()
                 .map(|t| t.track_type == ebml::TRACK_TYPE_SUBTITLE)
                 .collect(),
+            track_is_vobsub: tracks
+                .iter()
+                .map(|t| t.codec_id == ebml::CODEC_VOBSUB)
+                .collect(),
             primary_video_track: tracks
                 .iter()
                 .position(|t| t.track_type == ebml::TRACK_TYPE_VIDEO),
@@ -1175,6 +1225,8 @@ impl<W: Write + Seek> MkvMuxer<W> {
             cues: Vec::new(),
             frame_count: 0,
             dropped_pre_cluster: 0,
+            origin_lead_ticks: 0,
+            vobsub_open_end: std::collections::HashMap::new(),
             seek_fixups,
             cues_seek_entry_pos,
             info_offset,
@@ -1225,6 +1277,22 @@ impl<W: Write + Seek> MkvMuxer<W> {
         Ok(true)
     }
 
+    // Overwrite a 3-byte big-endian value at `pos`, then return to the end.
+    fn patch_u24(&mut self, pos: u64, val: u64) -> io::Result<()> {
+        let end = self.writer.stream_position()?;
+        self.writer.seek(io::SeekFrom::Start(pos))?;
+        self.writer.write_all(&val.to_be_bytes()[5..])?;
+        self.writer.seek(io::SeekFrom::Start(end))?;
+        Ok(())
+    }
+
+    /// Put the timeline origin `lead_ns` before the first video keyframe (the
+    /// earliest audio/video frame), and count `dropped` frames that lay before it.
+    pub(crate) fn set_origin_lead_ns(&mut self, lead_ns: i64, dropped: u64) {
+        self.origin_lead_ticks = lead_ns.max(0) / TIMESTAMP_SCALE_NS;
+        self.dropped_pre_cluster += dropped;
+    }
+
     /// Attach an opening-frame capture (`--log-level 3`). The capture writes the
     /// first ~100 coded frames per track to `<output>.opening.bin` and logs a
     /// per-frame summary, so opening-GOP / menu issues are diagnosable from a
@@ -1379,8 +1447,9 @@ impl<W: Write + Seek> MkvMuxer<W> {
                     self.dropped_pre_cluster += 1;
                     return Ok(());
                 }
-                self.base_pts_ticks = Some(raw_ticks);
+                let base = raw_ticks - self.origin_lead_ticks;
-                raw_ticks
+                self.base_pts_ticks = Some(base);
+                base
             }
         };
         // Floor at 0: a frame earlier than base (pre-keyframe audio, or a stream
@@ -1452,10 +1521,27 @@ impl<W: Write + Seek> MkvMuxer<W> {
             .get(track_idx)
             .copied()
             .unwrap_or(false);
+        let is_vobsub = self
+            .track_is_vobsub
+            .get(track_idx)
+            .copied()
+            .unwrap_or(false);
+        if is_vobsub && let Some((pos, start)) = self.vobsub_open_end.remove(&track_idx) {
+            let ticks = ((pts_ticks - start).max(1) as u64).min(VOBSUB_OPEN_END_TICKS);
+            self.patch_u24(pos, ticks)?;
+        }
+        let mut vobsub_open = false;
         let duration_ticks = match duration_ticks {
             Some(dt) => Some(dt),
+            None if is_vobsub => match vobsub_display_ns(data).filter(|&ns| ns > 0) {
+                Some(ns) => Some((ns as i64 / TIMESTAMP_SCALE_NS).max(1) as u64),
+                None => {
+                    vobsub_open = true;
+                    Some(VOBSUB_OPEN_END_TICKS)
+                }
+            },
             None if is_subtitle => {
-                Some((MIN_SUBTITLE_BLOCK_DURATION_NS / TIMESTAMP_SCALE_NS).max(1) as u64)
+                Some((PGS_FALLBACK_BLOCK_DURATION_NS / TIMESTAMP_SCALE_NS) as u64)
             }
             None => None,
         };
@@ -1533,6 +1619,11 @@ impl<W: Write + Seek> MkvMuxer<W> {
                 },
             }
         }
+        // BlockDuration is the group's last element: its 3 value bytes end the write.
+        if vobsub_open {
+            let end = self.writer.stream_position()?;
+            self.vobsub_open_end.insert(track_idx, (end - 3, pts_ticks));
+        }
         // Recorded per track (not a single global slot) so a later non-keyframe
         // references a keyframe on its OWN track — a shared slot produced cross-track
         // references on a multi-video-track title (MVC base+EL, or a two-angle disc).
@@ -1983,7 +2074,7 @@ impl<W: Write + Seek> MkvMuxer<W> {
         buf.push(flags);
         buf.extend_from_slice(data);
         if discard_padding_ns != 0 {
-            ebml::write_int(buf, 0x75A2, discard_padding_ns)?;
+            ebml::write_int(buf, ebml::DISCARD_PADDING, discard_padding_ns)?;
         }
         if let Some(dt) = duration_ticks {
             ebml::write_uint(buf, ebml::BLOCK_DURATION, dt)?;
@@ -5072,6 +5163,63 @@ mod tests {
         assert!(groups.iter().any(|g| g.track == 2), "sparse track present");
     }
 
+    // A VobSub SPU: 4-byte header, 2 pixel bytes, then a STA_DSP DCSQ (delay 0)
+    // chained to a STP_DSP DCSQ at `stop_delay` (units of 1024/90000 s).
+    fn vobsub_spu(stop_delay: u16) -> Vec<u8> {
+        let mut spu = vec![0, 18, 0, 6, 0xAB, 0xCD];
+        spu.extend_from_slice(&[0, 0, 0, 12, 0x01, 0xFF]);
+        spu.extend_from_slice(&stop_delay.to_be_bytes());
+        spu.extend_from_slice(&[0, 12, 0x02, 0xFF]);
+        spu
+    }
+
+    #[test]
+    fn vobsub_block_duration_comes_from_the_spu_stop_command() {
+        let mut track = make_subtitle_track();
+        track.codec_id = ebml::CODEC_VOBSUB;
+        // 0x0100 * 1024 / 90 kHz = 2.9127 s -> 29_127 ticks at 0.1 ms.
+        let data = mux_with_durations(&[track], &[(0, 0, true, vobsub_spu(0x0100), None)]);
+        let groups = all_block_groups(&data);
+        assert_eq!(groups.len(), 1);
+        assert_eq!(groups[0].duration, 29_127, "SPU stop time, not a 1 s guess");
+    }
+
+    // A VobSub SPU with no stop command ends at the next SPU on its track, else
+    // after a 10 s cap.
+    #[test]
+    fn vobsub_without_a_stop_ends_at_the_next_spu_or_the_cap() {
+        let mut track = make_subtitle_track();
+        track.codec_id = ebml::CODEC_VOBSUB;
+        // One DCSQ at offset 6: delay 0, next = itself, STA_DSP, CMD_END.
+        let spu = vec![0, 12, 0, 6, 0xAB, 0xCD, 0, 0, 0, 6, 0x01, 0xFF];
+        let data = mux_with_durations(
+            &[track],
+            &[
+                (0, 0, true, spu.clone(), None),
+                (0, 2_000_000_000, true, spu, None),
+            ],
+        );
+        let groups = all_block_groups(&data);
+        assert_eq!(groups.len(), 2);
+        assert_eq!(groups[0].duration, 20_000, "ends at the next SPU");
+        assert_eq!(groups[1].duration, 100_000, "last one capped at 10 s");
+    }
+
+    #[test]
+    fn a_durationless_pgs_block_is_not_cut_to_one_second() {
+        let data = mux_with_durations(
+            &[make_subtitle_track()],
+            &[(0, 0, true, vec![0x16, 0xAA], None)],
+        );
+        let groups = all_block_groups(&data);
+        assert_eq!(groups.len(), 1);
+        assert!(
+            groups[0].duration > 10_000,
+            "a self-terminating cue must not be hidden after 1 s (got {} ticks)",
+            groups[0].duration
+        );
+    }
+
     // Cluster boundary (CLUSTER_DURATION_TICKS): a new cluster opens on a video
     // keyframe once >= the cluster duration has elapsed since the open cluster's
     // timestamp; one just under the boundary stays in the current cluster.

src/mux/mp4/audio.rs

@@ -349,7 +349,7 @@ fn dts_channel_layout(amode: usize, lfe: bool) -> u16 {
 /// The `ddts` config box (ETSI TS 102 114 Annex; DTS-in-ISO registration).
 /// Describes the DTS core; whole access units (core + any extension) are passed
 /// through as samples, so a DTS-HD-aware decoder still finds the extension.
-fn ddts_box(c: &DtsConfig) -> Vec<u8> {
+fn ddts_box(c: &DtsConfig, max_rate: u32) -> Vec<u8> {
     // avg/max bitrate: derived from core frame size × frame rate (core RATE reads
     // "open/variable" for lossless, so unusable directly). Rate isn't integral
     // (e.g. 93.75 frames/s), so multiply before dividing, rounding to nearest.
@@ -362,14 +362,18 @@ fn ddts_box(c: &DtsConfig) -> Vec<u8> {
     };
 
     let mut out = Vec::new();
-    out.extend_from_slice(&c.sample_rate.to_be_bytes()); // DTSSamplingFrequency
+    out.extend_from_slice(&max_rate.to_be_bytes()); // DTSSamplingFrequency
     out.extend_from_slice(&bitrate.to_be_bytes()); // maxBitrate
     out.extend_from_slice(&bitrate.to_be_bytes()); // avgBitrate
     out.push(if c.has_extension { 24 } else { 16 }); // pcmSampleDepth
+    // FrameDuration counts samples at DTSSamplingFrequency, not the core rate.
+    let at_max = (u64::from(c.frame_samples) * u64::from(max_rate))
+        .checked_div(u64::from(c.sample_rate))
+        .unwrap_or(u64::from(c.frame_samples));
     // Bit-packed tail (56 bits): FrameDuration2 StreamConstruction5 CoreLFEPresent1
     // CoreLayout6 CoreSize14 StereoDownmix1 RepresentationType3 ChannelLayout16
     // MultiAssetFlag1 LBRDurationMod1 ReservedBoxPresent1 Reserved5
-    let frame_duration = match c.frame_samples {
+    let frame_duration = match at_max {
         0..=512 => 0,
         513..=1024 => 1,
         1025..=2048 => 2,
@@ -404,10 +408,26 @@ fn ddts_box(c: &DtsConfig) -> Vec<u8> {
     bx(b"ddts", &out)
 }
 
+// DTS sample-entry samplerate: the family base of the maximum rate (ETSI TS 102 114
+// E.2.2.2); the 4x core rates join their family per Table 5-6.
+fn dts_entry_rate(max_rate: u32) -> u32 {
+    match max_rate {
+        12_000 | 24_000 | 48_000 | 96_000 | 192_000 => 48_000,
+        11_025 | 22_050 | 44_100 | 88_200 | 176_400 => 44_100,
+        8_000 | 16_000 | 32_000 | 64_000 | 128_000 => 32_000,
+        other => other,
+    }
+}
+
 /// The MP4 fourcc + config box for an audio frame, or `None` if the codec has no
 /// MP4 mapping here. Together with [`audio_fits`] this is the fit oracle for
 /// audio: only what returns `Some` is muxable.
-pub(super) fn dolby_sample_entry(codec: Codec, first_frame: &[u8]) -> Option<Vec<u8>> {
+/// `stream_hz` is the title's rate for the track (0 if unknown).
+pub(super) fn dolby_sample_entry(
+    codec: Codec,
+    first_frame: &[u8],
+    stream_hz: u32,
+) -> Option<Vec<u8>> {
     match codec {
         Codec::Ac3 => {
             let c = parse_dolby(first_frame)?;
@@ -439,11 +459,17 @@ pub(super) fn dolby_sample_entry(codec: Codec, first_frame: &[u8]) -> Option<Vec
             let c = parse_dts(first_frame)?;
             // `dtsc` = DTS core; `dtsh` = DTS-HD (core + extension substreams).
             let fourcc: &[u8; 4] = if c.has_extension { b"dtsh" } else { b"dtsc" };
+            // The core may run at a divisor of the stream rate (48 kHz core in 96 kHz MA).
+            let max_rate = if stream_hz > c.sample_rate && stream_hz.is_multiple_of(c.sample_rate) {
+                stream_hz
+            } else {
+                c.sample_rate
+            };
             Some(audio_sample_entry(
                 fourcc,
                 c.channels,
-                c.sample_rate,
+                dts_entry_rate(max_rate),
-                &ddts_box(&c),
+                &ddts_box(&c, max_rate),
             ))
         }
         _ => None,
@@ -549,7 +575,7 @@ mod tests {
         // A Codec::Ac3Plus track whose first syncframe is legacy AC-3 (bsid 8 < 11)
         // has no nominal data rate; an `ec-3` entry built from it would declare
         // 0 kbps, violating ETSI TS 102 366 F.6.1 — emit ac-3/dac3 (Annex F.4) instead.
-        let e = dolby_sample_entry(Codec::Ac3Plus, &ac3_frame_5_1()).expect("entry built");
+        let e = dolby_sample_entry(Codec::Ac3Plus, &ac3_frame_5_1(), 0).expect("entry built");
         assert_eq!(
             &e[4..8],
             b"ac-3",
@@ -570,7 +596,7 @@ mod tests {
         // A real Annex-E syncframe still gets `ec-3` + `dec3`, and the data_rate
         // read back out of the emitted box is the frame's nominal rate, not 0.
         // 128 B / 1536 samples @ 48 kHz = 32 kbit/s.
-        let e = dolby_sample_entry(Codec::Ac3Plus, &eac3_frame_5_1()).expect("entry built");
+        let e = dolby_sample_entry(Codec::Ac3Plus, &eac3_frame_5_1(), 0).expect("entry built");
         assert_eq!(&e[4..8], b"ec-3");
         let i = e
             .windows(4)
@@ -642,7 +668,7 @@ mod tests {
         assert_eq!(c.core_size, 96);
         assert_eq!(c.frame_samples, 512);
 
-        let ddts = ddts_box(&c);
+        let ddts = ddts_box(&c, c.sample_rate);
         assert_eq!(&ddts[4..8], b"ddts");
         // DTSSamplingFrequency (first field) = 48000.
         assert_eq!(
@@ -650,10 +676,57 @@ mod tests {
             48_000
         );
         // Sample entry uses dtsc (no extension in this synthetic frame).
-        let e = dolby_sample_entry(Codec::DtsHdMa, &f).unwrap();
+        let e = dolby_sample_entry(Codec::DtsHdMa, &f, 0).unwrap();
         assert_eq!(&e[4..8], b"dtsc");
     }
 
+    // ETSI TS 102 114 E.2.2: a 48 kHz core in a 96 kHz stream → entry 48000, ddts 96000.
+    #[test]
+    fn a_96k_dts_stream_writes_its_max_rate_in_ddts_and_the_family_base_in_the_entry() {
+        let f = vec![
+            0x7F, 0xFE, 0x80, 0x01, 0x00, 0x3C, 0x05, 0xF2, 0x77, 0x00, 0x02, 0x00,
+        ];
+        let e = dolby_sample_entry(Codec::DtsHdMa, &f, 96_000).unwrap();
+        assert_eq!(&e[32..34], &48_000u16.to_be_bytes(), "entry samplerate");
+        let ddts = e.windows(4).position(|w| w == b"ddts").unwrap();
+        assert_eq!(&e[ddts + 4..ddts + 8], &96_000u32.to_be_bytes());
+        // FrameDuration counts samples at DTSSamplingFrequency: 512 core samples
+        // at 48 kHz are 1024 at 96 kHz → code 1.
+        let c = parse_dts(&f).unwrap();
+        assert_eq!(ddts_tail(&ddts_box(&c, 96_000))[0] >> 6, 1);
+        assert_eq!(ddts_tail(&ddts_box(&c, 48_000))[0] >> 6, 0);
+        // A title rate that is not a multiple of the core rate is ignored.
+        let e = dolby_sample_entry(Codec::DtsHdMa, &f, 44_100).unwrap();
+        let ddts = e.windows(4).position(|w| w == b"ddts").unwrap();
+        assert_eq!(&e[ddts + 4..ddts + 8], &48_000u32.to_be_bytes());
+    }
+
+    // ETSI TS 102 114 E.2.2.2 families, with the 4x-interpolated core rates of
+    // Table 5-6 (8/11.025/12 kHz) in the same family as their 2x siblings.
+    #[test]
+    fn dts_entry_rate_maps_every_rate_to_its_family_base() {
+        for (rate, base) in [
+            (12_000, 48_000),
+            (24_000, 48_000),
+            (48_000, 48_000),
+            (96_000, 48_000),
+            (192_000, 48_000),
+            (11_025, 44_100),
+            (22_050, 44_100),
+            (44_100, 44_100),
+            (88_200, 44_100),
+            (176_400, 44_100),
+            (8_000, 32_000),
+            (16_000, 32_000),
+            (32_000, 32_000),
+            (64_000, 32_000),
+            (128_000, 32_000),
+            (36_000, 36_000),
+        ] {
+            assert_eq!(dts_entry_rate(rate), base, "{rate}");
+        }
+    }
+
     #[test]
     fn dts_ext_sync_inside_core_is_not_a_false_positive() {
         // A 4-byte ext-sync pattern INSIDE the compressed core payload must NOT be
@@ -665,7 +738,7 @@ mod tests {
         ];
         let c = parse_dts(&f).expect("parses");
         assert!(!c.has_extension, "ext-sync inside core is not an extension");
-        let e = dolby_sample_entry(Codec::DtsHdMa, &f).unwrap();
+        let e = dolby_sample_entry(Codec::DtsHdMa, &f, 0).unwrap();
         assert_eq!(&e[4..8], b"dtsc");
     }
 
@@ -679,7 +752,7 @@ mod tests {
         let c = parse_dts(&f).expect("parses");
         assert_eq!(c.core_size, 9);
         assert!(c.has_extension, "EXSS sync at core end is a real extension");
-        let e = dolby_sample_entry(Codec::DtsHdMa, &f).unwrap();
+        let e = dolby_sample_entry(Codec::DtsHdMa, &f, 0).unwrap();
         assert_eq!(&e[4..8], b"dtsh");
     }
 
@@ -723,7 +796,7 @@ mod tests {
             has_extension: false,
             channel_layout: dts_channel_layout(9, true),
         };
-        let b = ddts_box(&c);
+        let b = ddts_box(&c, c.sample_rate);
         // Decode CoreSize back out of the emitted box: 8-byte header + 13 bytes
         // (freq/bitrates/depth) precede the 56-bit tail, where CoreSize sits after
         // FrameDuration(2)+StreamConstruction(5)+CoreLFEPresent(1)+CoreLayout(6) = bits 14..28.
@@ -752,7 +825,7 @@ mod tests {
         // Decode the flag from the emitted box: 8-byte header + 13 bytes precede the
         // 56-bit tail, where MultiAssetFlag is bit 48 — after FrameDuration+
         // StreamConstruction+CoreLFEPresent+CoreLayout+CoreSize+StereoDownmix+RepType+ChannelLayout — MSB of byte 7.
-        let b = ddts_box(&c);
+        let b = ddts_box(&c, c.sample_rate);
         let tail = &b[8 + 13..];
         assert_eq!(tail.len(), 7, "56-bit packed tail");
         assert_eq!(
@@ -1114,7 +1187,7 @@ mod tests {
             !c.has_extension,
             "the pattern ends at the core's last byte, inside it"
         );
-        let e = dolby_sample_entry(Codec::Dts, &f).expect("entry built");
+        let e = dolby_sample_entry(Codec::Dts, &f, 0).expect("entry built");
         assert_eq!(&e[4..8], b"dtsc");
     }
 
@@ -1133,7 +1206,7 @@ mod tests {
             has_extension: false,
             channel_layout: dts_channel_layout(9, true),
         };
-        let (max, avg) = ddts_bitrates(&ddts_box(&c));
+        let (max, avg) = ddts_bitrates(&ddts_box(&c, c.sample_rate));
         assert_eq!(max, 5513, "2822400/512 = 5512.5 → 5513");
         assert_eq!(avg, 5513, "both fields carry the same computed rate");
     }
@@ -1152,7 +1225,7 @@ mod tests {
             has_extension: false,
             channel_layout: dts_channel_layout(9, true),
         };
-        assert_eq!(ddts_bitrates(&ddts_box(&c)), (0, 0));
+        assert_eq!(ddts_bitrates(&ddts_box(&c, c.sample_rate)), (0, 0));
     }
 
     #[test]
@@ -1171,7 +1244,7 @@ mod tests {
                 has_extension: false,
                 channel_layout: dts_channel_layout(9, true),
             };
-            ddts_tail(&ddts_box(&c))[0] >> 6
+            ddts_tail(&ddts_box(&c, c.sample_rate))[0] >> 6
         };
         assert_eq!(dur(512), 0, "512 samples");
         assert_eq!(dur(1024), 1, "1024 samples");

src/mux/mp4/mod.rs

@@ -428,7 +428,11 @@ impl<W: Write + Seek + Send> Stream for Mp4Sink<W> {
         // here used to lose audio silently — finish() now reports that case loudly.
         if self.tracks[slot].media == Media::Audio
             && self.tracks[slot].audio_entry.is_none()
-            && let Some(entry) = audio::dolby_sample_entry(self.tracks[slot].codec, &frame.data)
+            && let Some(entry) = audio::dolby_sample_entry(
+                self.tracks[slot].codec,
+                &frame.data,
+                self.tracks[slot].audio_timescale,
+            )
         {
             self.tracks[slot].audio_entry = Some(entry);
         }

src/mux/mp4/read.rs

@@ -135,6 +135,8 @@ impl<R: Read + Seek> Mp4Reader<R> {
                 config,
                 channels,
                 sample_rate: entry_sample_rate,
+                dts_max_rate,
+                dts_hd,
             }) = parse_stsd(stsd)
             else {
                 tracing::warn!(
@@ -165,13 +167,10 @@ impl<R: Read + Seek> Mp4Reader<R> {
                     // `as u8` (a crafted 256 would alias to 0/Mono).
                     channels: AudioChannels::from_count(channels.min(u8::MAX as u16) as u8),
                     language: language.clone().unwrap_or_else(|| "und".into()),
-                    // Prefer the AudioSampleEntry's own samplerate; the mdhd media
+                    sample_rate: SampleRate::from_hz(
-                    // timescale is only a fallback when the entry omits it (0).
+                        dts_max_rate
-                    sample_rate: SampleRate::from_hz(if entry_sample_rate != 0 {
+                            .unwrap_or_else(|| audio_rate(entry_sample_rate, timescale, dts_hd)),
-                        entry_sample_rate
+                    ),
-                    } else {
-                        timescale
-                    }),
                     secondary: false,
                     purpose: LabelPurpose::Normal,
                     label: String::new(),
@@ -679,6 +678,10 @@ struct StsdInfo {
     /// field (high 16 bits). `0` when absent/video — the caller then falls
     /// back to the mdhd media timescale.
     sample_rate: u32,
+    /// DTS `ddts` DTSSamplingFrequency when present and non-zero.
+    dts_max_rate: Option<u32>,
+    /// `Some(dtsh|dtsl)` for a DTS entry; `None` otherwise.
+    dts_hd: Option<bool>,
 }
 
 /// stsd → codec + dimensions + codec_private + channel count (first entry).
@@ -727,35 +730,84 @@ fn parse_stsd(b: &[u8]) -> Option<StsdInfo> {
             config,
             channels: 0,
             sample_rate: 0,
+            dts_max_rate: None,
+            dts_hd: None,
         })
     } else {
-        // AudioSampleEntry header is 28 bytes, then child boxes. AAC (mp4a) carries
+        // AudioSampleEntry (ISO/IEC 14496-12 §12.2.3): 28 bytes then children. Under
-        // its AudioSpecificConfig in an `esds` box — the MKV CodecPrivate for
+        // a v0 stsd, entry v1 is QuickTime (+16 bytes) and v2 has a float64 rate at
-        // A_AAC. AC-3/DTS are self-describing in-band (None).
+        // 32, u32 channels at 40, children at 64.
-        let channels = if body.len() >= 28 { be16(body, 16) } else { 2 };
+        let version = if b[0] == 0 && body.len() >= 28 {
-        // Real audio sample rate: AudioSampleEntry 16.16 samplerate field at body
+            be16(body, 8)
-        // offset 24 (ISO/IEC 14496-12 §12.2.3), integer rate = high 16 bits. The mdhd
-        // media timescale is only a fallback (USUALLY the rate, but need not be).
-        let sample_rate = if body.len() >= 28 {
-            be16(body, 24) as u32
         } else {
             0
         };
-        let config = if matches!(codec, Codec::Aac) && body.len() >= 28 {
+        let (channels, sample_rate, children) = match version {
-            find_box(&body[28..], b"esds").and_then(parse_esds_asc)
+            2 if body.len() >= 64 => {
+                let rate = f64::from_bits(u64::from_be_bytes(body[32..40].try_into().ok()?));
+                let rate = if rate.is_finite() && rate >= 1.0 && rate <= u32::MAX as f64 {
+                    rate.round() as u32
+                } else {
+                    0
+                };
+                let ch = be32(body, 40).min(u16::MAX as u32) as u16;
+                (ch, rate, 64)
+            }
+            // The 16.16 field's integer part; `audio_rate` falls back to mdhd.
+            1 if body.len() >= 44 => (be16(body, 16), be16(body, 24) as u32, 44),
+            _ if body.len() >= 28 => (be16(body, 16), be16(body, 24) as u32, 28),
+            _ => (2, 0, body.len()),
+        };
+        let config = if matches!(codec, Codec::Aac) {
+            find_box(&body[children..], b"esds").and_then(parse_esds_asc)
         } else {
             None
         };
+        // ISO AudioSampleEntryV1 carries rates above 65535 in `srat`.
+        let sample_rate = find_box(&body[children..], b"srat")
+            .filter(|b| b.len() >= 8)
+            .map_or(sample_rate, |b| be32(b, 4));
+        // ETSI TS 102 114 E.2.2.3: DTSSamplingFrequency is the stream's maximum rate.
+        // An unnameable rate stays Unknown; the entry holds only its family base.
+        let dts_max_rate = match codec {
+            Codec::Dts => find_box(&body[children..], b"ddts")
+                .filter(|b| b.len() >= 4)
+                .map(|b| be32(b, 0))
+                .filter(|&hz| hz != 0),
+            _ => None,
+        };
         Some(StsdInfo {
             codec,
             height: 0,
             config,
             channels,
             sample_rate,
+            dts_max_rate,
+            dts_hd: (codec == Codec::Dts).then_some(matches!(&fourcc, b"dtsh" | b"dtsl")),
         })
     }
 }
 
+// Sample rate from the entry rate and the mdhd timescale. The entry wins when
+// it is known; a DTS-HD entry (`dts_hd`) holds the family base of a 2x/4x rate.
+fn audio_rate(entry: u32, timescale: u32, dts_hd: Option<bool>) -> u32 {
+    let known = |hz| SampleRate::from_hz(hz) != SampleRate::Unknown;
+    if matches!(entry, 0 | 1 | 0xFFFF) {
+        return timescale;
+    }
+    let is = |k: u32| entry.checked_mul(k) == Some(timescale) && known(timescale);
+    let wider = match dts_hd {
+        Some(hd) => hd && (is(2) || is(4)),
+        None => {
+            !known(entry)
+                && timescale > 0xFFFF
+                && known(timescale)
+                && timescale.is_multiple_of(entry)
+        }
+    };
+    if wider { timescale } else { entry }
+}
+
 /// Read an MPEG-4 expandable descriptor length (ISO/IEC 14496-1), advancing `pos`.
 /// Each byte contributes 7 bits, continued while the high bit is set (max 4 bytes).
 fn read_descriptor_len(b: &[u8], pos: &mut usize) -> usize {
@@ -2251,13 +2303,12 @@ mod tests {
     /// (mdhd timescale = 90_000, AudioSampleEntry samplerate = 48_000); the
     /// entry value must win. Old code read the rate from the timescale, so a
     /// 90 kHz mdhd yielded SampleRate::Unknown, dropping the real 48 kHz.
-    #[test]
+    // One-sample audio moov whose mdhd timescale and sample entry are given.
-    fn audio_sample_entry_samplerate_wins_over_mdhd_timescale() {
+    fn audio_moov(timescale: u32, fourcc: &[u8; 4], entry: &[u8]) -> Vec<u8> {
-        use std::io::Cursor;
         let mdhd = {
             // v0: version+flags(4) creation(4) modification(4) timescale(4) duration(4).
             let mut p = vec![0u8; 24];
-            p[12..16].copy_from_slice(&90_000u32.to_be_bytes()); // NOT the audio rate
+            p[12..16].copy_from_slice(&timescale.to_be_bytes());
             mp4_box(b"mdhd", &p)
         };
         let hdlr = {
@@ -2265,8 +2316,7 @@ mod tests {
             p[8..12].copy_from_slice(b"soun");
             mp4_box(b"hdlr", &p)
         };
-        // Full AudioSampleEntry: samplerate 48_000 in the 16.16 field at body 24.
+        let stsd = mp4_box(b"stsd", &stsd_with(fourcc, entry));
-        let stsd = mp4_box(b"stsd", &stsd_with(b"ac-3", &audio_entry(6, 0, &[])));
         // stsz: sample_size=10, count=1 (one 10-byte sample).
         let stsz = {
             let mut p = vec![0u8, 0, 0, 0];
@@ -2305,19 +2355,216 @@ mod tests {
         mdia.extend_from_slice(&hdlr);
         mdia.extend_from_slice(&minf);
         let trak = mp4_box(b"trak", &mp4_box(b"mdia", &mdia));
-        let moov = mp4_box(b"moov", &trak);
+        mp4_box(b"moov", &trak)
+    }
 
-        let rd = Mp4Reader::from_reader(Cursor::new(moov), "sr".into()).unwrap();
+    fn read_audio(moov: Vec<u8>) -> AudioStream {
+        let rd = Mp4Reader::from_reader(std::io::Cursor::new(moov), "sr".into()).unwrap();
         match &rd.info().streams[0] {
-            DiscStream::Audio(a) => assert_eq!(
+            DiscStream::Audio(a) => a.clone(),
-                a.sample_rate,
-                SampleRate::S48,
-                "the AudioSampleEntry 48 kHz must win over the 90 kHz mdhd timescale"
-            ),
             other => panic!("expected an audio stream, got {other:?}"),
         }
     }
 
+    /// Regression: audio sample rate must come from the AudioSampleEntry 16.16
+    /// samplerate field, NOT the mdhd media timescale, when they merely differ
+    /// (a 90 kHz mdhd is not an audio rate).
+    #[test]
+    fn audio_sample_entry_samplerate_wins_over_mdhd_timescale() {
+        let a = read_audio(audio_moov(90_000, b"ac-3", &audio_entry(6, 0, &[])));
+        assert_eq!(a.sample_rate, SampleRate::S48);
+    }
+
+    // The 16-bit entry field cannot hold 96 kHz: our own Mp4Sink saturates it to
+    // 65535, and a DTS-HD entry may carry the 48 kHz core rate. mdhd wins then.
+    #[test]
+    fn a_high_rate_mdhd_timescale_beats_an_unrepresentable_entry_rate() {
+        for entry_rate in [0xFFFFu32, 48_000] {
+            let mut e = audio_entry(6, 0, &[]);
+            e[24..28].copy_from_slice(&(entry_rate << 16).to_be_bytes());
+            let a = read_audio(audio_moov(96_000, b"dtsh", &e));
+            assert_eq!(a.sample_rate, SampleRate::S96, "entry {entry_rate}");
+        }
+    }
+
+    // Without ddts, only a DTS-HD entry takes a 2x/4x mdhd timescale; dtsc keeps its rate.
+    #[test]
+    fn a_dts_timescale_counts_only_as_2x_or_4x_on_dts_hd_entries() {
+        let entry = |hz: u32| {
+            let mut e = audio_entry(6, 0, &[]);
+            e[24..28].copy_from_slice(&(hz << 16).to_be_bytes());
+            e
+        };
+        for (fourcc, entry_hz, ts, want) in [
+            (b"dtsc", 48_000, 96_000, SampleRate::S48),
+            (b"dtsh", 48_000, 192_000, SampleRate::S192),
+            (b"dtsl", 44_100, 88_200, SampleRate::S88_2),
+            (b"dtsh", 32_000, 96_000, SampleRate::from_hz(32_000)),
+        ] {
+            let a = read_audio(audio_moov(ts, fourcc, &entry(entry_hz)));
+            assert_eq!(a.sample_rate, want, "{fourcc:?} {entry_hz} {ts}");
+        }
+    }
+
+    // ETSI TS 102 114 E.2.2.3: ddts DTSSamplingFrequency is the stream's maximum rate.
+    #[test]
+    fn a_dts_ddts_max_rate_beats_the_family_entry_rate() {
+        let mut ddts = 96_000u32.to_be_bytes().to_vec();
+        ddts.extend_from_slice(&[0u8; 15]);
+        let e = audio_entry(6, 0, &mp4_box(b"ddts", &ddts));
+        let a = read_audio(audio_moov(90_000, b"dtsh", &e));
+        assert_eq!(a.sample_rate, SampleRate::S96);
+    }
+
+    // A ddts rate SampleRate cannot name reads as Unknown, never the entry's family base.
+    #[test]
+    fn a_dts_ddts_rate_outside_sample_rate_is_not_read_as_its_family_base() {
+        for hz in [12_000u32, 24_000, 32_000] {
+            let mut ddts = hz.to_be_bytes().to_vec();
+            ddts.extend_from_slice(&[0u8; 15]);
+            // audio_entry's samplerate is 48000, the family base of 12/24 kHz.
+            let e = audio_entry(6, 0, &mp4_box(b"ddts", &ddts));
+            let a = read_audio(audio_moov(48_000, b"dtsc", &e));
+            assert_eq!(a.sample_rate, SampleRate::Unknown, "{hz}");
+        }
+    }
+
+    // A 96 kHz DTS-HD MA track written by Mp4Sink reads back as 96 kHz.
+    #[test]
+    fn a_96k_dts_hd_track_round_trips_its_rate() {
+        use crate::disc::{AudioChannels, AudioStream, Codec, DiscTitle, LabelPurpose};
+        use crate::mux::mp4::Mp4Sink;
+        use crate::pes::{PesFrame, Stream as _};
+        use std::io::Cursor;
+
+        let mut t = DiscTitle::empty();
+        t.streams = vec![
+            DiscStream::Video(VideoStream {
+                pid: 0x1011,
+                codec: Codec::Hevc,
+                resolution: Resolution::R1080p,
+                frame_rate: crate::disc::FrameRate::F23_976,
+                hdr: crate::disc::HdrFormat::Sdr,
+                color_space: crate::disc::ColorSpace::Unknown,
+                display_aspect: None,
+                secondary: false,
+                label: String::new(),
+                measured_cicp: None,
+            }),
+            DiscStream::Audio(AudioStream {
+                pid: 0x1100,
+                codec: Codec::DtsHdMa,
+                channels: AudioChannels::Surround51,
+                language: "eng".into(),
+                sample_rate: SampleRate::S96,
+                secondary: false,
+                purpose: LabelPurpose::Normal,
+                label: String::new(),
+            }),
+        ];
+        t.codec_privates = vec![Some(vec![0xAA, 0xBB, 0xCC]), None];
+        // 48 kHz DTS core (SFREQ=13), as on a 96 kHz DTS-HD MA disc track.
+        let core = vec![
+            0x7F, 0xFE, 0x80, 0x01, 0x00, 0x3C, 0x05, 0xF2, 0x77, 0x00, 0x02, 0x00,
+        ];
+        let mut buf = Vec::new();
+        {
+            let mut sink = Mp4Sink::create(Cursor::new(&mut buf), &t).unwrap();
+            for i in 0..2i64 {
+                sink.write(&PesFrame {
+                    discard_padding_ns: 0,
+                    track: 0,
+                    pts: i * 41_708_333,
+                    keyframe: i == 0,
+                    data: vec![0x11; 64],
+                    duration_ns: None,
+                    source: None,
+                    coding: None,
+                })
+                .unwrap();
+                sink.write(&PesFrame {
+                    discard_padding_ns: 0,
+                    track: 1,
+                    pts: i * 5_333_333,
+                    keyframe: true,
+                    data: core.clone(),
+                    duration_ns: None,
+                    source: None,
+                    coding: None,
+                })
+                .unwrap();
+            }
+            sink.finish().unwrap();
+        }
+        let rd = Mp4Reader::from_reader(Cursor::new(buf), "dts96".into()).unwrap();
+        match &rd.info().streams[1] {
+            DiscStream::Audio(a) => assert_eq!(a.sample_rate, SampleRate::S96),
+            other => panic!("expected audio, got {other:?}"),
+        }
+    }
+
+    // A valid known entry rate is kept even when mdhd uses a higher timescale.
+    #[test]
+    fn a_known_entry_rate_beats_a_higher_timescale() {
+        let a = read_audio(audio_moov(96_000, b"ac-3", &audio_entry(2, 0, &[])));
+        assert_eq!(a.sample_rate, SampleRate::S48);
+    }
+
+    // ISO AudioSampleEntryV1 (stsd version 1): not a QuickTime layout; children
+    // follow the 28-byte part and `srat` carries a rate above 65535.
+    #[test]
+    fn an_iso_v1_sample_entry_reads_srat_and_keeps_its_children() {
+        let asc = [0x11u8, 0x90];
+        let mut esds = vec![0u8, 0, 0, 0, 0x03, 22, 0, 1, 0, 0x04, 17, 0x40, 0x15];
+        esds.extend_from_slice(&[0u8; 11]);
+        esds.extend_from_slice(&[0x05, 2]);
+        esds.extend_from_slice(&asc);
+        let mut srat = vec![0u8; 4];
+        srat.extend_from_slice(&96_000u32.to_be_bytes());
+        let mut e = audio_entry(2, 0, &[]);
+        e[8..10].copy_from_slice(&1u16.to_be_bytes());
+        e[24..28].copy_from_slice(&(1u32 << 16).to_be_bytes());
+        e.extend_from_slice(&mp4_box(b"esds", &esds));
+        e.extend_from_slice(&mp4_box(b"srat", &srat));
+        let mut stsd = stsd_with(b"mp4a", &e);
+        stsd[0] = 1;
+        let info = parse_stsd(&stsd).expect("mp4a parses");
+        assert_eq!(info.config.as_deref(), Some(&asc[..]));
+        assert_eq!(info.sample_rate, 96_000);
+    }
+
+    // QuickTime SoundDescription v2: the 16.16 field is fixed at 1.0 and the real
+    // rate is a float64 at body offset 32; channels at 40; children at 64.
+    #[test]
+    fn a_quicktime_v2_sound_description_reads_its_float_rate_and_channels() {
+        let mut e = vec![0u8; 64];
+        e[8..10].copy_from_slice(&2u16.to_be_bytes());
+        e[16..18].copy_from_slice(&3u16.to_be_bytes());
+        e[24..28].copy_from_slice(&(1u32 << 16).to_be_bytes());
+        e[32..40].copy_from_slice(&44_100f64.to_bits().to_be_bytes());
+        e[40..44].copy_from_slice(&6u32.to_be_bytes());
+        let a = read_audio(audio_moov(1_000, b"ac-3", &e));
+        assert_eq!(a.sample_rate, SampleRate::S44_1);
+        assert_eq!(a.channels, AudioChannels::from_count(6));
+    }
+
+    // QuickTime SoundDescription v1 inserts 16 bytes before the child boxes, so an
+    // mp4a esds must be looked up at body offset 44, not 28.
+    #[test]
+    fn a_quicktime_v1_sound_description_finds_its_esds_after_the_v1_fields() {
+        let asc = [0x12u8, 0x10];
+        let mut esds = vec![0u8, 0, 0, 0, 0x03, 22, 0, 1, 0, 0x04, 17, 0x40, 0x15];
+        esds.extend_from_slice(&[0u8; 11]);
+        esds.extend_from_slice(&[0x05, 2]);
+        esds.extend_from_slice(&asc);
+        let mut e = audio_entry(2, 0, &[]);
+        e[8..10].copy_from_slice(&1u16.to_be_bytes());
+        e.extend_from_slice(&[0u8; 16]);
+        e.extend_from_slice(&mp4_box(b"esds", &esds));
+        let info = parse_stsd(&stsd_with(b"mp4a", &e)).expect("mp4a parses");
+        assert_eq!(info.config.as_deref(), Some(&asc[..]));
+    }
+
     // ============================================================
     // Edit lists — ISO/IEC 14496-12 §8.6.5 (`edts`) / §8.6.6 (`elst`).
     // ============================================================

src/mux/network.rs

@@ -8,12 +8,117 @@
 
 use super::meta;
 use crate::disc::DiscTitle;
-use std::io::{self, BufReader, BufWriter, Write};
+use crate::halt::{Halt, POLL_INTERVAL};
+use rustix::event::{PollFd, PollFlags, Timespec};
+use std::io::{self, BufReader, BufWriter, Read, Write};
 use std::net::{IpAddr, TcpListener, TcpStream, ToSocketAddrs};
 
 /// I/O buffer size for network reads/writes.
 const NET_BUF_SIZE: usize = 256 * 1024;
 
+// Accept poll period while waiting for the sender (a halt is checked each tick).
+const ACCEPT_POLL: std::time::Duration = std::time::Duration::from_millis(50);
+
+// A slow sender (bad-sector retries) may be silent for minutes, so there is no
+// data-idle limit; a vanished peer is detected by TCP keepalive instead.
+const KEEPALIVE_IDLE: std::time::Duration = std::time::Duration::from_secs(60);
+#[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))]
+const KEEPALIVE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(10);
+#[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))]
+const KEEPALIVE_RETRIES: u32 = 6;
+
+// Enable keepalive probes so a crashed or unplugged sender fails the read.
+fn enable_keepalive(stream: &TcpStream) -> io::Result<()> {
+    let ka = socket2::TcpKeepalive::new().with_time(KEEPALIVE_IDLE);
+    #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))]
+    let ka = ka
+        .with_interval(KEEPALIVE_INTERVAL)
+        .with_retries(KEEPALIVE_RETRIES);
+    socket2::SockRef::from(stream).set_tcp_keepalive(&ka)
+}
+
+// Keepalive is best effort: a stack that rejects an option (e.g. TCP_KEEPCNT on
+// old Windows) still receives, just without dead-peer detection.
+fn arm_keepalive(stream: &TcpStream) {
+    if let Err(e) = enable_keepalive(stream) {
+        tracing::warn!(
+            target: "mux",
+            error = %e,
+            "network: TCP keepalive unavailable"
+        );
+    }
+}
+
+// Poll outcome: `Ok(true)` reads now (data, EOF or a socket error for `read` to
+// surface), `Ok(false)` is a halt-check tick. Readiness, not SO_RCVTIMEO: on
+// Windows a receive timeout can cancel a completing recv and lose its data.
+fn poll_says_read(res: io::Result<usize>, revents: PollFlags) -> io::Result<bool> {
+    match res {
+        Err(e) if e.kind() == io::ErrorKind::Interrupted => Ok(false),
+        Err(e) => Err(e),
+        Ok(0) => Ok(false),
+        Ok(_) => Ok(!revents.is_empty()),
+    }
+}
+
+// Receive side: waits for readability in `tick` slices so a halt is observed
+// instead of blocking in `read` forever.
+struct HaltRead {
+    stream: TcpStream,
+    halt: Option<Halt>,
+    halted: bool,
+    tick: Timespec,
+}
+
+impl HaltRead {
+    fn new(stream: TcpStream, halt: Option<Halt>) -> Self {
+        Self::with_tick(stream, halt, POLL_INTERVAL)
+    }
+
+    fn with_tick(stream: TcpStream, halt: Option<Halt>, tick: std::time::Duration) -> Self {
+        let tick = Timespec {
+            tv_sec: tick.as_secs() as _,
+            tv_nsec: tick.subsec_nanos() as _,
+        };
+        Self {
+            stream,
+            halt,
+            halted: false,
+            tick,
+        }
+    }
+
+    // `e`, or `Error::Halted` when the read was aborted by the halt.
+    fn halted_or(&self, e: io::Error) -> io::Error {
+        if self.halted {
+            crate::error::Error::Halted.into()
+        } else {
+            e
+        }
+    }
+}
+
+impl Read for HaltRead {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        let Some(halt) = &self.halt else {
+            return self.stream.read(buf);
+        };
+        loop {
+            if halt.is_cancelled() {
+                // Halted maps to ErrorKind::Interrupted, which std's read_exact
+                // retries forever; abort with another kind and remap in `halted_or`.
+                self.halted = true;
+                return Err(io::ErrorKind::ConnectionAborted.into());
+            }
+            let mut fds = [PollFd::new(&self.stream, PollFlags::IN)];
+            let res = rustix::event::poll(&mut fds, Some(&self.tick)).map_err(io::Error::from);
+            if poll_says_read(res, fds[0].revents())? {
+                return self.stream.read(buf);
+            }
+        }
+    }
+}
+
 // True if `ip` must never be a `network://` connect target (loopback, private, link-local,
 // unspecified, multicast). Re-checked here at connect time to close a DNS-rebinding TOCTOU.
 pub(crate) fn is_blocked_ip(ip: IpAddr) -> bool {
@@ -94,9 +199,12 @@ enum Mode {
     Write {
         writer: BufWriter<TcpStream>,
         header_written: bool,
+        timings: Vec<crate::pes::TrackTiming>,
+        padded: bool,
     },
     Read {
-        reader: BufReader<TcpStream>,
+        reader: BufReader<HaltRead>,
+        meta: meta::M2tsMeta,
     },
 }
 
@@ -135,6 +243,8 @@ impl NetworkStream {
             mode: Mode::Write {
                 writer: BufWriter::with_capacity(NET_BUF_SIZE, stream),
                 header_written: false,
+                timings: Vec::new(),
+                padded: false,
             },
         })
     }
@@ -158,7 +268,13 @@ impl NetworkStream {
     /// `accept`, so the bound port is freed and any subsequent connection
     /// attempt to the same address is refused.
     pub fn listen(addr: &str) -> io::Result<Self> {
-        Self::accept_from(TcpListener::bind(addr)?)
+        Self::listen_with_halt(addr, None)
+    }
+
+    /// [`listen`](Self::listen) that a [`Halt`] can interrupt, while waiting for
+    /// the sender to connect and while blocked on a stalled sender.
+    pub fn listen_with_halt(addr: &str, halt: Option<Halt>) -> io::Result<Self> {
+        Self::accept_from_with_halt(TcpListener::bind(addr)?, halt)
     }
 
     /// Accept one connection from an already-bound listener and read from it.
@@ -166,33 +282,69 @@ impl NetworkStream {
     /// `:0` bind) and hand the listener in, closing the bind/drop/re-bind race
     /// that `listen(addr)` would otherwise have.
     pub fn accept_from(listener: TcpListener) -> io::Result<Self> {
-        let (stream, _peer) = listener.accept()?;
+        Self::accept_from_with_halt(listener, None)
+    }
+
+    /// [`accept_from`](Self::accept_from) that a [`Halt`] can interrupt.
+    pub fn accept_from_with_halt(listener: TcpListener, halt: Option<Halt>) -> io::Result<Self> {
+        let Some(h) = halt else {
+            let (stream, _peer) = listener.accept()?;
+            stream.set_nodelay(true)?;
+            arm_keepalive(&stream);
+            return Self::read_from(stream, None);
+        };
+        listener.set_nonblocking(true)?;
+        let halt = Some(h);
+        let stream = loop {
+            if halt.as_ref().is_some_and(Halt::is_cancelled) {
+                return Err(crate::error::Error::Halted.into());
+            }
+            match listener.accept() {
+                Ok((stream, _peer)) => break stream,
+                Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
+                    std::thread::sleep(ACCEPT_POLL);
+                }
+                Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
+                Err(e) => return Err(e),
+            }
+        };
+        // Accepted sockets can inherit O_NONBLOCK (BSD/macOS); reads wait on readiness.
+        stream.set_nonblocking(false)?;
         stream.set_nodelay(true)?;
-        let mut reader = BufReader::with_capacity(NET_BUF_SIZE, stream);
+        arm_keepalive(&stream);
+        Self::read_from(stream, halt)
+    }
+
+    // Wrap an accepted connection and read its FMKV header.
+    fn read_from(stream: TcpStream, halt: Option<Halt>) -> io::Result<Self> {
+        let mut reader = BufReader::with_capacity(NET_BUF_SIZE, HaltRead::new(stream, halt));
 
         // Read FMKV metadata header
-        let disc_title = meta::read_header(&mut reader)?
+        let meta = meta::read_header(&mut reader)
-            .ok_or_else(|| -> io::Error { crate::error::Error::NoMetadata.into() })?
+            .map_err(|e| reader.get_ref().halted_or(e))?
-            .to_title();
+            .ok_or_else(|| -> io::Error { crate::error::Error::NoMetadata.into() })?;
 
         Ok(Self {
-            disc_title,
+            disc_title: meta.to_title(),
-            mode: Mode::Read { reader },
+            mode: Mode::Read { reader, meta },
         })
     }
 }
 
-// Write the FMKV header exactly once, before any frames. Always writes
+// Write the FMKV header once, before any frames, even for a zero-frame stream
-// (even for a title with no streams) so read_header() finds the magic
+// (else read_header() reports NoMetadata). Sets `padded` when frames carry the
-// and doesn't fall into NoMetadata on a zero-frame stream.
+// DiscardPadding extension (header v2).
 fn ensure_header_written(
     writer: &mut BufWriter<TcpStream>,
     header_written: &mut bool,
     disc_title: &DiscTitle,
+    timings: &[crate::pes::TrackTiming],
+    padded: &mut bool,
 ) -> io::Result<()> {
     if !*header_written {
-        let m = meta::M2tsMeta::from_title(disc_title);
+        let m = meta::M2tsMeta::from_title(disc_title).with_timings(timings);
         meta::write_header(writer, &m)?;
+        *padded = m.frame_padding;
         *header_written = true;
     }
     Ok(())
@@ -201,7 +353,10 @@ fn ensure_header_written(
 impl crate::pes::Stream for NetworkStream {
     fn read(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
         match &mut self.mode {
-            Mode::Read { reader } => crate::pes::PesFrame::deserialize(reader),
+            Mode::Read { reader, meta } => {
+                crate::pes::PesFrame::deserialize_ext(reader, meta.frame_padding)
+                    .map_err(|e| reader.get_ref().halted_or(e))
+            }
             _ => Err(crate::error::Error::StreamWriteOnly.into()),
         }
     }
@@ -210,9 +365,11 @@ impl crate::pes::Stream for NetworkStream {
             Mode::Write {
                 writer,
                 header_written,
+                timings,
+                padded,
             } => {
-                ensure_header_written(writer, header_written, &self.disc_title)?;
+                ensure_header_written(writer, header_written, &self.disc_title, timings, padded)?;
-                frame.serialize(writer)
+                frame.serialize_ext(writer, *padded)
             }
             _ => Err(crate::error::Error::StreamReadOnly.into()),
         }
@@ -221,12 +378,14 @@ impl crate::pes::Stream for NetworkStream {
         if let Mode::Write {
             writer,
             header_written,
+            timings,
+            padded,
         } = &mut self.mode
         {
             // Always emit the FMKV header before shutdown, even for a zero-frame stream,
             // or the receiver's read_header() sees a clean EOF and rejects the stream
             // with NoMetadata.
-            ensure_header_written(writer, header_written, &self.disc_title)?;
+            ensure_header_written(writer, header_written, &self.disc_title, timings, padded)?;
             writer.flush()?;
             writer.get_ref().shutdown(std::net::Shutdown::Write)?;
         }
@@ -235,6 +394,49 @@ impl crate::pes::Stream for NetworkStream {
     fn info(&self) -> &DiscTitle {
         &self.disc_title
     }
+    fn track_timing(&self, track: usize) -> crate::pes::TrackTiming {
+        match &self.mode {
+            Mode::Read { meta, .. } => meta.timing(track),
+            Mode::Write { .. } => Default::default(),
+        }
+    }
+    fn set_track_timing(
+        &mut self,
+        track: usize,
+        timing: crate::pes::TrackTiming,
+    ) -> io::Result<()> {
+        match &mut self.mode {
+            Mode::Write {
+                header_written: false,
+                timings,
+                ..
+            } => set_timing(timings, track, timing, self.disc_title.streams.len()),
+            // The header (which carries it) is already on the wire.
+            Mode::Write { .. } => Err(crate::error::Error::StreamHeaderWritten.into()),
+            Mode::Read { .. } => Err(crate::error::Error::StreamReadOnly.into()),
+        }
+    }
+    // The sender's codec privates travel in the FMKV header (as for stdio).
+    fn codec_private(&self, track: usize) -> Option<Vec<u8>> {
+        self.disc_title.codec_privates.get(track).cloned().flatten()
+    }
+}
+
+// Record `timing` for stream `track` of a `tracks`-stream title.
+pub(crate) fn set_timing(
+    timings: &mut Vec<crate::pes::TrackTiming>,
+    track: usize,
+    timing: crate::pes::TrackTiming,
+    tracks: usize,
+) -> io::Result<()> {
+    if track >= tracks {
+        return Err(crate::error::Error::MuxTrackRange { track, tracks }.into());
+    }
+    if timings.len() < tracks {
+        timings.resize(tracks, Default::default());
+    }
+    timings[track] = timing;
+    Ok(())
 }
 
 // NetworkStream is PES-only — no IOStream/Read/Write byte interface.
@@ -729,4 +931,341 @@ mod tests {
         assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
         handle.join().unwrap();
     }
+
+    // A Stop must interrupt a receiver still waiting for its sender to connect.
+    #[test]
+    fn a_halt_interrupts_a_pending_accept() {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let halt = crate::halt::Halt::new();
+        let h = halt.clone();
+        let (tx, rx) = std::sync::mpsc::channel();
+        std::thread::spawn(move || {
+            let r = NetworkStream::accept_from_with_halt(listener, Some(h)).map(|_| ());
+            let _ = tx.send(r);
+        });
+        std::thread::sleep(std::time::Duration::from_millis(100));
+        halt.cancel();
+        let r = rx
+            .recv_timeout(std::time::Duration::from_secs(5))
+            .expect("accept must observe the halt");
+        assert!(crate::error::is_halt(&r.unwrap_err()));
+    }
+
+    // ... and a receiver blocked on a sender that stalled mid-stream.
+    #[test]
+    fn a_halt_interrupts_a_stalled_read() {
+        use crate::pes;
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap();
+        let halt = crate::halt::Halt::new();
+        let h = halt.clone();
+        let (tx, rx) = std::sync::mpsc::channel();
+        std::thread::spawn(move || {
+            let mut ns = NetworkStream::accept_from_with_halt(listener, Some(h)).unwrap();
+            let _ = tx.send(pes::Stream::read(&mut ns).map(|_| ()));
+        });
+        let mut writer = NetworkStream::connect_vetted(&addr.to_string(), false)
+            .unwrap()
+            .meta(&sample_title());
+        if let Mode::Write {
+            writer: w,
+            header_written,
+            padded,
+            ..
+        } = &mut writer.mode
+        {
+            ensure_header_written(w, header_written, &sample_title(), &[], padded).unwrap();
+            w.flush().unwrap();
+        }
+        std::thread::sleep(std::time::Duration::from_millis(300));
+        halt.cancel();
+        let r = rx
+            .recv_timeout(std::time::Duration::from_secs(5))
+            .expect("read must observe the halt");
+        assert!(crate::error::is_halt(&r.unwrap_err()));
+        drop(writer);
+    }
+
+    // A halt while the sender has connected but not finished its header.
+    #[test]
+    fn a_halt_interrupts_a_stalled_header_read() {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap();
+        let halt = crate::halt::Halt::new();
+        let h = halt.clone();
+        let (tx, rx) = std::sync::mpsc::channel();
+        std::thread::spawn(move || {
+            let r = NetworkStream::accept_from_with_halt(listener, Some(h)).map(|_| ());
+            let _ = tx.send(r);
+        });
+        let mut sender = TcpStream::connect(addr).unwrap();
+        sender.write_all(b"FM").unwrap();
+        std::thread::sleep(std::time::Duration::from_millis(300));
+        halt.cancel();
+        let r = rx
+            .recv_timeout(std::time::Duration::from_secs(5))
+            .expect("the header read must observe the halt");
+        assert!(crate::error::is_halt(&r.unwrap_err()));
+    }
+
+    // A receiver without a halt (`listen`/`accept_from`) arms keepalive too.
+    #[test]
+    fn a_receiver_without_a_halt_arms_keepalive() {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap();
+        let handle = std::thread::spawn(move || {
+            let ns = NetworkStream::accept_from(listener).unwrap();
+            match &ns.mode {
+                Mode::Read { reader, .. } => socket2::SockRef::from(&reader.get_ref().stream)
+                    .keepalive()
+                    .unwrap(),
+                Mode::Write { .. } => false,
+            }
+        });
+        let mut writer = NetworkStream::connect_vetted(&addr.to_string(), false)
+            .unwrap()
+            .meta(&sample_title());
+        if let Mode::Write {
+            writer: w,
+            header_written,
+            padded,
+            ..
+        } = &mut writer.mode
+        {
+            ensure_header_written(w, header_written, &sample_title(), &[], padded).unwrap();
+            w.flush().unwrap();
+        }
+        assert!(handle.join().unwrap(), "keepalive must be armed");
+    }
+
+    // Windows may drop data when SO_RCVTIMEO cancels a recv, so a halt-aware
+    // receiver waits on readiness instead and never arms a read timeout.
+    #[test]
+    fn a_halt_aware_receiver_sets_no_read_timeout() {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap();
+        let handle = std::thread::spawn(move || {
+            let halt = crate::halt::Halt::new();
+            let ns = NetworkStream::accept_from_with_halt(listener, Some(halt)).unwrap();
+            match &ns.mode {
+                Mode::Read { reader, .. } => reader.get_ref().stream.read_timeout().unwrap(),
+                Mode::Write { .. } => None,
+            }
+        });
+        let mut writer = NetworkStream::connect_vetted(&addr.to_string(), false)
+            .unwrap()
+            .meta(&sample_title());
+        crate::pes::Stream::finish(&mut writer).unwrap();
+        assert_eq!(handle.join().unwrap(), None);
+    }
+
+    // Only a readiness report reads; a timeout or EINTR ticks; a poll failure surfaces.
+    #[test]
+    fn poll_says_read_only_when_the_socket_reports_readiness() {
+        use io::ErrorKind::{ConnectionReset, Interrupted, TimedOut, WouldBlock};
+        let ok = |n: usize, f: PollFlags| poll_says_read(Ok(n), f).unwrap();
+        assert!(!ok(0, PollFlags::empty()), "timeout is a tick");
+        assert!(!ok(1, PollFlags::empty()));
+        for f in [
+            PollFlags::IN,
+            PollFlags::HUP,
+            PollFlags::ERR,
+            PollFlags::NVAL,
+        ] {
+            assert!(ok(1, f), "{f:?} must hand over to read");
+        }
+        assert!(!poll_says_read(Err(Interrupted.into()), PollFlags::empty()).unwrap());
+        for kind in [WouldBlock, TimedOut, ConnectionReset] {
+            let e = poll_says_read(Err(kind.into()), PollFlags::empty()).unwrap_err();
+            assert_eq!(e.kind(), kind);
+        }
+    }
+
+    // A sender dribbling bytes with gaps many ticks long is read byte-identically.
+    #[test]
+    fn a_slow_sender_across_many_poll_ticks_arrives_byte_identical() {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap();
+        let data: Vec<u8> = (0..200_000u32).map(|i| (i * 7 + i / 251) as u8).collect();
+        let sent = data.clone();
+        let sender = std::thread::spawn(move || {
+            let mut s = TcpStream::connect(addr).unwrap();
+            let mut off = 0;
+            for i in 0usize.. {
+                if off == sent.len() {
+                    break;
+                }
+                let n = (1 + i * 37 % 4093).min(sent.len() - off);
+                s.write_all(&sent[off..off + n]).unwrap();
+                off += n;
+                if i % 8 == 0 {
+                    std::thread::sleep(std::time::Duration::from_millis(6));
+                }
+            }
+        });
+        let (stream, _) = listener.accept().unwrap();
+        let tick = std::time::Duration::from_millis(1);
+        let mut r = HaltRead::with_tick(stream, Some(crate::halt::Halt::new()), tick);
+        let mut got = Vec::new();
+        let mut small = [0u8; 1500];
+        loop {
+            match r.read(&mut small).unwrap() {
+                0 => break,
+                n => got.extend_from_slice(&small[..n]),
+            }
+        }
+        sender.join().unwrap();
+        assert_eq!(got.len(), data.len());
+        assert!(
+            got == data,
+            "stream bytes must survive the poll ticks unchanged"
+        );
+    }
+
+    // A halt-aware receiver arms keepalive (dead-peer detection) and sets no
+    // data-idle limit: a sender silent for a while is still read afterwards.
+    #[test]
+    fn a_halt_aware_receiver_uses_keepalive_not_an_idle_limit() {
+        use crate::pes;
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap();
+        let handle = std::thread::spawn(move || {
+            let halt = crate::halt::Halt::new();
+            let mut ns = NetworkStream::accept_from_with_halt(listener, Some(halt)).unwrap();
+            let keepalive = match &ns.mode {
+                Mode::Read { reader, .. } => socket2::SockRef::from(&reader.get_ref().stream)
+                    .keepalive()
+                    .unwrap(),
+                Mode::Write { .. } => false,
+            };
+            (
+                keepalive,
+                pes::Stream::read(&mut ns).map(|f| f.map(|f| f.data)),
+            )
+        });
+        let mut writer = NetworkStream::connect_vetted(&addr.to_string(), false)
+            .unwrap()
+            .meta(&sample_title());
+        if let Mode::Write {
+            writer: w,
+            header_written,
+            padded,
+            ..
+        } = &mut writer.mode
+        {
+            ensure_header_written(w, header_written, &sample_title(), &[], padded).unwrap();
+            w.flush().unwrap();
+        }
+        std::thread::sleep(std::time::Duration::from_millis(700));
+        let frame = pes::PesFrame {
+            discard_padding_ns: 0,
+            coding: None,
+            source: None,
+            track: 0,
+            pts: 0,
+            keyframe: true,
+            data: vec![7; 4],
+            duration_ns: None,
+        };
+        pes::Stream::write(&mut writer, &frame).unwrap();
+        pes::Stream::finish(&mut writer).unwrap();
+        let (keepalive, read) = handle.join().unwrap();
+        assert!(keepalive, "keepalive must be armed");
+        assert_eq!(read.unwrap(), Some(vec![7; 4]));
+    }
+
+    // The FMKV header's codec privates (e.g. FLAC STREAMINFO, OpusHead, avcC)
+    // must be exposed on the receive side, or the MKV writer drops them.
+    #[test]
+    fn the_receiver_exposes_the_header_codec_privates() {
+        use crate::pes;
+        let mut title = sample_title();
+        title.codec_privates = vec![Some(vec![1, 2, 3]), Some(b"OpusHead".to_vec())];
+        let (addr, handle) = spawn_codec_private_reader();
+        let mut writer = NetworkStream::connect_vetted(&addr.to_string(), false)
+            .unwrap()
+            .meta(&title);
+        pes::Stream::finish(&mut writer).unwrap();
+        let cps = handle.join().unwrap();
+        assert_eq!(cps, title.codec_privates);
+    }
+
+    fn spawn_codec_private_reader() -> (
+        std::net::SocketAddr,
+        std::thread::JoinHandle<Vec<Option<Vec<u8>>>>,
+    ) {
+        use crate::pes::Stream as _;
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap();
+        let handle = std::thread::spawn(move || {
+            let ns = NetworkStream::accept_from(listener).unwrap();
+            (0..2).map(|i| ns.codec_private(i)).collect()
+        });
+        (addr, handle)
+    }
+
+    // mkv(opus) -> network -> mkv: CodecDelay/SeekPreRoll and DiscardPadding
+    // survive the PES wire hop.
+    #[test]
+    fn opus_track_timing_and_padding_survive_network_to_mkv() {
+        use crate::pes::{self, Stream as _};
+        let mut title = sample_title();
+        title.streams.truncate(1);
+        title.streams[0] = Stream::Audio(AudioStream {
+            pid: 0x1100,
+            codec: Codec::Opus,
+            channels: AudioChannels::Stereo,
+            language: "eng".into(),
+            sample_rate: SampleRate::S48,
+            secondary: false,
+            purpose: crate::disc::LabelPurpose::Normal,
+            label: String::new(),
+        });
+        title.codec_privates = vec![Some(b"OpusHead\x01\x02".to_vec())];
+        let timing = pes::TrackTiming {
+            codec_delay_ns: 6_500_000,
+            seek_preroll_ns: 80_000_000,
+        };
+        let frame = pes::PesFrame {
+            discard_padding_ns: -2_500_000,
+            coding: None,
+            source: None,
+            track: 0,
+            pts: 0,
+            keyframe: true,
+            data: vec![0xFC, 1, 2, 3],
+            duration_ns: Some(20_000_000),
+        };
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap();
+        let handle = std::thread::spawn(move || {
+            let mut ns = NetworkStream::accept_from(listener).unwrap();
+            let got = ns.track_timing(0);
+            let f = ns.read().unwrap().unwrap();
+            (ns.info().clone(), got, f)
+        });
+        let mut w = NetworkStream::connect_vetted(&addr.to_string(), false)
+            .unwrap()
+            .meta(&title);
+        w.set_track_timing(0, timing).unwrap();
+        w.write(&frame).unwrap();
+        w.finish().unwrap();
+        let (rx_title, rx_timing, rx_frame) = handle.join().unwrap();
+        assert_eq!(rx_timing, timing);
+        assert_eq!(rx_frame.discard_padding_ns, frame.discard_padding_ns);
+        assert_eq!(rx_frame.data, frame.data);
+
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("out.mkv");
+        let file = std::fs::File::create(&path).unwrap();
+        let mut mkv =
+            crate::mux::mkvstream::MkvStream::create(Box::new(file), &rx_title, None).unwrap();
+        mkv.set_track_timing(0, rx_timing).unwrap();
+        mkv.write(&rx_frame).unwrap();
+        mkv.finish().unwrap();
+        let mut back =
+            crate::mux::mkvstream::MkvStream::open(std::fs::File::open(&path).unwrap()).unwrap();
+        assert_eq!(back.track_timing(0), timing);
+        assert_eq!(back.read().unwrap().unwrap().discard_padding_ns, -2_500_000);
+    }
 }

src/mux/pipelined_stream.rs

@@ -25,6 +25,8 @@ pub struct PipelinedPesStream {
     parsers: Vec<(u16, Box<dyn CodecParser>)>,
     pid_to_track: Vec<(u16, usize)>,
 
+    /// Declared BEFORE `demux_thread`: dropped first, so a worker blocked on a
+    /// full channel exits before `DemuxThread::drop` joins it.
     demux_rx: Receiver<DemuxBatch>,
     /// Kept alive so dropping this stream joins the demux + producer
     /// workers deterministically. Never poked directly after spawn.
@@ -412,6 +414,8 @@ impl Stream for PipelinedPesStream {
             .iter()
             .find(|(p, _)| *p == pid)
             .and_then(|(_, parser)| parser.codec_private())
+            // FLAC/Opus ES carry no init data: keep what the source header supplied.
+            .or_else(|| self.title.codec_privates.get(track).cloned().flatten())
     }
 
     // `lost_bytes` uses the trait default (0): the file-backed highway has no
@@ -1030,6 +1034,24 @@ mod tests {
         );
     }
 
+    /// A codec whose ES carries no init data (FLAC, Opus) keeps the CodecPrivate
+    /// the source header supplied (FMKV `m2ts://`) when its parser has none.
+    #[test]
+    fn codec_private_falls_back_to_the_source_title() {
+        let mut title = video_title(false);
+        title.codec_privates = vec![Some(b"fLaC".to_vec())];
+        let parsers: Vec<(u16, Box<dyn CodecParser>)> = vec![(
+            0x1011,
+            Box::new(CountingParser {
+                per_pes: 0,
+                flush_n: 0,
+                cp: None,
+            }),
+        )];
+        let (stream, _tx) = make_stream(title, parsers, vec![(0x1011u16, 0usize)]);
+        assert_eq!(stream.codec_private(0).as_deref(), Some(&b"fLaC"[..]));
+    }
+
     /// codec_private(track) returns None for a track index not present in
     /// pid_to_track — no panic, no wrong-track data.
     #[test]

src/mux/resolve.rs

@@ -356,6 +356,15 @@ impl std::fmt::Debug for InputOptions {
 
 /// Open a PES input stream (produces PES frames).
 pub fn input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn crate::pes::Stream>> {
+    input_with_halt(url, opts, None)
+}
+
+// `input` whose blocking network receive (`network://`) a halt can interrupt.
+pub(crate) fn input_with_halt(
+    url: &str,
+    opts: &InputOptions,
+    halt: Option<&crate::halt::Halt>,
+) -> io::Result<Box<dyn crate::pes::Stream>> {
     let parsed = parse_url(url);
     match parsed {
         StreamUrl::Disc { .. } => {
@@ -415,7 +424,10 @@ pub fn input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn crate::pes::S
         }
         StreamUrl::Network { ref addr } => {
             validate_network_addr(addr)?;
-            Ok(Box::new(NetworkStream::listen(addr)?))
+            Ok(Box::new(NetworkStream::listen_with_halt(
+                addr,
+                halt.cloned(),
+            )?))
         }
         StreamUrl::Stdio => Ok(Box::new(StdioStream::input())),
         StreamUrl::Null => Err(crate::error::Error::StreamWriteOnly.into()),

src/mux/stdio.rs

@@ -21,6 +21,11 @@ pub struct StdioStream {
     /// when no header was present — `headers_ready()` must gate on the
     /// metadata actually being available, not merely on having looked.
     meta_parsed: bool,
+    /// Write side: per-track decoder timing, sent in the header. Read side: the
+    /// header's timing.
+    timings: Vec<crate::pes::TrackTiming>,
+    /// Frames carry the DiscardPadding extension (FMKV header v2).
+    padded: bool,
 }
 
 impl StdioStream {
@@ -33,6 +38,8 @@ impl StdioStream {
             header_written: false,
             header_read: false,
             meta_parsed: false,
+            timings: Vec::new(),
+            padded: false,
         }
     }
 
@@ -45,6 +52,8 @@ impl StdioStream {
             header_written: false,
             header_read: false,
             meta_parsed: false,
+            timings: Vec::new(),
+            padded: false,
         }
     }
 
@@ -54,8 +63,9 @@ impl StdioStream {
         if let Some(w) = &mut self.writer
             && !self.header_written
         {
-            let m = meta::M2tsMeta::from_title(&self.disc_title);
+            let m = meta::M2tsMeta::from_title(&self.disc_title).with_timings(&self.timings);
             meta::write_header(w, &m)?;
+            self.padded = m.frame_padding;
             self.header_written = true;
         }
         Ok(())
@@ -73,6 +83,10 @@ impl StdioStream {
             // garbage. Ok(None) (magic mismatch / clean EOF) stays non-error (default title).
             if let Some(m) = meta::read_header(r)? {
                 self.disc_title = m.to_title();
+                self.timings = (0..self.disc_title.streams.len())
+                    .map(|i| m.timing(i))
+                    .collect();
+                self.padded = m.frame_padding;
                 self.meta_parsed = true;
             }
         }
@@ -84,7 +98,7 @@ impl crate::pes::Stream for StdioStream {
     fn read(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
         self.ensure_header_read()?;
         match &mut self.reader {
-            Some(r) => crate::pes::PesFrame::deserialize(r),
+            Some(r) => crate::pes::PesFrame::deserialize_ext(r, self.padded),
             None => Err(crate::error::Error::StreamWriteOnly.into()),
         }
     }
@@ -94,7 +108,7 @@ impl crate::pes::Stream for StdioStream {
         }
         self.ensure_header_written()?;
         match &mut self.writer {
-            Some(w) => frame.serialize(w),
+            Some(w) => frame.serialize_ext(w, self.padded),
             None => Err(crate::error::Error::StreamReadOnly.into()),
         }
     }
@@ -112,6 +126,26 @@ impl crate::pes::Stream for StdioStream {
         &self.disc_title
     }
 
+    // Read side: the header's timing, available once the first read parsed it.
+    fn track_timing(&self, track: usize) -> crate::pes::TrackTiming {
+        self.timings.get(track).copied().unwrap_or_default()
+    }
+
+    fn set_track_timing(
+        &mut self,
+        track: usize,
+        timing: crate::pes::TrackTiming,
+    ) -> io::Result<()> {
+        if self.writer.is_none() {
+            return Err(crate::error::Error::StreamReadOnly.into());
+        }
+        if self.header_written {
+            return Err(crate::error::Error::StreamHeaderWritten.into());
+        }
+        let tracks = self.disc_title.streams.len();
+        super::network::set_timing(&mut self.timings, track, timing, tracks)
+    }
+
     fn codec_private(&self, track: usize) -> Option<Vec<u8>> {
         // Single source of truth: the title's own codec_privates. (The
         // previous `stored_codec_privates` field was a redundant clone

src/mux/ts.rs

@@ -611,6 +611,37 @@ fn psi_payload_base(pkt: &[u8]) -> Option<usize> {
     }
 }
 
+// MPEG audio layer (1-3) from the first frame header at the start of `pid`'s
+// first PES in `data`; `None` when no such PES or header is in the scan window.
+fn mpeg_audio_layer(data: &[u8], pid: u16) -> Option<u8> {
+    let mut offset = 0;
+    while offset + BD_SOURCE_PACKET_BYTES <= data.len() {
+        if !is_resync_point(data, offset) {
+            offset += 1;
+            continue;
+        }
+        let pkt = &data[offset..offset + BD_SOURCE_PACKET_BYTES];
+        let pkt_pid = (((pkt[5] & 0x1F) as u16) << 8) | pkt[6] as u16;
+        if pkt_pid == pid && pkt[5] & 0x40 != 0 {
+            let pes = &pkt[psi_payload_base(pkt)?..];
+            if pes.get(..3)? != [0, 0, 1] {
+                return None;
+            }
+            let es = pes.get(9 + *pes.get(8)? as usize..)?;
+            let (b0, b1) = (*es.first()?, *es.get(1)?);
+            if b0 != 0xFF || b1 & 0xE0 != 0xE0 {
+                return None;
+            }
+            return match (b1 >> 1) & 0x03 {
+                0 => None,
+                l => Some(4 - l),
+            };
+        }
+        offset += BD_SOURCE_PACKET_BYTES;
+    }
+    None
+}
+
 // Reassemble a single PSI section (PAT/PMT) for `target_pid`/`table_id` across TS-packet
 // boundaries. Returns the section bytes (from table_id) or None.
 fn collect_psi_section(data: &[u8], target_pid: u16, table_id: u8) -> Option<Vec<u8>> {
@@ -767,6 +798,8 @@ pub fn scan_streams(data: &[u8]) -> Option<Vec<crate::disc::Stream>> {
             // PMTs may also carry ISO/IEC 13818-1 audio stream types that are
             // absent from Blu-ray's STN table. Keep that distinction local to TS.
             let codec = match stream_type {
+                // MPEG-1/2 audio covers Layers I-III; only the ES says which.
+                0x03 | 0x04 if mpeg_audio_layer(data, es_pid) == Some(3) => Codec::Mp3,
                 0x03 | 0x04 => Codec::Mp2,
                 0x0f => Codec::Aac,
                 0x87 => Codec::Ac3Plus,
@@ -1416,6 +1449,26 @@ mod tests {
         }
     }
 
+    // stream_type 0x03/0x04 covers Layers I-III; the layer comes from the ES.
+    #[test]
+    fn scan_streams_labels_mpeg_audio_by_the_es_layer() {
+        use crate::disc::{Codec, Stream};
+        for (header, expected) in [([0xFF, 0xFB], Codec::Mp3), ([0xFF, 0xFD], Codec::Mp2)] {
+            let mut pes = vec![0x00, 0x00, 0x01, 0xC0, 0x00, 0x00, 0x80, 0x00, 0x00];
+            pes.extend_from_slice(&header);
+            pes.extend_from_slice(&[0x90, 0x64]);
+            let mut data = pat_packet(0x100);
+            data.extend(pmt_packet(0x100, &[(0x1b, 0x1011), (0x03, 0x1100)]));
+            data.extend(es_packet_exact(0x1100, true, &pes));
+            data.extend(pat_packet(0x100));
+            let streams = scan_streams(&data).unwrap();
+            assert!(
+                matches!(&streams[1], Stream::Audio(a) if a.codec == expected),
+                "{expected:?}"
+            );
+        }
+    }
+
     #[test]
     fn scan_streams_maps_lpcm_via_from_coding_type() {
         use crate::disc::{Codec, Stream};

src/mux/tsmux.rs

@@ -27,6 +27,10 @@ fn is_video_pid(pid: u16) -> bool {
     VIDEO_PID_RANGE.contains(&pid)
 }
 
+// Headroom below the first frame's PTS for frames presented before it but emitted
+// after it. ISO/IEC 13818-1 §2.4.2.6 caps T-STD buffer delay at 1 s.
+pub(crate) const ORIGIN_HEADROOM_NS: i64 = 1_000_000_000;
+
 /// BD-TS muxer: PES frames in, 192-byte BD-TS packets out.
 ///
 /// Constructed over an output writer and a slice of per-track PIDs. The `track` index passed to
@@ -45,8 +49,8 @@ pub struct TsMuxer<W: Write> {
     video_codec: Vec<Codec>,
     /// Global PTS origin (nanoseconds), seeded by the FIRST frame of any kind
     /// (video or audio) so a single fixed origin rebases every frame and the
-    /// audio/video offset is preserved. A frame earlier than the origin (e.g.
+    /// audio/video offset is preserved. It lies [`ORIGIN_HEADROOM_NS`] before
-    /// audio reordered before the seeding frame) saturates to 0.
+    /// the seeding frame; only a frame earlier than that saturates to 0.
     base_pts_ns: Option<i64>,
     /// Count of PES frames actually emitted (a frame dropped as non-key
     /// before the first keyframe does NOT count). `finish()` returns
@@ -153,10 +157,12 @@ impl<W: Write> TsMuxer<W> {
             return Ok(());
         }
 
-        // Seed the global PTS origin from the FIRST frame of ANY kind, so one fixed
+        // One origin for every track, seeded by the FIRST frame of any kind and set
-        // origin rebases every frame and the a/v offset is preserved. Video-ONLY seeding
+        // ORIGIN_HEADROOM_NS below it, so a frame emitted later but presented
-        // made pre-video audio fall back to `base = unwrap_or(pts_ns)` (own pts) → PTS 0, spacing lost.
+        // earlier (video behind its GOP) keeps its offset.
-        let base = *self.base_pts_ns.get_or_insert(pts_ns);
+        let base = *self
+            .base_pts_ns
+            .get_or_insert(pts_ns.saturating_sub(ORIGIN_HEADROOM_NS));
         let pts_ns = pts_ns.saturating_sub(base);
 
         // NAL video (HEVC/H.264): convert length-prefixed NALUs to Annex B and prepend
@@ -439,6 +445,9 @@ fn build_pes_header(pid: u16, pts_90k: Option<u64>, data_len: usize) -> Vec<u8>
 mod tests {
     use super::*;
 
+    // The origin headroom in 90 kHz ticks: the seeding frame's encoded PTS.
+    const H: u64 = ORIGIN_HEADROOM_NS as u64 * 9 / 100_000;
+
     use crate::consts::BD_SOURCE_PACKET_BYTES;
     const VIDEO_PID: u16 = 0x1011;
 
@@ -747,12 +756,12 @@ mod tests {
     #[test]
     fn av_offset_preserved_with_audio_before_first_video() {
         // Audio at t=0 before the first video keyframe at t=1s. The FIRST frame of any
-        // kind fixes the origin, so audio becomes origin (PTS 0), video lands 1s=90_000
+        // kind fixes the origin, so audio lands at the headroom, video 1s=90_000
         // ticks later. OLD video-only seeding made both fall back to own pts → collapse to 0, offset lost.
         let mut sink: Vec<u8> = Vec::new();
         {
             let mut mux = TsMuxer::new(&mut sink, &[VIDEO_PID, AUDIO_PID]);
-            // Audio frame first, at PTS 0 — this seeds the origin.
+            // Audio frame first, at t=0 — this seeds the origin.
             mux.write_frame(1, 0, false, &[0x0B, 0x77, 0x00, 0x00])
                 .unwrap();
             // Video keyframe at PTS 1s — 1s after the origin.
@@ -763,14 +772,12 @@ mod tests {
         let packets = parse_bd_ts(&sink);
         let video_pts = first_pts_90k(&packets, VIDEO_PID);
         let audio_pts = first_pts_90k(&packets, AUDIO_PID);
-        // The leading audio frame is the origin ⇒ its relative PTS is 0.
+        // The leading audio frame seeds the origin one headroom below itself.
+        assert_eq!(audio_pts, H, "the first (audio) frame seeds the origin");
+        // Video is 1s after the audio ⇒ the audio→video offset is preserved.
         assert_eq!(
-            audio_pts, 0,
+            video_pts,
-            "the first (audio) frame seeds the origin at 0"
+            H + 90_000,
-        );
-        // Video is 1s after the origin ⇒ the audio→video offset is preserved.
-        assert_eq!(
-            video_pts, 90_000,
             "video 1s after the origin must stay 90_000 ticks ahead, not collapse to 0"
         );
     }
@@ -1010,13 +1017,13 @@ mod tests {
     #[test]
     fn pts_encoded_at_90khz_decodes_correctly() {
         // pts_ns → 90 kHz ticks = pts_ns * 9 / 100_000. 1 second (1e9 ns)
-        // = 90_000 ticks. The first (base) video frame rebases to 0, so use
+        // = 90_000 ticks. The first (base) video frame lands at the headroom, so use
         // a second frame at a known offset and check its encoded PTS.
         let mut sink: Vec<u8> = Vec::new();
         {
             let mut mux = TsMuxer::new(&mut sink, &[VIDEO_PID]);
             let idr = fake_hevc_nal(19, 50);
-            mux.write_frame(0, 0, true, &idr).unwrap(); // base = 0
+            mux.write_frame(0, 0, true, &idr).unwrap(); // seeds the origin
             let p = fake_hevc_nal(1, 50);
             // +1 second relative to base.
             mux.write_frame(0, 1_000_000_000, false, &p).unwrap();
@@ -1035,7 +1042,7 @@ mod tests {
             | (((p[11] >> 1) as u64) << 15)
             | ((p[12] as u64) << 7)
             | ((p[13] >> 1) as u64);
-        assert_eq!(pts, 90_000, "1s offset encodes to 90000 ticks @ 90 kHz");
+        assert_eq!(pts, H + 90_000, "1s offset encodes to 90000 ticks @ 90 kHz");
     }
 
     #[test]
@@ -1082,8 +1089,8 @@ mod tests {
     #[test]
     fn negative_relative_pts_saturates_to_zero() {
         // A frame earlier than the origin (negative relative PTS) must encode PTS 0, not
-        // an underflowed huge value. Video keyframe at t=2s seeds the origin; a later
+        // an underflowed huge value. Video keyframe at t=2s seeds the origin at t=1s; a
-        // audio frame at t=0 is 2s BEFORE it → -2s, which must floor to 0.
+        // later audio frame at t=0 is 1s BEFORE it, which must floor to 0.
         let mut sink: Vec<u8> = Vec::new();
         {
             let mut mux = TsMuxer::new(&mut sink, &[VIDEO_PID, AUDIO_PID]);
@@ -1120,9 +1127,13 @@ mod tests {
         let packets = parse_bd_ts(&sink);
         let pts = all_pts_90k(&packets, AUDIO_PID);
         assert_eq!(pts.len(), 2, "two audio frames → two PTS-bearing PES");
-        assert_eq!(pts[0], 0, "the first audio frame is the origin → 0");
         assert_eq!(
-            pts[1], 90_000,
+            pts[0], H,
+            "the first audio frame sits one headroom past the origin"
+        );
+        assert_eq!(
+            pts[1],
+            H + 90_000,
             "the second frame (1s later) must keep its 1s spacing, not collapse to 0"
         );
     }
@@ -1341,10 +1352,30 @@ mod tests {
             | (((p[11] >> 1) as u64) << 15)
             | ((p[12] as u64) << 7)
             | ((p[13] >> 1) as u64);
-        assert_eq!(decoded, 0, "base video frame stays at relative PTS 0");
+        assert_eq!(decoded, H, "base video frame stays at the headroom");
         assert_eq!(
-            pts, big_pts_ticks,
+            pts,
+            H + big_pts_ticks,
             "the high bits (29..32) of a large PTS must round-trip through encoding"
         );
     }
+
+    // Audio emitted first but presented AFTER the first video frame (the video
+    // parser holds its opening GOP) must not clamp that video onto the audio's PTS.
+    #[test]
+    fn video_presented_before_the_seeding_audio_keeps_its_offset() {
+        let mut sink: Vec<u8> = Vec::new();
+        {
+            let mut mux = TsMuxer::new(&mut sink, &[VIDEO_PID, AUDIO_PID]);
+            mux.write_frame(1, 100_000_000, false, &[0x0B, 0x77, 0x00, 0x00])
+                .unwrap();
+            let idr = fake_hevc_nal(19, 100);
+            mux.write_frame(0, 50_000_000, true, &idr).unwrap();
+            mux.finish().unwrap();
+        }
+        let packets = parse_bd_ts(&sink);
+        let video = first_pts_90k(&packets, VIDEO_PID);
+        let audio = first_pts_90k(&packets, AUDIO_PID);
+        assert_eq!(audio - video, 4_500, "50 ms apart, video first");
+    }
 }

src/pes.rs

@@ -57,7 +57,7 @@ pub struct PesFrame {
     /// preserved across network:// and stdio:// hops.
     pub duration_ns: Option<u64>,
     /// Matroska DiscardPadding in nanoseconds: positive trims the end, negative
-    /// trims the beginning. Local container remux metadata; not on the PES wire.
+    /// trims the beginning. On the PES wire only for FMKV header v2 streams.
     pub discard_padding_ns: i64,
     /// Byte-exact source provenance of this frame's first byte, stamped at the
     /// demux seam. `None` for synthetic sources / the `skip_parse` path and for
@@ -81,6 +81,10 @@ impl PesFrame {
     ///
     /// `duration_ns` is encoded as `u64::MAX` when `None`, or the value when `Some`.
     pub fn serialize(&self, w: &mut dyn std::io::Write) -> std::io::Result<()> {
+        self.serialize_inner(w, false)
+    }
+
+    fn serialize_inner(&self, w: &mut dyn std::io::Write, padded: bool) -> std::io::Result<()> {
         if self.track > 255 {
             return Err(crate::error::Error::PesTrackTooLarge { track: self.track }.into());
         }
@@ -98,9 +102,34 @@ impl PesFrame {
         w.write_all(&[if self.keyframe { 1 } else { 0 }])?;
         w.write_all(&duration_wire.to_le_bytes())?;
         w.write_all(&(self.data.len() as u32).to_le_bytes())?;
+        if padded {
+            w.write_all(&self.discard_padding_ns.to_le_bytes())?;
+        }
         w.write_all(&self.data)
     }
 
+    /// [`serialize`](Self::serialize) plus, when `padded` (FMKV header v2), an
+    /// 8-byte LE `discard_padding_ns` after the fixed header.
+    pub(crate) fn serialize_ext(
+        &self,
+        w: &mut dyn std::io::Write,
+        padded: bool,
+    ) -> std::io::Result<()> {
+        self.serialize_inner(w, padded)
+    }
+
+    /// [`deserialize`](Self::deserialize) for a stream whose FMKV header says
+    /// frames carry the `discard_padding_ns` extension (`padded`).
+    pub(crate) fn deserialize_ext(
+        r: &mut dyn std::io::Read,
+        padded: bool,
+    ) -> std::io::Result<Option<Self>> {
+        if !padded {
+            return Self::deserialize(r);
+        }
+        Self::deserialize_inner(r, true)
+    }
+
     /// Deserialize from bytes. Returns None at a clean end of stream.
     ///
     /// A clean EOF is exactly zero bytes available before the next frame.
@@ -108,6 +137,10 @@ impl PesFrame {
     /// error (`UnexpectedEof`), not silently treated as EOF — otherwise
     /// truncated `.pes` data would be accepted as a graceful end.
     pub fn deserialize(r: &mut dyn std::io::Read) -> std::io::Result<Option<Self>> {
+        Self::deserialize_inner(r, false)
+    }
+
+    fn deserialize_inner(r: &mut dyn std::io::Read, padded: bool) -> std::io::Result<Option<Self>> {
         // Probe one byte first to distinguish clean EOF from a truncated header.
         // Loop on EINTR so a recoverable interrupted read doesn't fail — symmetric
         // with read_exact's internal retry for the rest of the header and data.
@@ -141,6 +174,12 @@ impl PesFrame {
             Some(duration_wire)
         };
         let len = u32::from_le_bytes([header[18], header[19], header[20], header[21]]) as usize;
+        let mut discard_padding_ns = 0;
+        if padded {
+            let mut ext = [0u8; 8];
+            r.read_exact(&mut ext)?;
+            discard_padding_ns = i64::from_le_bytes(ext);
+        }
         if len > MAX_FRAME_SIZE {
             return Err(crate::error::Error::PesFrameTooLarge { size: len }.into());
         }
@@ -157,7 +196,7 @@ impl PesFrame {
             filled += want;
         }
         Ok(Some(Self {
-            discard_padding_ns: 0,
+            discard_padding_ns,
             track,
             pts,
             keyframe,
@@ -226,8 +265,9 @@ pub trait Stream: Send {
         TrackTiming::default()
     }
 
-    /// Set timing before the first frame is written. Sinks without this metadata
+    /// Set timing before the first frame is written. Carried by `mkv://` and the
-    /// retain their existing behavior.
+    /// FMKV `network://` / `stdio://` wire; `m2ts://` has no field for it and
+    /// `mp4://` carries no Opus track, so those keep the no-op default.
     fn set_track_timing(&mut self, _track: usize, _timing: TrackTiming) -> std::io::Result<()> {
         Ok(())
     }