v0.21.2: bump version

Empirical regression observed during 0.21.1 mux test on rip1/unraid-1:
historical 0.20.7 baseline averaged ~18 MB/s mux throughput; 0.21.1
dropped to ~7-8 MB/s flat. Same NFS source + destination, same disc.

Suspect: 32 MiB FileSectorSource readahead + posix_fadvise(SEQUENTIAL)
together saturate the TCP connection on read bursts, starving the
writer thread's concurrent NFS writes (mux reads the source ISO and
writes the MKV over the same connection).

- READAHEAD_BUF_BYTES: 32 MiB -> 4 MiB. Matches NFS rsize=1 MiB * 4
  round-trips per refill, interleaves cleanly with writes.
- linux/hint_sequential: now no-op. Kernel's default ~128 KiB
  readahead is what we want on NFS-backed ISOs (the dominant case).
  Per-OS file stays so we can re-enable a hint cleanly later if a
  different path benefits.

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "libfreemkv"
-version = "0.20.9"
+version = "0.21.2"
 edition = "2024"
 rust-version = "1.86"
 license = "AGPL-3.0-only"

src/io/file_sector_source/linux.rs

@@ -1,15 +1,18 @@
-//! Linux: hint the kernel that this fd will be read sequentially so
-//! readahead widens. `posix_fadvise(POSIX_FADV_SEQUENTIAL)` is a hint,
-//! not a guarantee — the kernel still owns the policy decision.
+//! Linux: kernel readahead hint for the ISO file.
+//!
+//! Originally `POSIX_FADV_SEQUENTIAL` to widen the readahead window.
+//! On NFS that turned out to cause aggressive multi-MB readahead bursts
+//! that saturated the TCP connection and starved concurrent writes
+//! during mux — observed empirically as a ~3× drop in mux throughput
+//! on the rip1/unraid-1 setup (0.21.0 vs 0.20.7 baseline). The kernel's
+//! default readahead (~128 KiB on Linux) interleaves more naturally
+//! with the muxer's concurrent NFS writes, so we no longer issue any
+//! hint here. The per-OS file stays so the convention is honoured and
+//! we can re-enable a hint cleanly if a different storage path benefits.
 
 use std::fs::File;
-use std::os::unix::io::AsRawFd;
 
-pub(super) fn hint_sequential(file: &File, _len_bytes: u64) {
-    // Best-effort: return value ignored. A fadvise failure has no
-    // user-observable consequence (reads still work, just without the
-    // widened readahead window).
-    unsafe {
-        libc::posix_fadvise(file.as_raw_fd(), 0, 0, libc::POSIX_FADV_SEQUENTIAL);
-    }
+pub(super) fn hint_sequential(_file: &File, _len_bytes: u64) {
+    // No-op: see module-level comment. Kernel default readahead is
+    // what we want on NFS-backed ISOs, which is the dominant case.
 }

src/io/file_sector_source/mod.rs

@@ -65,8 +65,15 @@ use crate::sector::SectorSource;
 /// than per-sector pread, and large enough to coast through a typical
 /// NFS server commit blip.
 ///
+/// 0.21.2: shrunk from 32 MiB → 4 MiB. On NFS-backed ISOs with
+/// concurrent NFS writes (the mux phase), a 32 MiB refill bursts the
+/// TCP connection hard enough to starve the writer thread, observed
+/// empirically as a ~3× drop in sustained mux throughput on the
+/// rip1/unraid-1 setup. 4 MiB matches `rsize=1 MiB` × 4 round-trips
+/// and interleaves cleanly with writes.
+///
 /// Tweakable. Named const, not a magic number.
-pub const READAHEAD_BUF_BYTES: usize = 32 * 1024 * 1024;
+pub const READAHEAD_BUF_BYTES: usize = 4 * 1024 * 1024;
 
 const SECTOR_SIZE: usize = 2048;
 /// Sectors per refill: [`READAHEAD_BUF_BYTES`] / [`SECTOR_SIZE`]. The

src/io/mod.rs

@@ -26,6 +26,7 @@
 pub(crate) mod bounded;
 pub mod byte_channel;
 pub mod file_sector_source;
+pub mod sink;
 mod writeback;
 mod writeback_file;
 

src/io/sink/local_file.rs

@@ -0,0 +1,162 @@
+//! `LocalFileSink` — `BufWriter<File>` for the common local-disk case.
+//!
+//! Buffering: 4 MiB internal `BufWriter`. Sized to coalesce the small
+//! per-PES writes that come out of the muxer into kernel-page-aligned
+//! flushes without making the buffer big enough to matter for memory
+//! pressure on a single concurrent rip.
+//!
+//! `Seek` flushes the underlying `BufWriter` first; otherwise a seek
+//! could leapfrog buffered data and silently corrupt the file. This is
+//! the same shape `BufWriter` itself uses when it impls `Seek` in
+//! stdlib, and is necessary for MKV's seek-back operations (cluster
+//! size patch, Cues index, segment header backpatch) to land on the
+//! right offset.
+//!
+//! `RandomAccessSink` is satisfied via the blanket impl in
+//! [`super::mod`]; no explicit impl needed here.
+
+use std::fs::{File, OpenOptions};
+use std::io::{self, BufWriter, Seek, SeekFrom, Write};
+use std::path::Path;
+
+use super::preallocate;
+
+const BUFFER_BYTES: usize = 4 * 1024 * 1024;
+
+/// Random-access write sink for local disks.
+///
+/// Wraps a `BufWriter<File>` with a 4 MiB internal buffer and forwards
+/// `Write`/`Seek` so any call site that previously held a `File` or
+/// `WritebackFile` can drop this in. `finish()` flushes the buffer and
+/// runs `sync_all` on the underlying file so the caller can drop it
+/// without losing data.
+///
+/// Construction always opens the file `create + truncate + read +
+/// write`. `read` is enabled so the same handle can be reused for a
+/// verification re-read after the mux (the existing
+/// `FileSectorSink::create` pattern). On Linux, [`with_size_hint`]
+/// additionally calls `fallocate(FALLOC_FL_KEEP_SIZE)` to pre-reserve
+/// extents.
+///
+/// [`with_size_hint`]: Self::with_size_hint
+pub struct LocalFileSink {
+    inner: BufWriter<File>,
+}
+
+impl LocalFileSink {
+    /// Open `path` for writing, truncating any existing contents.
+    pub fn create(path: &Path) -> io::Result<Self> {
+        let file = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create(true)
+            .truncate(true)
+            .open(path)?;
+        Ok(Self {
+            inner: BufWriter::with_capacity(BUFFER_BYTES, file),
+        })
+    }
+
+    /// Like [`Self::create`] but additionally calls the per-OS
+    /// preallocate path with `size_bytes`. On Linux this is
+    /// `fallocate(FALLOC_FL_KEEP_SIZE)` so the on-disk extents are
+    /// reserved up front (reducing fragmentation for big sequential
+    /// muxer output); on other OSes it is a no-op today. Failures
+    /// from the preallocate call are non-fatal — the file is still
+    /// returned, just without the size reservation.
+    pub fn with_size_hint(path: &Path, size_bytes: u64) -> io::Result<Self> {
+        let file = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create(true)
+            .truncate(true)
+            .open(path)?;
+        preallocate::preallocate(&file, size_bytes);
+        Ok(Self {
+            inner: BufWriter::with_capacity(BUFFER_BYTES, file),
+        })
+    }
+
+    /// Drain the internal buffer and `fsync` the underlying file.
+    /// Idempotent with `Drop` (the `BufWriter` also flushes on drop;
+    /// this call additionally surfaces fsync errors to the caller).
+    #[allow(dead_code)] // exposed for parity with WritebackFile::sync_all
+    pub fn sync_all(&mut self) -> io::Result<()> {
+        self.inner.flush()?;
+        self.inner.get_ref().sync_all()
+    }
+}
+
+impl Write for LocalFileSink {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.inner.write(buf)
+    }
+
+    fn write_all(&mut self, buf: &[u8]) -> io::Result<()> {
+        self.inner.write_all(buf)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.inner.flush()
+    }
+}
+
+impl Seek for LocalFileSink {
+    fn seek(&mut self, from: SeekFrom) -> io::Result<u64> {
+        // Flush before seeking so buffered bytes land at the offset
+        // they were written for, not the new one.
+        self.inner.flush()?;
+        self.inner.get_mut().seek(from)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Read;
+
+    #[test]
+    fn write_seek_roundtrip() {
+        let dir = tempfile::tempdir().unwrap();
+        let p = dir.path().join("rt.bin");
+        let mut s = LocalFileSink::create(&p).unwrap();
+        s.write_all(b"AAAA").unwrap();
+        s.write_all(b"BBBB").unwrap();
+        // Seek back over the second word and overwrite.
+        s.seek(SeekFrom::Start(4)).unwrap();
+        s.write_all(b"CCCC").unwrap();
+        s.sync_all().unwrap();
+        drop(s);
+
+        let mut f = File::open(&p).unwrap();
+        let mut got = Vec::new();
+        f.read_to_end(&mut got).unwrap();
+        assert_eq!(&got[..], b"AAAACCCC");
+    }
+
+    #[test]
+    fn drop_flushes() {
+        let dir = tempfile::tempdir().unwrap();
+        let p = dir.path().join("drop.bin");
+        {
+            let mut s = LocalFileSink::create(&p).unwrap();
+            s.write_all(b"buffered").unwrap();
+            // No explicit flush / sync_all — BufWriter drop runs the
+            // flush and the file should land on disk.
+        }
+        let bytes = std::fs::read(&p).unwrap();
+        assert_eq!(&bytes[..], b"buffered");
+    }
+
+    #[test]
+    fn with_size_hint_creates_writable_file() {
+        let dir = tempfile::tempdir().unwrap();
+        let p = dir.path().join("sz.bin");
+        let mut s = LocalFileSink::with_size_hint(&p, 64 * 1024).unwrap();
+        s.write_all(b"hint-ok").unwrap();
+        s.sync_all().unwrap();
+        drop(s);
+        let bytes = std::fs::read(&p).unwrap();
+        assert_eq!(&bytes[..], b"hint-ok");
+    }
+}

src/io/sink/mod.rs

@@ -0,0 +1,161 @@
+//! Output-sink trait split for the buffering architecture.
+//!
+//! Two traits, one for each capability axis of an output destination:
+//!
+//! - [`SequentialSink`] — anything you can `Write` to in order. Sockets,
+//!   pipes, append-only stores, plain files. Containers that don't need
+//!   seek (M2TS, fMP4, HEVC elementary) target this.
+//! - [`RandomAccessSink`] — everything `SequentialSink` plus a working
+//!   `Seek`. Local files, NFS files, anything with random-write
+//!   semantics. Containers that need backpatch (MKV cluster sizes, Cues
+//!   index, MP4 moov-at-end) target this.
+//!
+//! `RandomAccessSink: SequentialSink` — every random-access sink is
+//! also a valid sequential sink. The muxer is generic over which it
+//! requires (`MkvMux<S: RandomAccessSink>`, `M2tsMux<S: SequentialSink>`)
+//! so an attempt to mux MKV to a network socket is a compile error.
+//!
+//! Buffering policy belongs to the concrete sink, not to a wrapper at
+//! the call site. `LocalFileSink` wraps a `BufWriter<File>` with a
+//! 4 MiB buffer for the common local-disk case; `WritebackFile`
+//! (separate module) wraps a `File` with the adaptive-chunk
+//! `sync_file_range` machinery for the Linux+NFS case.
+//!
+//! See `(internal)/memory/project_buffering_architecture.md` for
+//! the full design and the source/sink matrix.
+
+use std::io::{Seek, Write};
+
+mod local_file;
+mod preallocate;
+mod socket;
+
+pub use local_file::LocalFileSink;
+pub use socket::{SocketSink, UdpSocketSink};
+
+/// Sequential-only write destination. Sockets, pipes, append-only
+/// stores. No seek. Implementations own their write buffering — the
+/// trait does not impose or hide any buffering of its own.
+///
+/// `finish` drains any internal buffering and signals end-of-stream to
+/// the underlying transport (close-write on a socket, flush on a
+/// buffered writer, etc.). The default impl is a no-op; concrete
+/// implementations that need explicit shutdown can override it but the
+/// blanket impl below keeps it optional for adapter types like
+/// `&mut File`.
+pub trait SequentialSink: Write + Send {
+    fn finish(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+/// Random-access write destination. Local files, NFS files, anything
+/// with a working `Seek`. Inherits the `SequentialSink` contract — a
+/// random-access sink is always usable as a sequential sink.
+pub trait RandomAccessSink: SequentialSink + Seek {}
+
+// Blanket impls so any `Write + Send` type acts as a `SequentialSink`
+// (with default `finish`), and any sink that also impls `Seek` is
+// automatically a `RandomAccessSink`. Keeps call-site ergonomics simple
+// — `&mut File`, `LocalFileSink`, `WritebackFile`, `BufWriter<File>`,
+// and `Cursor<Vec<u8>>` all satisfy the right trait without per-type
+// boilerplate.
+impl<T: Write + Send + ?Sized> SequentialSink for T {}
+impl<T: SequentialSink + Seek + ?Sized> RandomAccessSink for T {}
+
+/// Pick the right `RandomAccessSink` impl for `dest` based on its
+/// filesystem type.
+///
+/// - Linux + NFS path → `WritebackFile` with its adaptive-chunk
+///   sync_file_range machinery and (when supported) `fallocate` size
+///   hint.
+/// - everything else → [`LocalFileSink`] over `BufWriter<File>`. On
+///   non-Linux there is no `WritebackFile` machinery to opt into, and
+///   on local Linux the kernel's default writeback policy is already
+///   fine.
+///
+/// `size_hint`, when present, is forwarded to the per-OS preallocate
+/// path (`fallocate(KEEP_SIZE)` on Linux, `F_PREALLOCATE` on macOS when
+/// implemented, no-op elsewhere).
+///
+/// Returns a boxed trait object so the call site (mux construction)
+/// stays agnostic of which concrete sink got picked.
+#[allow(dead_code)] // wiring to mux::resolve is a follow-up commit
+pub fn open_for_mkv(
+    dest: &std::path::Path,
+    size_hint: Option<u64>,
+) -> std::io::Result<Box<dyn RandomAccessSink>> {
+    #[cfg(not(target_os = "linux"))]
+    use crate::platform::fs_type::detect;
+    #[cfg(target_os = "linux")]
+    use crate::platform::fs_type::{FsType, detect};
+
+    #[cfg(target_os = "linux")]
+    {
+        if detect(dest) == FsType::Nfs {
+            let wf = match size_hint {
+                Some(n) => crate::io::WritebackFile::create_with_size_hint(dest, n)?,
+                None => crate::io::WritebackFile::create(dest)?,
+            };
+            return Ok(Box::new(wf));
+        }
+    }
+    // Silence the unused-binding warning on non-Linux where the only
+    // branch above is cfg-gated out.
+    #[cfg(not(target_os = "linux"))]
+    {
+        let _ = detect(dest);
+    }
+
+    let sink = match size_hint {
+        Some(n) => LocalFileSink::with_size_hint(dest, n)?,
+        None => LocalFileSink::create(dest)?,
+    };
+    Ok(Box::new(sink))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::fs::File;
+
+    // Type-level assertion: the blanket impls cover the shapes we care
+    // about. These functions never run; they just have to type-check.
+    fn _assert_file_is_sequential(_: &mut dyn SequentialSink) {}
+    fn _assert_file_is_random_access(_: &mut dyn RandomAccessSink) {}
+
+    #[test]
+    fn blanket_impls_cover_file_and_localfilesink() {
+        // `File` directly via blanket impls.
+        let dir = tempfile::tempdir().unwrap();
+        let mut f = File::create(dir.path().join("a.bin")).unwrap();
+        _assert_file_is_sequential(&mut f);
+        _assert_file_is_random_access(&mut f);
+
+        // `LocalFileSink` ditto.
+        let mut s = LocalFileSink::create(&dir.path().join("b.bin")).unwrap();
+        _assert_file_is_sequential(&mut s);
+        _assert_file_is_random_access(&mut s);
+
+        // `WritebackFile` — confirms the Phase 1 type still satisfies
+        // the trait via the blanket impl without needing an explicit
+        // `impl RandomAccessSink for WritebackFile {}`.
+        let mut wf = crate::io::WritebackFile::create(&dir.path().join("c.bin")).unwrap();
+        _assert_file_is_sequential(&mut wf);
+        _assert_file_is_random_access(&mut wf);
+    }
+
+    #[test]
+    fn open_for_mkv_returns_a_random_access_sink() {
+        let dir = tempfile::tempdir().unwrap();
+        let p = dir.path().join("c.bin");
+        let mut sink = open_for_mkv(&p, Some(64 * 1024)).unwrap();
+        use std::io::{Seek, SeekFrom, Write};
+        sink.write_all(b"hello").unwrap();
+        sink.seek(SeekFrom::Start(0)).unwrap();
+        sink.finish().unwrap();
+        drop(sink);
+        let bytes = std::fs::read(&p).unwrap();
+        assert_eq!(&bytes[..5], b"hello");
+    }
+}

src/io/sink/preallocate/linux.rs

@@ -0,0 +1,20 @@
+//! Linux `fallocate(FALLOC_FL_KEEP_SIZE)` preallocation.
+//!
+//! `KEEP_SIZE` reserves extents without changing the apparent file
+//! length, which matches the muxer's expectation that writes still grow
+//! the file naturally.
+
+use std::fs::File;
+#[cfg(unix)]
+use std::os::unix::io::AsRawFd;
+
+pub(super) fn preallocate_impl(file: &File, size_bytes: u64) {
+    let fd = file.as_raw_fd();
+    // FALLOC_FL_KEEP_SIZE = 0x01.
+    let rc = unsafe { libc::fallocate(fd, libc::FALLOC_FL_KEEP_SIZE, 0, size_bytes as i64) };
+    tracing::debug!(
+        target: "mux",
+        "LocalFileSink fallocate size_hint={size_bytes} rc={rc} ok={}",
+        rc == 0
+    );
+}

src/io/sink/preallocate/macos.rs

@@ -0,0 +1,48 @@
+//! macOS `F_PREALLOCATE` extent reservation.
+//!
+//! `fcntl(F_PREALLOCATE)` with `F_ALLOCATECONTIG` first (try for a
+//! contiguous run) and fall back to `F_ALLOCATEALL` (non-contig OK).
+//! Reported file size is unchanged — the muxer's writes still grow it.
+
+use std::fs::File;
+use std::os::unix::io::AsRawFd;
+
+// Mirror the Darwin `fstore_t` struct from `<sys/fcntl.h>`. libc on
+// some Rust toolchains/versions doesn't ship this binding, so define
+// it locally with the layout the kernel ABI requires.
+#[repr(C)]
+struct Fstore {
+    fst_flags: libc::c_uint,
+    fst_posmode: libc::c_int,
+    fst_offset: libc::off_t,
+    fst_length: libc::off_t,
+    fst_bytesalloc: libc::off_t,
+}
+
+// Constants from <sys/fcntl.h>.
+const F_PREALLOCATE: libc::c_int = 42;
+const F_ALLOCATECONTIG: libc::c_uint = 0x0000_0002;
+const F_ALLOCATEALL: libc::c_uint = 0x0000_0004;
+const F_PEOFPOSMODE: libc::c_int = 3;
+
+pub(super) fn preallocate_impl(file: &File, size_bytes: u64) {
+    let fd = file.as_raw_fd();
+    let mut store = Fstore {
+        fst_flags: F_ALLOCATECONTIG,
+        fst_posmode: F_PEOFPOSMODE,
+        fst_offset: 0,
+        fst_length: size_bytes as libc::off_t,
+        fst_bytesalloc: 0,
+    };
+    let mut rc = unsafe { libc::fcntl(fd, F_PREALLOCATE, &mut store as *mut Fstore) };
+    if rc == -1 {
+        // Fall back to non-contiguous.
+        store.fst_flags = F_ALLOCATEALL;
+        rc = unsafe { libc::fcntl(fd, F_PREALLOCATE, &mut store as *mut Fstore) };
+    }
+    tracing::debug!(
+        target: "mux",
+        "LocalFileSink F_PREALLOCATE size_hint={size_bytes} rc={rc} bytesalloc={}",
+        store.fst_bytesalloc
+    );
+}

src/io/sink/preallocate/mod.rs

@@ -0,0 +1,27 @@
+//! Per-OS extent preallocation. Best-effort; failures are logged at
+//! debug and otherwise swallowed because the file is still usable
+//! without the size reservation — only large-file fragmentation gets
+//! marginally worse.
+
+use std::fs::File;
+
+#[cfg(target_os = "linux")]
+mod linux;
+#[cfg(target_os = "macos")]
+mod macos;
+#[cfg(not(any(target_os = "linux", target_os = "macos")))]
+mod other;
+
+#[cfg(target_os = "linux")]
+use linux::preallocate_impl;
+#[cfg(target_os = "macos")]
+use macos::preallocate_impl;
+#[cfg(not(any(target_os = "linux", target_os = "macos")))]
+use other::preallocate_impl;
+
+/// Reserve `size_bytes` of disk space for `file`'s on-disk extents.
+/// Reported file size is unchanged — writes still grow the file
+/// naturally; only the allocator's extent map is primed.
+pub(super) fn preallocate(file: &File, size_bytes: u64) {
+    preallocate_impl(file, size_bytes);
+}

src/io/sink/preallocate/other.rs

@@ -0,0 +1,10 @@
+//! Fallback preallocate impl. No-op.
+
+use std::fs::File;
+
+pub(super) fn preallocate_impl(_file: &File, size_bytes: u64) {
+    tracing::debug!(
+        target: "mux",
+        "LocalFileSink preallocate size_hint={size_bytes} skipped (no platform impl)"
+    );
+}

src/io/sink/socket.rs

@@ -0,0 +1,278 @@
+//! TCP / UDP socket sinks (sequential-only).
+//!
+//! [`SocketSink`] wraps a `TcpStream` in a 1 MiB `BufWriter`. Constructor
+//! tunes `SO_SNDBUF` to a caller hint when provided. `finish()` flushes
+//! the buffer then `shutdown(Write)`s the socket so the peer sees clean
+//! end-of-stream.
+//!
+//! [`UdpSocketSink`] wraps a connected `UdpSocket`. Each `write` call
+//! emits exactly one datagram — the caller is responsible for packetizing
+//! to a reasonable MTU (188 × 7 = 1316 bytes for MPEG-TS-over-UDP is the
+//! conventional choice). `finish()` is a no-op; UDP has no end-of-stream
+//! marker.
+//!
+//! Both types satisfy [`SequentialSink`] via the blanket impl in
+//! `super::mod`. Neither implements `Seek`, so neither satisfies
+//! [`RandomAccessSink`] — using one with `MkvMux` is a compile error,
+//! which is the design intent.
+//!
+//! [`SequentialSink`]: super::SequentialSink
+//! [`RandomAccessSink`]: super::RandomAccessSink
+
+use std::io::{self, BufWriter, Write};
+use std::net::{Shutdown, TcpStream, ToSocketAddrs, UdpSocket};
+
+/// `BufWriter` capacity for [`SocketSink`]. 1 MiB matches the typical
+/// kernel send-buffer ceiling and keeps small-write amplification from
+/// containers (TS = 188-byte packets, fMP4 fragment headers = ~100 bytes)
+/// from translating into syscall storms.
+const TCP_BUF_CAPACITY: usize = 1024 * 1024;
+
+/// Sequential-only sink over a TCP connection.
+///
+/// Wraps a `BufWriter<TcpStream>`; the inner `TcpStream` is kept as a
+/// clone so [`finish`](Self::finish) can call `shutdown(Write)` after
+/// flushing the buffer (the buffered writer doesn't expose the socket
+/// directly).
+pub struct SocketSink {
+    /// Buffered write half. All payload bytes go through this.
+    buf: BufWriter<TcpStream>,
+    /// Shutdown handle — clone of the socket inside `buf`. Used only by
+    /// `finish()` for `shutdown(Write)`; never read or written through.
+    shutdown_handle: TcpStream,
+}
+
+impl SocketSink {
+    /// Open a TCP connection to `addr` and wrap it for sequential
+    /// writing. `sndbuf_bytes`, when present, is forwarded to
+    /// `setsockopt(SO_SNDBUF)` as a kernel hint — the OS may clamp it.
+    ///
+    /// `addr` accepts anything `ToSocketAddrs` does: `"10.0.0.1:1234"`,
+    /// `("host", 1234)`, a `SocketAddr`, etc.
+    pub fn connect<A: ToSocketAddrs>(addr: A, sndbuf_bytes: Option<usize>) -> io::Result<Self> {
+        let stream = TcpStream::connect(addr)?;
+        // `set_nodelay(true)` keeps small writes (TS packet trains, fMP4
+        // moof headers) from sitting in Nagle's algorithm until the buffer
+        // fills. The BufWriter already absorbs syscall overhead; Nagle
+        // would just add latency without coalescing more.
+        stream.set_nodelay(true)?;
+        if let Some(n) = sndbuf_bytes {
+            set_send_buffer(&stream, n)?;
+        }
+        let shutdown_handle = stream.try_clone()?;
+        Ok(Self {
+            buf: BufWriter::with_capacity(TCP_BUF_CAPACITY, stream),
+            shutdown_handle,
+        })
+    }
+}
+
+impl Write for SocketSink {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.buf.write(buf)
+    }
+    fn flush(&mut self) -> io::Result<()> {
+        self.buf.flush()
+    }
+}
+
+impl SocketSink {
+    /// Drain the BufWriter and `shutdown(Write)` the underlying socket
+    /// so the peer sees a clean EOF.
+    ///
+    /// Note: [`SequentialSink::finish`](super::SequentialSink::finish)'s
+    /// blanket-impl default is a no-op. Trait-object call sites that
+    /// need socket shutdown should call this inherent method directly
+    /// before dropping the sink, or hold the concrete `SocketSink` type
+    /// (typical pattern: each muxer's `finish()` calls the appropriate
+    /// inherent close method on its captured concrete sink).
+    pub fn finish(&mut self) -> io::Result<()> {
+        self.buf.flush()?;
+        // `shutdown(Write)` signals clean EOF to the peer. Errors here
+        // are non-fatal — the connection may have already been torn down
+        // by the peer — but we surface them so callers can log.
+        self.shutdown_handle.shutdown(Shutdown::Write)
+    }
+}
+
+/// Sequential-only sink over a connected UDP socket.
+///
+/// Each [`write`](Write::write) call sends exactly one datagram. The
+/// caller is responsible for splitting payload at packet boundaries —
+/// for MPEG-TS this means 7 × 188 = 1316 bytes per datagram, the
+/// industry standard for MPEG-TS-over-UDP. No buffering happens here;
+/// adding it would silently merge datagrams.
+///
+/// `finish()` is a no-op: UDP has no end-of-stream marker. Closing the
+/// socket happens on drop.
+pub struct UdpSocketSink {
+    socket: UdpSocket,
+}
+
+impl UdpSocketSink {
+    /// Bind a local UDP socket to an ephemeral port and `connect` it to
+    /// `peer`. `connect` doesn't open a connection — it just fixes the
+    /// peer address so subsequent `send` calls don't need to repeat it,
+    /// and so receive-side filtering rejects packets from other sources.
+    ///
+    /// `sndbuf_bytes`, when present, is a hint to `SO_SNDBUF`.
+    pub fn connect<A: ToSocketAddrs>(peer: A, sndbuf_bytes: Option<usize>) -> io::Result<Self> {
+        // Bind to all-zeros / any port. The kernel picks an ephemeral
+        // source port and the source IP at first send.
+        let socket = UdpSocket::bind("0.0.0.0:0")?;
+        socket.connect(peer)?;
+        if let Some(n) = sndbuf_bytes {
+            set_udp_send_buffer(&socket, n)?;
+        }
+        Ok(Self { socket })
+    }
+}
+
+impl Write for UdpSocketSink {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        // `send` writes the entire datagram or fails — no partial sends
+        // for UDP. Match `Write::write`'s contract by reporting bytes
+        // accepted.
+        self.socket.send(buf)
+    }
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+impl UdpSocketSink {
+    /// No-op — UDP has no end-of-stream marker. Provided for parity
+    /// with [`SocketSink::finish`] so call sites can treat them uniformly.
+    pub fn finish(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+// ── Platform `SO_SNDBUF` tuning ────────────────────────────────────────────
+//
+// std's `TcpStream` / `UdpSocket` don't expose `SO_SNDBUF`. We drop to
+// libc on Linux + macOS (the libc-dep targets in Cargo.toml). On other
+// targets the hint is silently ignored — the socket still works, the
+// kernel just picks its own send-buffer size.
+
+#[cfg(any(target_os = "linux", target_os = "macos"))]
+fn set_send_buffer(stream: &TcpStream, bytes: usize) -> io::Result<()> {
+    use std::os::unix::io::AsRawFd;
+    setsockopt_sndbuf(stream.as_raw_fd(), bytes)
+}
+
+#[cfg(any(target_os = "linux", target_os = "macos"))]
+fn set_udp_send_buffer(socket: &UdpSocket, bytes: usize) -> io::Result<()> {
+    use std::os::unix::io::AsRawFd;
+    setsockopt_sndbuf(socket.as_raw_fd(), bytes)
+}
+
+#[cfg(not(any(target_os = "linux", target_os = "macos")))]
+fn set_send_buffer(_stream: &TcpStream, _bytes: usize) -> io::Result<()> {
+    // Non-Linux-non-macOS targets aren't in Cargo.toml's libc dep list;
+    // silently ignore the hint rather than failing the connect. Callers
+    // can detect via the lack of an explicit "sndbuf applied" signal
+    // (not provided, intentionally — this is a hint, not a guarantee).
+    Ok(())
+}
+
+#[cfg(not(any(target_os = "linux", target_os = "macos")))]
+fn set_udp_send_buffer(_socket: &UdpSocket, _bytes: usize) -> io::Result<()> {
+    Ok(())
+}
+
+#[cfg(any(target_os = "linux", target_os = "macos"))]
+fn setsockopt_sndbuf(fd: std::os::unix::io::RawFd, bytes: usize) -> io::Result<()> {
+    // Clamp into c_int range; SO_SNDBUF takes an `int` argument.
+    let want: libc::c_int = bytes.try_into().unwrap_or(libc::c_int::MAX);
+    let ret = unsafe {
+        libc::setsockopt(
+            fd,
+            libc::SOL_SOCKET,
+            libc::SO_SNDBUF,
+            &want as *const _ as *const libc::c_void,
+            std::mem::size_of::<libc::c_int>() as libc::socklen_t,
+        )
+    };
+    if ret != 0 {
+        return Err(io::Error::last_os_error());
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Read;
+    use std::net::{TcpListener, UdpSocket};
+    use std::thread;
+
+    /// Bind a listener, accept on a thread, return (listener_addr,
+    /// accepted-bytes future via JoinHandle).
+    #[test]
+    fn socket_sink_round_trips_bytes() {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap();
+        let accept = thread::spawn(move || {
+            let (mut sock, _) = listener.accept().unwrap();
+            let mut buf = Vec::new();
+            sock.read_to_end(&mut buf).unwrap();
+            buf
+        });
+
+        let mut sink = SocketSink::connect(addr, Some(256 * 1024)).unwrap();
+        // Write enough to overflow the BufWriter at least once, then a
+        // tail that lives in the buffer until `finish` flushes.
+        let big: Vec<u8> = (0..(2 * TCP_BUF_CAPACITY))
+            .map(|i| (i & 0xff) as u8)
+            .collect();
+        sink.write_all(&big).unwrap();
+        sink.write_all(b"tail\n").unwrap();
+        sink.finish().unwrap();
+        drop(sink);
+
+        let received = accept.join().unwrap();
+        assert_eq!(received.len(), big.len() + 5);
+        assert_eq!(&received[..big.len()], &big[..]);
+        assert_eq!(&received[big.len()..], b"tail\n");
+    }
+
+    #[test]
+    fn socket_sink_is_sequential_only() {
+        // Compile-time assertion via dyn — if this ever started
+        // satisfying `RandomAccessSink`, the trait split would be broken.
+        fn _assert_seq(_: &mut dyn super::super::SequentialSink) {}
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap();
+        let _accept = thread::spawn(move || {
+            let _ = listener.accept();
+        });
+        let mut sink = SocketSink::connect(addr, None).unwrap();
+        _assert_seq(&mut sink);
+        // The negative is harder to assert directly (no `is_not<T>`),
+        // but `SocketSink` does not impl `Seek`, so it can't unify with
+        // `RandomAccessSink`'s super-bound. The Phase 2 blanket impl
+        // `impl<T: SequentialSink + Seek> RandomAccessSink for T {}` thus
+        // excludes it by construction.
+    }
+
+    #[test]
+    fn udp_socket_sink_delivers_datagrams() {
+        let receiver = UdpSocket::bind("127.0.0.1:0").unwrap();
+        receiver
+            .set_read_timeout(Some(std::time::Duration::from_secs(2)))
+            .unwrap();
+        let addr = receiver.local_addr().unwrap();
+        let mut sink = UdpSocketSink::connect(addr, Some(128 * 1024)).unwrap();
+
+        sink.write_all(&[1, 2, 3, 4, 5]).unwrap();
+        sink.write_all(&[9, 9, 9]).unwrap();
+        sink.finish().unwrap();
+
+        let mut buf = [0u8; 64];
+        let n1 = receiver.recv(&mut buf).unwrap();
+        assert_eq!(&buf[..n1], &[1, 2, 3, 4, 5]);
+        let n2 = receiver.recv(&mut buf).unwrap();
+        assert_eq!(&buf[..n2], &[9, 9, 9]);
+    }
+}

src/io/writeback/linux.rs

@@ -321,8 +321,12 @@ fn detect_nfs(fd: RawFd) -> bool {
     }
     // `f_type` is signed (`__fsword_t`) on glibc and unsigned
     // (`c_ulong`) on musl. Cast both sides to i64 for a portable
-    // comparison.
+    // comparison. On glibc x86_64 both already are i64 — clippy flags
+    // the cast as unnecessary on that target only, but we need it for
+    // musl, so silence the lint.
+    #[allow(clippy::unnecessary_cast)]
     let f_type = buf.f_type as i64;
+    #[allow(clippy::unnecessary_cast)]
     let nfs_magic = libc::NFS_SUPER_MAGIC as i64;
     f_type == nfs_magic
 }

src/mux/fmp4/mod.rs

@@ -0,0 +1,404 @@
+//! Fragmented MP4 muxer — **stub** for Phase 3.
+//!
+//! Goal: ISO/IEC 14496-12 fragmented MP4 (`ftyp` + `moov` init segment,
+//! then a sequence of `moof+mdat` media fragments) targeting a
+//! [`SequentialSink`](crate::io::sink::SequentialSink). DASH-friendly,
+//! no Cues backpatch.
+//!
+//! Status (v0.21.0 Phase 3): **STUB**. We ship the init segment
+//! (`ftyp` + a minimal HEVC `moov` skeleton with one video track) so
+//! the muxer's shape and call site are validated, but media fragments
+//! are NOT yet emitted — calls to [`Fmp4Mux::write_video`] currently
+//! accumulate frames into an internal buffer and discard them on
+//! [`Fmp4Mux::finish`].
+//!
+//! ## What's TODO (tracked in Phase 4 / v0.22.0 scope)
+//!
+//! - `moof` box: `mfhd` (sequence_number) + `traf` (`tfhd` + `tfdt`
+//!   + `trun` with sample sizes, durations, flags, composition offsets).
+//! - `mdat` box: concatenated sample data.
+//! - Fragment cadence: one fragment per GOP or every N seconds,
+//!   whichever comes first.
+//! - HEVC `hvcC` box inside `moov.trak.mdia.minf.stbl.stsd` so the
+//!   init segment is self-describing.
+//! - Sample-flags computation (sync vs. delta, depends_on, etc.).
+//! - Edit lists / fragment_duration for accurate seeking.
+//!
+//! Reference: ISO/IEC 14496-12 §8 (Movie Fragments).
+
+use std::io::{self, Write};
+
+// Box type literals — four-character codes per ISO/IEC 14496-12 §4.2.
+
+const FTYP: [u8; 4] = *b"ftyp";
+const MOOV: [u8; 4] = *b"moov";
+const MVHD: [u8; 4] = *b"mvhd";
+const TRAK: [u8; 4] = *b"trak";
+const TKHD: [u8; 4] = *b"tkhd";
+const MDIA: [u8; 4] = *b"mdia";
+const MDHD: [u8; 4] = *b"mdhd";
+const HDLR: [u8; 4] = *b"hdlr";
+const MINF: [u8; 4] = *b"minf";
+const VMHD: [u8; 4] = *b"vmhd";
+const DINF: [u8; 4] = *b"dinf";
+const DREF: [u8; 4] = *b"dref";
+const URL_: [u8; 4] = *b"url ";
+const STBL: [u8; 4] = *b"stbl";
+const STSD: [u8; 4] = *b"stsd";
+const STTS: [u8; 4] = *b"stts";
+const STSC: [u8; 4] = *b"stsc";
+const STSZ: [u8; 4] = *b"stsz";
+const STCO: [u8; 4] = *b"stco";
+const MVEX: [u8; 4] = *b"mvex";
+const TREX: [u8; 4] = *b"trex";
+
+/// Default movie timescale — 90 kHz lines up with MPEG-TS PTS and the
+/// HEVC SPS `vui_time_scale` for film content, simplifying the math
+/// when fragment emission lands.
+const MOVIE_TIMESCALE: u32 = 90_000;
+/// Video track ID. fMP4 init segments conventionally use track_ID=1
+/// for the primary video track; a single-track DASH representation has
+/// no reason to deviate.
+const VIDEO_TRACK_ID: u32 = 1;
+
+/// Fragmented MP4 muxer — stub.
+///
+/// See the module-level doc comment for what is and isn't shipped in
+/// this stub.
+pub struct Fmp4Mux<W: Write> {
+    writer: W,
+    header_written: bool,
+    /// Pending frames — held for the future fragment-emit path. The
+    /// stub drops these on `finish` but keeping them around lets the
+    /// post-stub work re-attach without changing the public API.
+    pending: Vec<PendingSample>,
+    /// hvcC bytes, if provided. Embedded in the `moov.…stsd.hvc1.hvcC`
+    /// box once that path lands.
+    #[allow(dead_code)]
+    codec_private: Option<Vec<u8>>,
+}
+
+struct PendingSample {
+    #[allow(dead_code)]
+    pts_ns: i64,
+    #[allow(dead_code)]
+    keyframe: bool,
+    #[allow(dead_code)]
+    data: Vec<u8>,
+}
+
+impl<W: Write> Fmp4Mux<W> {
+    pub fn new(writer: W) -> Self {
+        Self {
+            writer,
+            header_written: false,
+            pending: Vec::new(),
+            codec_private: None,
+        }
+    }
+
+    /// Provide the `HEVCDecoderConfigurationRecord` for the video track.
+    /// The stub stores it but doesn't yet embed it in `moov` — that's
+    /// part of the post-stub work.
+    pub fn set_video_codec_private(&mut self, hvcc: Vec<u8>) {
+        self.codec_private = Some(hvcc);
+    }
+
+    /// Write one video PES frame.
+    ///
+    /// **Stub behaviour:** the first call emits the init segment
+    /// (`ftyp` + `moov`) so any consumer that just wants the shape can
+    /// receive it. Subsequent calls accumulate frames in memory for
+    /// the future fragmenting path; **no media bytes are written yet**.
+    pub fn write_video(&mut self, pts_ns: i64, keyframe: bool, data: &[u8]) -> io::Result<()> {
+        if !self.header_written {
+            self.write_init_segment()?;
+            self.header_written = true;
+        }
+        // TODO(0.22.0): emit one `moof+mdat` per GOP. For now stash the
+        // frame so the future patch can hot-wire emission without API
+        // churn.
+        self.pending.push(PendingSample {
+            pts_ns,
+            keyframe,
+            data: data.to_vec(),
+        });
+        Ok(())
+    }
+
+    /// Flush. The stub additionally drops accumulated `pending` frames.
+    pub fn finish(&mut self) -> io::Result<()> {
+        // TODO(0.22.0): emit final fragment from pending; today the
+        // stub just clears the buffer to release memory.
+        self.pending.clear();
+        self.writer.flush()
+    }
+
+    fn write_init_segment(&mut self) -> io::Result<()> {
+        let ftyp = build_ftyp();
+        let moov = build_moov();
+        self.writer.write_all(&ftyp)?;
+        self.writer.write_all(&moov)?;
+        Ok(())
+    }
+}
+
+/// Build the `ftyp` box. `major_brand = "iso6"`, `minor_version = 1`,
+/// compatible brands `iso6 dash msdh hvc1` — the same conservative set
+/// shaka-packager uses for HEVC-in-fMP4 outputs.
+fn build_ftyp() -> Vec<u8> {
+    let mut body = Vec::new();
+    body.extend_from_slice(b"iso6");
+    body.extend_from_slice(&1u32.to_be_bytes());
+    body.extend_from_slice(b"iso6");
+    body.extend_from_slice(b"dash");
+    body.extend_from_slice(b"msdh");
+    body.extend_from_slice(b"hvc1");
+    wrap_box(&FTYP, &body)
+}
+
+/// Build the `moov` box — minimal skeleton. Single video trak, no
+/// hvcC inside stsd yet (TODO: full hvc1 sample entry).
+fn build_moov() -> Vec<u8> {
+    let mvhd = build_mvhd();
+    let trak = build_video_trak();
+    let mvex = build_mvex();
+    let mut body = Vec::new();
+    body.extend_from_slice(&mvhd);
+    body.extend_from_slice(&trak);
+    body.extend_from_slice(&mvex);
+    wrap_box(&MOOV, &body)
+}
+
+fn build_mvhd() -> Vec<u8> {
+    // Version 0, 100 bytes total body. Fields per ISO/IEC 14496-12 §8.2.2.
+    let mut body = Vec::new();
+    body.extend_from_slice(&[0, 0, 0, 0]); // version + flags
+    body.extend_from_slice(&0u32.to_be_bytes()); // creation_time
+    body.extend_from_slice(&0u32.to_be_bytes()); // modification_time
+    body.extend_from_slice(&MOVIE_TIMESCALE.to_be_bytes());
+    body.extend_from_slice(&0u32.to_be_bytes()); // duration = 0 (fragmented)
+    body.extend_from_slice(&0x0001_0000u32.to_be_bytes()); // rate 1.0
+    body.extend_from_slice(&0x0100u16.to_be_bytes()); // volume 1.0
+    body.extend_from_slice(&[0u8; 2]); // reserved
+    body.extend_from_slice(&[0u8; 8]); // reserved
+    // 3x3 identity transformation matrix in 16.16 fixed point.
+    for v in [0x1_0000u32, 0, 0, 0, 0x1_0000, 0, 0, 0, 0x4000_0000] {
+        body.extend_from_slice(&v.to_be_bytes());
+    }
+    body.extend_from_slice(&[0u8; 24]); // pre_defined[6]
+    body.extend_from_slice(&2u32.to_be_bytes()); // next_track_ID (1 reserved for video)
+    wrap_box(&MVHD, &body)
+}
+
+fn build_video_trak() -> Vec<u8> {
+    let tkhd = build_tkhd();
+    let mdia = build_mdia();
+    let mut body = Vec::new();
+    body.extend_from_slice(&tkhd);
+    body.extend_from_slice(&mdia);
+    wrap_box(&TRAK, &body)
+}
+
+fn build_tkhd() -> Vec<u8> {
+    let mut body = Vec::new();
+    // version=0 | flags=0x000007 (track_enabled | in_movie | in_preview)
+    body.extend_from_slice(&[0, 0, 0, 7]);
+    body.extend_from_slice(&0u32.to_be_bytes()); // creation_time
+    body.extend_from_slice(&0u32.to_be_bytes()); // modification_time
+    body.extend_from_slice(&VIDEO_TRACK_ID.to_be_bytes());
+    body.extend_from_slice(&[0u8; 4]); // reserved
+    body.extend_from_slice(&0u32.to_be_bytes()); // duration
+    body.extend_from_slice(&[0u8; 8]); // reserved
+    body.extend_from_slice(&0u16.to_be_bytes()); // layer
+    body.extend_from_slice(&0u16.to_be_bytes()); // alternate_group
+    body.extend_from_slice(&0u16.to_be_bytes()); // volume (video=0)
+    body.extend_from_slice(&[0u8; 2]); // reserved
+    // 3x3 identity matrix.
+    for v in [0x1_0000u32, 0, 0, 0, 0x1_0000, 0, 0, 0, 0x4000_0000] {
+        body.extend_from_slice(&v.to_be_bytes());
+    }
+    // width / height in 16.16 fixed point — placeholder 1920x1080.
+    body.extend_from_slice(&(1920u32 << 16).to_be_bytes());
+    body.extend_from_slice(&(1080u32 << 16).to_be_bytes());
+    wrap_box(&TKHD, &body)
+}
+
+fn build_mdia() -> Vec<u8> {
+    let mdhd = build_mdhd();
+    let hdlr = build_hdlr_vide();
+    let minf = build_minf();
+    let mut body = Vec::new();
+    body.extend_from_slice(&mdhd);
+    body.extend_from_slice(&hdlr);
+    body.extend_from_slice(&minf);
+    wrap_box(&MDIA, &body)
+}
+
+fn build_mdhd() -> Vec<u8> {
+    let mut body = Vec::new();
+    body.extend_from_slice(&[0, 0, 0, 0]); // version + flags
+    body.extend_from_slice(&0u32.to_be_bytes()); // creation_time
+    body.extend_from_slice(&0u32.to_be_bytes()); // modification_time
+    body.extend_from_slice(&MOVIE_TIMESCALE.to_be_bytes());
+    body.extend_from_slice(&0u32.to_be_bytes()); // duration
+    // language: 'und' in 5-bit-per-char ISO 639-2 packed (bit 15 = 0).
+    body.extend_from_slice(&[0x55, 0xC4]);
+    body.extend_from_slice(&0u16.to_be_bytes()); // pre_defined
+    wrap_box(&MDHD, &body)
+}
+
+fn build_hdlr_vide() -> Vec<u8> {
+    let mut body = Vec::new();
+    body.extend_from_slice(&[0, 0, 0, 0]); // version + flags
+    body.extend_from_slice(&0u32.to_be_bytes()); // pre_defined
+    body.extend_from_slice(b"vide");
+    body.extend_from_slice(&[0u8; 12]); // reserved
+    body.extend_from_slice(b"VideoHandler\0");
+    wrap_box(&HDLR, &body)
+}
+
+fn build_minf() -> Vec<u8> {
+    let vmhd = build_vmhd();
+    let dinf = build_dinf();
+    let stbl = build_stbl();
+    let mut body = Vec::new();
+    body.extend_from_slice(&vmhd);
+    body.extend_from_slice(&dinf);
+    body.extend_from_slice(&stbl);
+    wrap_box(&MINF, &body)
+}
+
+fn build_vmhd() -> Vec<u8> {
+    let mut body = Vec::new();
+    body.extend_from_slice(&[0, 0, 0, 1]); // version + flags=1
+    body.extend_from_slice(&0u16.to_be_bytes()); // graphicsmode
+    body.extend_from_slice(&[0u8; 6]); // opcolor
+    wrap_box(&VMHD, &body)
+}
+
+fn build_dinf() -> Vec<u8> {
+    let mut dref_body = Vec::new();
+    dref_body.extend_from_slice(&[0, 0, 0, 0]);
+    dref_body.extend_from_slice(&1u32.to_be_bytes()); // entry_count
+    // url with flags=1 (self-contained) and zero name.
+    let url_body = [0u8, 0, 0, 1];
+    dref_body.extend_from_slice(&wrap_box(&URL_, &url_body));
+    let dref = wrap_box(&DREF, &dref_body);
+    wrap_box(&DINF, &dref)
+}
+
+fn build_stbl() -> Vec<u8> {
+    // Stub stsd: empty sample description (zero entries). Replace with
+    // hvc1+hvcC once the fragmenting path lands so the init segment is
+    // actually decodable.
+    let mut stsd_body = Vec::new();
+    stsd_body.extend_from_slice(&[0, 0, 0, 0]);
+    stsd_body.extend_from_slice(&0u32.to_be_bytes()); // entry_count
+    let stsd = wrap_box(&STSD, &stsd_body);
+
+    // Empty stts/stsc/stsz/stco — fragmented init has no samples here.
+    let stts = wrap_box(&STTS, &[0, 0, 0, 0, 0, 0, 0, 0]); // version+flags, count=0
+    let stsc = wrap_box(&STSC, &[0, 0, 0, 0, 0, 0, 0, 0]);
+    let stsz = wrap_box(
+        &STSZ,
+        &[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], // version+flags, sample_size=0, count=0
+    );
+    let stco = wrap_box(&STCO, &[0, 0, 0, 0, 0, 0, 0, 0]);
+
+    let mut body = Vec::new();
+    body.extend_from_slice(&stsd);
+    body.extend_from_slice(&stts);
+    body.extend_from_slice(&stsc);
+    body.extend_from_slice(&stsz);
+    body.extend_from_slice(&stco);
+    wrap_box(&STBL, &body)
+}
+
+fn build_mvex() -> Vec<u8> {
+    // trex: track_ID=1, default_sample_description_index=1, others=0.
+    let mut trex_body = Vec::new();
+    trex_body.extend_from_slice(&[0, 0, 0, 0]); // version + flags
+    trex_body.extend_from_slice(&VIDEO_TRACK_ID.to_be_bytes());
+    trex_body.extend_from_slice(&1u32.to_be_bytes()); // default_sample_description_index
+    trex_body.extend_from_slice(&0u32.to_be_bytes()); // default_sample_duration
+    trex_body.extend_from_slice(&0u32.to_be_bytes()); // default_sample_size
+    trex_body.extend_from_slice(&0u32.to_be_bytes()); // default_sample_flags
+    let trex = wrap_box(&TREX, &trex_body);
+    wrap_box(&MVEX, &trex)
+}
+
+/// Wrap a box body in `[size:u32-BE][type:4]`. Suitable for any body
+/// that fits in u32; oversized boxes (size > 4 GiB) need the 64-bit
+/// large-size extension which we don't generate in the stub.
+fn wrap_box(box_type: &[u8; 4], body: &[u8]) -> Vec<u8> {
+    let size = (body.len() + 8) as u32;
+    let mut out = Vec::with_capacity(body.len() + 8);
+    out.extend_from_slice(&size.to_be_bytes());
+    out.extend_from_slice(box_type);
+    out.extend_from_slice(body);
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Decode the first box's size + type from `buf`.
+    fn read_box_header(buf: &[u8]) -> (u32, [u8; 4]) {
+        let size = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]);
+        let bt = [buf[4], buf[5], buf[6], buf[7]];
+        (size, bt)
+    }
+
+    #[test]
+    fn init_segment_starts_with_ftyp_then_moov() {
+        let mut sink: Vec<u8> = Vec::new();
+        let mut mux = Fmp4Mux::new(&mut sink);
+        // Trigger init emission via a single (stubbed) write.
+        mux.write_video(0, true, &[0x00, 0x00, 0x00, 0x01, 0x40])
+            .unwrap();
+        mux.finish().unwrap();
+        drop(mux);
+
+        let (ftyp_size, ftyp_type) = read_box_header(&sink);
+        assert_eq!(&ftyp_type, b"ftyp");
+        assert!(ftyp_size >= 24, "ftyp too small: {ftyp_size}");
+
+        let (moov_size, moov_type) = read_box_header(&sink[ftyp_size as usize..]);
+        assert_eq!(&moov_type, b"moov");
+        assert!(moov_size > 100, "moov skeleton too small: {moov_size}");
+
+        // Stub guarantee: no media bytes after the init segment.
+        let total = ftyp_size as usize + moov_size as usize;
+        assert_eq!(sink.len(), total, "stub leaked media bytes past moov");
+    }
+
+    #[test]
+    fn moov_contains_trak_mvex() {
+        let mut sink: Vec<u8> = Vec::new();
+        let mut mux = Fmp4Mux::new(&mut sink);
+        mux.write_video(0, true, &[]).unwrap();
+        mux.finish().unwrap();
+        drop(sink);
+
+        // Re-emit into a fresh buffer for parsing.
+        let mut buf: Vec<u8> = Vec::new();
+        let mut mux2 = Fmp4Mux::new(&mut buf);
+        mux2.write_video(0, true, &[]).unwrap();
+        mux2.finish().unwrap();
+        drop(mux2);
+
+        // Find moov payload start.
+        let (ftyp_size, _) = read_box_header(&buf);
+        let moov_start = ftyp_size as usize;
+        let (moov_size, _) = read_box_header(&buf[moov_start..]);
+        let moov_payload = &buf[moov_start + 8..moov_start + moov_size as usize];
+
+        // Scan for the trak and mvex four-CC anywhere in the moov payload.
+        let has_trak = moov_payload.windows(4).any(|w| w == b"trak");
+        let has_mvex = moov_payload.windows(4).any(|w| w == b"mvex");
+        assert!(has_trak, "moov missing trak");
+        assert!(has_mvex, "moov missing mvex");
+    }
+}

src/mux/hevc/mod.rs

@@ -0,0 +1,279 @@
+//! HEVC (H.265) elementary stream muxer — Annex B byte stream.
+//!
+//! Consumes [`PesFrame`](crate::pes::PesFrame)s for a single video track
+//! and writes them as a raw `.hevc` / `.h265` Annex B byte stream:
+//! `00 00 00 01 | NAL_unit | 00 00 00 01 | NAL_unit | …` with no
+//! container framing.
+//!
+//! On the first frame the muxer emits the codec_private's VPS, SPS, PPS
+//! (parsed from a `HEVCDecoderConfigurationRecord` in
+//! `length-prefixed-in-hvcC` form), then converts each PES frame's
+//! length-prefixed NAL units to Annex B and writes them.
+//!
+//! Sequential-only — no Cues, no backpatch. Target sink is any
+//! [`SequentialSink`](crate::io::sink::SequentialSink): file, socket,
+//! pipe, anything `Write + Send`.
+
+use std::io::{self, Write};
+
+/// Annex B 4-byte start code.
+pub(crate) const START_CODE: [u8; 4] = [0x00, 0x00, 0x00, 0x01];
+
+/// HEVC NAL unit type bits live in `(byte0 >> 1) & 0x3F` in Annex B.
+/// We don't filter NAL types here — the muxer is format-only — but we
+/// keep the constant as documentation of the field layout.
+#[allow(dead_code)]
+const HEVC_NAL_TYPE_MASK: u8 = 0x3F;
+
+/// Streaming HEVC Annex B muxer.
+///
+/// One instance per output stream. Tracks whether parameter sets have
+/// already been emitted so they're written exactly once at the head of
+/// the stream, mirroring the convention used by `ffmpeg -c:v copy -f
+/// hevc`.
+pub struct HevcMux<W: Write> {
+    writer: W,
+    /// `HEVCDecoderConfigurationRecord` payload (hvcC). Parsed lazily
+    /// on the first `write_frame` so callers can set it after
+    /// construction but before the first frame.
+    codec_private: Option<Vec<u8>>,
+    /// Set once VPS/SPS/PPS have been written to the stream. Subsequent
+    /// frames write only their own NAL units.
+    params_written: bool,
+}
+
+impl<W: Write> HevcMux<W> {
+    /// Construct over `writer`. The muxer does not impose any extra
+    /// buffering of its own — the sink owns its write buffering policy
+    /// (see [`LocalFileSink`](crate::io::sink::LocalFileSink) and
+    /// [`SocketSink`](crate::io::sink::SocketSink)).
+    pub fn new(writer: W) -> Self {
+        Self {
+            writer,
+            codec_private: None,
+            params_written: false,
+        }
+    }
+
+    /// Provide the `HEVCDecoderConfigurationRecord` (hvcC) so the muxer
+    /// can prepend VPS/SPS/PPS Annex B NALs at stream start. Optional —
+    /// if the PES frames already carry inline parameter sets (some
+    /// upstream demuxers do this), skipping this call is fine.
+    pub fn set_codec_private(&mut self, data: Vec<u8>) {
+        self.codec_private = Some(data);
+    }
+
+    /// Write one PES frame (= one access unit) as Annex B NAL units.
+    ///
+    /// Input may be either:
+    ///   - Length-prefixed: `[u32-BE len][NAL bytes]` repeated. This is
+    ///     the form emitted by libfreemkv's HEVC parser (the MKV-native
+    ///     layout). Converted to Annex B.
+    ///   - Already Annex B: bytes containing `00 00 00 01` start codes
+    ///     anywhere in the buffer. Passed through unchanged.
+    ///
+    /// `_pts_ns` is accepted for symmetry with other muxers but ignored
+    /// — Annex B has no timing layer.
+    pub fn write_frame(&mut self, _pts_ns: i64, data: &[u8]) -> io::Result<()> {
+        if !self.params_written {
+            if let Some(cp) = &self.codec_private {
+                if let Some(params) = hvcc_to_annex_b(cp) {
+                    self.writer.write_all(&params)?;
+                }
+            }
+            self.params_written = true;
+        }
+        let annex_b = length_prefixed_to_annex_b(data);
+        self.writer.write_all(&annex_b)
+    }
+
+    /// Flush the underlying writer. No trailer NAL is needed — an Annex
+    /// B stream ends whenever the file/socket ends.
+    pub fn finish(&mut self) -> io::Result<()> {
+        self.writer.flush()
+    }
+}
+
+/// Convert a `HEVCDecoderConfigurationRecord` (hvcC) into Annex B NAL
+/// units. Returns `Some(bytes)` if at least one NAL was extracted, else
+/// `None`.
+///
+/// Layout (per ISO/IEC 14496-15 §8.3.3.1.2):
+///   - 22-byte fixed header
+///   - byte 22 = `numOfArrays`
+///   - each array: `array_completeness:1 | reserved:1 | NAL_unit_type:6`,
+///     `numNalus:u16-BE`, then `numNalus` × `(nalUnitLength:u16-BE +
+///     NAL bytes)`.
+///
+/// We don't filter on NAL type — VPS (32), SPS (33), PPS (34), and any
+/// SEI arrays included in hvcC all get the same Annex B treatment.
+fn hvcc_to_annex_b(hvcc: &[u8]) -> Option<Vec<u8>> {
+    if hvcc.len() < 23 {
+        return None;
+    }
+    let num_arrays = hvcc[22] as usize;
+    let mut out = Vec::new();
+    let mut offset = 23;
+    for _ in 0..num_arrays {
+        if offset + 3 > hvcc.len() {
+            break;
+        }
+        offset += 1; // array_completeness + nal_type byte
+        let num_nalus = u16::from_be_bytes([hvcc[offset], hvcc[offset + 1]]) as usize;
+        offset += 2;
+        for _ in 0..num_nalus {
+            if offset + 2 > hvcc.len() {
+                break;
+            }
+            let nal_len = u16::from_be_bytes([hvcc[offset], hvcc[offset + 1]]) as usize;
+            offset += 2;
+            if offset + nal_len > hvcc.len() {
+                break;
+            }
+            out.extend_from_slice(&START_CODE);
+            out.extend_from_slice(&hvcc[offset..offset + nal_len]);
+            offset += nal_len;
+        }
+    }
+    if out.is_empty() { None } else { Some(out) }
+}
+
+/// Convert length-prefixed NAL units (`[u32-BE len][NAL]` repeated) to
+/// Annex B (`00 00 00 01 [NAL]` repeated).
+///
+/// If the input doesn't parse as length-prefixed (no valid lengths
+/// extracted), it's returned unchanged on the assumption that it's
+/// already Annex B — some upstream paths (raw HEVC ES from disc) pass
+/// Annex B straight through the PES layer.
+pub(crate) fn length_prefixed_to_annex_b(data: &[u8]) -> Vec<u8> {
+    let mut out = Vec::with_capacity(data.len() + (data.len() / 32));
+    let mut offset = 0;
+    while offset + 4 <= data.len() {
+        let len = u32::from_be_bytes([
+            data[offset],
+            data[offset + 1],
+            data[offset + 2],
+            data[offset + 3],
+        ]) as usize;
+        offset += 4;
+        if offset + len > data.len() {
+            // Mid-NAL truncation — fall through to the pass-through path
+            // rather than emitting a half-NAL.
+            return data.to_vec();
+        }
+        out.extend_from_slice(&START_CODE);
+        out.extend_from_slice(&data[offset..offset + len]);
+        offset += len;
+    }
+    if out.is_empty() && !data.is_empty() {
+        // No length prefixes found — input is likely already Annex B.
+        return data.to_vec();
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn length_prefixed_converts_to_annex_b() {
+        // Two NALs: [3-byte payload AA BB CC] and [2-byte payload DD EE].
+        let mut buf = Vec::new();
+        buf.extend_from_slice(&3u32.to_be_bytes());
+        buf.extend_from_slice(&[0xAA, 0xBB, 0xCC]);
+        buf.extend_from_slice(&2u32.to_be_bytes());
+        buf.extend_from_slice(&[0xDD, 0xEE]);
+
+        let got = length_prefixed_to_annex_b(&buf);
+        let want = [
+            0x00, 0x00, 0x00, 0x01, 0xAA, 0xBB, 0xCC, // first NAL
+            0x00, 0x00, 0x00, 0x01, 0xDD, 0xEE, // second NAL
+        ];
+        assert_eq!(&got[..], &want[..]);
+    }
+
+    #[test]
+    fn already_annex_b_passes_through_when_no_lengths_match() {
+        // A buffer < 4 bytes can't parse a length prefix at all →
+        // pass-through path triggers.
+        let raw = [0xAA, 0xBB, 0xCC];
+        let got = length_prefixed_to_annex_b(&raw);
+        assert_eq!(&got[..], &raw[..]);
+    }
+
+    #[test]
+    fn mid_nal_truncation_returns_original() {
+        // `[u32-BE 100][only 3 bytes]` — length prefix claims 100 bytes
+        // but the input only has 3 after the prefix. We treat that as
+        // malformed and pass the original buffer through so receivers
+        // can attempt their own recovery.
+        let mut raw = Vec::new();
+        raw.extend_from_slice(&100u32.to_be_bytes());
+        raw.extend_from_slice(&[0xAA, 0xBB, 0xCC]);
+        let got = length_prefixed_to_annex_b(&raw);
+        assert_eq!(&got[..], &raw[..]);
+    }
+
+    #[test]
+    fn hvcc_extracts_vps_sps_pps() {
+        // Build a minimal-but-valid hvcC: 22-byte header, then 3 arrays
+        // (VPS / SPS / PPS), each with 1 NAL of a 4-byte payload that
+        // we can spot in the output.
+        let mut hvcc = vec![0u8; 22];
+        hvcc.push(3); // numOfArrays
+        for (nal_type, payload) in [
+            (32u8, [0x40, 0x01, 0x0C, 0x01]),
+            (33, [0x42, 0x01, 0x01, 0x01]),
+            (34, [0x44, 0x01, 0xC1, 0x72]),
+        ] {
+            hvcc.push(nal_type & 0x3F);
+            hvcc.extend_from_slice(&1u16.to_be_bytes()); // numNalus
+            hvcc.extend_from_slice(&(payload.len() as u16).to_be_bytes());
+            hvcc.extend_from_slice(&payload);
+        }
+
+        let annex_b = hvcc_to_annex_b(&hvcc).expect("at least one NAL");
+        // Three NALs × (4-byte start + 4-byte payload) = 24 bytes.
+        assert_eq!(annex_b.len(), 24);
+        assert_eq!(&annex_b[..4], &START_CODE);
+        assert_eq!(&annex_b[8..12], &START_CODE);
+        assert_eq!(&annex_b[16..20], &START_CODE);
+        assert_eq!(annex_b[4], 0x40); // VPS first byte
+        assert_eq!(annex_b[12], 0x42); // SPS first byte
+        assert_eq!(annex_b[20], 0x44); // PPS first byte
+    }
+
+    #[test]
+    fn mux_writes_params_then_frames() {
+        // Build hvcC with one SPS to verify params-once semantics.
+        let mut hvcc = vec![0u8; 22];
+        hvcc.push(1);
+        hvcc.push(33);
+        hvcc.extend_from_slice(&1u16.to_be_bytes());
+        hvcc.extend_from_slice(&3u16.to_be_bytes());
+        hvcc.extend_from_slice(&[0x42, 0x01, 0x01]);
+
+        let mut frame_data = Vec::new();
+        frame_data.extend_from_slice(&2u32.to_be_bytes());
+        frame_data.extend_from_slice(&[0xAA, 0xBB]);
+
+        let mut sink: Vec<u8> = Vec::new();
+        let mut mux = HevcMux::new(&mut sink);
+        mux.set_codec_private(hvcc);
+        mux.write_frame(0, &frame_data).unwrap();
+        // Second frame — no SPS re-emission.
+        mux.write_frame(40_000_000, &frame_data).unwrap();
+        mux.finish().unwrap();
+
+        // SPS NAL (7 bytes) + 2× frame NAL (6 bytes) = 19 bytes.
+        assert_eq!(sink.len(), 7 + 6 + 6);
+        // Start codes at offsets 0 (SPS), 7 (frame1), 13 (frame2).
+        assert_eq!(&sink[0..4], &START_CODE);
+        assert_eq!(&sink[7..11], &START_CODE);
+        assert_eq!(&sink[13..17], &START_CODE);
+        assert_eq!(sink[4], 0x42);
+        assert_eq!(sink[11], 0xAA);
+        assert_eq!(sink[17], 0xAA);
+    }
+}

src/mux/m2ts_mux/mod.rs

@@ -0,0 +1,681 @@
+//! Standard MPEG-TS (188-byte packets) muxer — sequential-only.
+//!
+//! Distinct from `super::tsmux::TsMuxer` (BD-TS with 192-byte packets
+//! and the 4-byte TP_extra_header). This muxer emits the IETF / ISO/IEC
+//! 13818-1 wire format that ffmpeg, VLC, and `m2tsindex` consume
+//! out of the box. Use it for plain `.ts` / `.m2ts` files over a
+//! [`SequentialSink`](crate::io::sink::SequentialSink), and for
+//! MPEG-TS-over-UDP via [`UdpSocketSink`](crate::io::sink::UdpSocketSink).
+//!
+//! ## Wire format
+//!
+//! Every output packet is exactly 188 bytes:
+//!
+//! ```text
+//! sync_byte:8 = 0x47
+//! transport_error_indicator:1 = 0
+//! payload_unit_start_indicator:1
+//! transport_priority:1 = 0
+//! PID:13
+//! transport_scrambling_control:2 = 0
+//! adaptation_field_control:2
+//! continuity_counter:4
+//! [adaptation field, if signalled]
+//! [payload, if signalled]
+//! ```
+//!
+//! ## Wiring
+//!
+//! Single program (PMT PID `0x1000`, program number `1`):
+//!   - PAT on PID `0x0000`
+//!   - PMT on PID `0x1000`
+//!   - Video (HEVC, stream_type `0x24`) on PID `0x0100`
+//!   - First audio (AC3, stream_type `0x81`) on PID `0x0101`
+//!     (or TrueHD, stream_type `0x83`, if hinted)
+//!
+//! ## Scope vs. full TS
+//!
+//! This is a deliberately minimal viable muxer:
+//!   - One program, one video track, optionally one audio track.
+//!   - PAT + PMT re-emitted every `PSI_INTERVAL_PACKETS` packets so a
+//!     mid-stream receiver can lock on within ~100 ms at typical
+//!     UHD bitrates.
+//!   - PCR clock derived from video PTS (`pts - PCR_LEAD_90KHZ`),
+//!     attached to the video PID's adaptation field every
+//!     `PCR_INTERVAL_PACKETS` packets.
+//!   - No language / descriptor tags, no SCTE-35 markers, no per-PID
+//!     PMT version bumps, no SDT/EIT. Sufficient for "ffmpeg can play
+//!     this back", not for full broadcast deployment.
+
+use std::io::{self, Write};
+
+mod packet;
+
+use packet::{Packet, PacketWriter};
+
+/// PID `0x0000` — PAT, mandated by spec.
+const PID_PAT: u16 = 0x0000;
+/// PID for the single program's PMT. `0x1000` is the conventional
+/// choice; anything outside reserved/null ranges works.
+const PID_PMT: u16 = 0x1000;
+/// PID for the video elementary stream.
+const PID_VIDEO: u16 = 0x0100;
+/// PID for the (optional) first audio elementary stream.
+const PID_AUDIO: u16 = 0x0101;
+/// Null PID — reserved by spec; never used here.
+#[allow(dead_code)]
+const PID_NULL: u16 = 0x1FFF;
+
+/// Re-emit PAT/PMT every N TS packets. ~250 × 188 B = 47 KB; at
+/// 80 Mb/s UHD that's ~5 ms — well under typical receiver lock budget.
+const PSI_INTERVAL_PACKETS: u64 = 250;
+/// Re-stamp PCR every N TS packets carrying the video PID. Spec MAX
+/// is 100 ms; 40 packets × 188 B at moderate bitrate keeps us under.
+const PCR_INTERVAL_PACKETS: u64 = 40;
+/// PCR lead time — the PCR must precede the PTS of the first byte of
+/// the picture it timestamps. 200 ms in 90 kHz ticks.
+const PCR_LEAD_90KHZ: u64 = 90_000 / 5;
+
+/// Stream-type codes from ISO/IEC 13818-1 Table 2-29 + later amendments.
+const STREAM_TYPE_HEVC: u8 = 0x24;
+const STREAM_TYPE_AC3: u8 = 0x81;
+const STREAM_TYPE_TRUEHD: u8 = 0x83;
+
+/// Audio codec hint for [`M2tsMux::new`] / [`M2tsMux::set_audio`]. The
+/// muxer needs to know the codec to pick the right PMT `stream_type`
+/// and the right PES stream_id; it doesn't decode the audio.
+#[derive(Debug, Clone, Copy)]
+pub enum AudioCodec {
+    /// AC-3 / E-AC-3, stream_type `0x81`, PES stream_id `0xBD`.
+    Ac3,
+    /// Dolby TrueHD, stream_type `0x83`, PES stream_id `0xBD`.
+    TrueHd,
+}
+
+impl AudioCodec {
+    fn stream_type(self) -> u8 {
+        match self {
+            AudioCodec::Ac3 => STREAM_TYPE_AC3,
+            AudioCodec::TrueHd => STREAM_TYPE_TRUEHD,
+        }
+    }
+}
+
+/// Sequential MPEG-TS muxer with one video and optional one audio track.
+pub struct M2tsMux<W: Write> {
+    out: PacketWriter<W>,
+    /// hvcC parameter set bytes to prepend to the first video PES.
+    /// Optional — if the upstream frames already carry inline params,
+    /// callers can omit this.
+    video_codec_private: Option<Vec<u8>>,
+    /// Set on first video frame: have we emitted VPS/SPS/PPS?
+    params_written: bool,
+    /// Audio codec, if an audio track is configured. `None` = video-only.
+    audio: Option<AudioCodec>,
+    /// First seen PTS (90 kHz). All subsequent PTS / PCR values are
+    /// relative to this so output streams start near t=0 and don't
+    /// confuse downstream parsers that don't tolerate huge starting
+    /// timestamps.
+    base_pts_90k: Option<u64>,
+    /// Per-PID continuity counter, 4 bits, monotonically increasing.
+    cc_video: u8,
+    cc_audio: u8,
+    cc_pat: u8,
+    cc_pmt: u8,
+    /// Total packets written, used to gate PSI / PCR cadence.
+    packets_written: u64,
+    /// Video packets written since last PCR, used to gate PCR cadence.
+    video_packets_since_pcr: u64,
+}
+
+impl<W: Write> M2tsMux<W> {
+    /// Construct a muxer wrapping `writer`. By default audio is
+    /// disabled; call [`set_audio`](Self::set_audio) before the first
+    /// frame to enable.
+    pub fn new(writer: W) -> Self {
+        Self {
+            out: PacketWriter::new(writer),
+            video_codec_private: None,
+            params_written: false,
+            audio: None,
+            base_pts_90k: None,
+            cc_video: 0,
+            cc_audio: 0,
+            cc_pat: 0,
+            cc_pmt: 0,
+            packets_written: 0,
+            video_packets_since_pcr: 0,
+        }
+    }
+
+    /// Provide the `HEVCDecoderConfigurationRecord` for video so the
+    /// muxer prepends VPS/SPS/PPS Annex B NALs at stream start.
+    pub fn set_video_codec_private(&mut self, hvcc: Vec<u8>) {
+        self.video_codec_private = Some(hvcc);
+    }
+
+    /// Enable a single audio track. Must be called before
+    /// [`write_audio`](Self::write_audio).
+    pub fn set_audio(&mut self, codec: AudioCodec) {
+        self.audio = Some(codec);
+    }
+
+    /// Write one video PES frame. `data` is either length-prefixed
+    /// NALUs (MKV-style) or already Annex B; both are accepted.
+    pub fn write_video(&mut self, pts_ns: i64, data: &[u8]) -> io::Result<()> {
+        let pts_90k = self.base_relative_pts(pts_ns);
+        // PCR comes "before" the PTS it timestamps; clamp at 0 for the
+        // first frame so we don't underflow.
+        let pcr = pts_90k.saturating_sub(PCR_LEAD_90KHZ);
+
+        // Annex-B-ify the frame and prepend VPS/SPS/PPS once.
+        let mut es = Vec::with_capacity(data.len() + 64);
+        if !self.params_written {
+            if let Some(cp) = &self.video_codec_private {
+                let payload = hvcc_payload(cp);
+                if !payload.is_empty() {
+                    let params = super::hevc::length_prefixed_to_annex_b(&payload);
+                    es.extend_from_slice(&params);
+                }
+            }
+            self.params_written = true;
+        }
+        let annex_b = super::hevc::length_prefixed_to_annex_b(data);
+        es.extend_from_slice(&annex_b);
+
+        let pes = build_video_pes(pts_90k, &es);
+        self.write_pes(PID_VIDEO, &pes, Some(pcr))
+    }
+
+    /// Write one audio PES frame. Returns `Ok(())` and silently drops
+    /// the frame if no audio track was configured — the design assumes
+    /// the upstream picks tracks and won't ship audio to a video-only
+    /// muxer, but defending against it keeps the API a single shape.
+    pub fn write_audio(&mut self, pts_ns: i64, data: &[u8]) -> io::Result<()> {
+        if self.audio.is_none() {
+            return Ok(());
+        }
+        let pts_90k = self.base_relative_pts(pts_ns);
+        let pes = build_audio_pes(pts_90k, data);
+        self.write_pes(PID_AUDIO, &pes, None)
+    }
+
+    /// Drain the underlying writer. No TS-level trailer is mandatory —
+    /// receivers detect end-of-stream from socket close or file EOF.
+    pub fn finish(&mut self) -> io::Result<()> {
+        self.out.flush()
+    }
+
+    /// Convert input PTS (nanoseconds) to 90 kHz ticks rebased on the
+    /// first frame's PTS. Saturating at 0 keeps the math friendly when
+    /// frames arrive slightly out of decode order.
+    fn base_relative_pts(&mut self, pts_ns: i64) -> u64 {
+        let raw_90k = if pts_ns > 0 {
+            (pts_ns as u64) * 9 / 100_000
+        } else {
+            0
+        };
+        let base = *self.base_pts_90k.get_or_insert(raw_90k);
+        raw_90k.saturating_sub(base)
+    }
+
+    /// Emit one PES payload as a chain of TS packets on `pid`. If `pcr`
+    /// is provided the first packet carries an adaptation field with
+    /// the PCR. PAT/PMT are re-emitted every `PSI_INTERVAL_PACKETS`.
+    ///
+    /// Packet-size math (TS = 188 bytes, header = 4 bytes):
+    ///   - 184 B remain after the header for the adaptation-field area
+    ///     plus the payload area.
+    ///   - With AF body of `b` bytes and `s` stuffing bytes: AF total =
+    ///     `1 + b + s` (the leading `1` is the `adaptation_field_length`
+    ///     byte itself). Payload = `184 - (1 + b + s)`.
+    ///   - With no AF area at all: payload = `184`.
+    ///
+    /// The fit-the-tail logic on the last packet of the PES uses
+    /// stuffing rather than a separate small packet, which is the
+    /// standard MPEG-TS convention.
+    fn write_pes(&mut self, pid: u16, pes: &[u8], pcr: Option<u64>) -> io::Result<()> {
+        // PSI cadence is enforced per TS packet — interleave a fresh
+        // PAT+PMT into the packet stream every PSI_INTERVAL_PACKETS so
+        // long single-PES emissions (e.g. one 60 KB video frame) don't
+        // starve receivers tuning in mid-stream.
+        let mut offset = 0;
+        let mut first = true;
+        while offset < pes.len() {
+            self.maybe_emit_psi()?;
+
+            let attach_pcr = first
+                && (pid == PID_VIDEO)
+                && (pcr.is_some())
+                && (self.packets_written == 0
+                    || self.video_packets_since_pcr >= PCR_INTERVAL_PACKETS);
+
+            let af_body: Vec<u8> = if attach_pcr {
+                build_pcr_adaptation(pcr.unwrap_or(0))
+            } else {
+                Vec::new()
+            };
+
+            let remaining = pes.len() - offset;
+            // Capacity for payload given AF body and 1-byte AF length.
+            // When AF body is empty we can still skip the AF entirely
+            // and get the full 184 B; only invoke the AF when we'd
+            // otherwise need stuffing.
+            let (af_present, payload_len, stuffing): (bool, usize, usize) = if !af_body.is_empty() {
+                // AF is mandatory (PCR). 1 byte length + body + stuffing
+                // + payload = 184.
+                let max_payload = 184 - 1 - af_body.len();
+                let p = remaining.min(max_payload);
+                let s = max_payload - p;
+                (true, p, s)
+            } else if remaining >= 184 {
+                // Full payload packet — no AF at all.
+                (false, 184, 0)
+            } else {
+                // Last (small) packet — stuff via empty AF.
+                // 1 byte length + 0 body + stuffing + payload = 184.
+                let max_payload = 183;
+                let p = remaining.min(max_payload);
+                let s = max_payload - p;
+                (true, p, s)
+            };
+
+            let cc = self.advance_cc(pid);
+            let mut packet = Packet::new();
+            packet.set_header(pid, first, true, af_present, cc);
+            if af_present {
+                packet.append_adaptation(&af_body, stuffing);
+            }
+            packet.append_payload(&pes[offset..offset + payload_len]);
+            debug_assert_eq!(packet.len(), 188, "packet not 188 bytes");
+            self.out.write_packet(&packet)?;
+
+            self.packets_written += 1;
+            if pid == PID_VIDEO {
+                if attach_pcr {
+                    self.video_packets_since_pcr = 0;
+                } else {
+                    self.video_packets_since_pcr += 1;
+                }
+            }
+            offset += payload_len;
+            first = false;
+        }
+        Ok(())
+    }
+
+    fn advance_cc(&mut self, pid: u16) -> u8 {
+        let slot = match pid {
+            PID_VIDEO => &mut self.cc_video,
+            PID_AUDIO => &mut self.cc_audio,
+            PID_PAT => &mut self.cc_pat,
+            PID_PMT => &mut self.cc_pmt,
+            _ => return 0,
+        };
+        let cc = *slot;
+        *slot = (*slot + 1) & 0x0F;
+        cc
+    }
+
+    fn maybe_emit_psi(&mut self) -> io::Result<()> {
+        if self.packets_written == 0 || self.packets_written % PSI_INTERVAL_PACKETS == 0 {
+            self.emit_pat()?;
+            self.emit_pmt()?;
+        }
+        Ok(())
+    }
+
+    fn emit_pat(&mut self) -> io::Result<()> {
+        let payload = build_pat(PID_PMT);
+        let cc = self.advance_cc(PID_PAT);
+        let mut packet = Packet::new();
+        packet.set_header(PID_PAT, true, true, false, cc);
+        packet.append_payload(&payload);
+        packet.pad_to_188();
+        self.out.write_packet(&packet)?;
+        self.packets_written += 1;
+        Ok(())
+    }
+
+    fn emit_pmt(&mut self) -> io::Result<()> {
+        let payload = build_pmt(self.audio);
+        let cc = self.advance_cc(PID_PMT);
+        let mut packet = Packet::new();
+        packet.set_header(PID_PMT, true, true, false, cc);
+        packet.append_payload(&payload);
+        packet.pad_to_188();
+        self.out.write_packet(&packet)?;
+        self.packets_written += 1;
+        Ok(())
+    }
+}
+
+/// Extract the raw hvcC bytes for handoff to `length_prefixed_to_annex_b`.
+/// hvcC layout: 22-byte fixed header, then `numOfArrays` arrays of
+/// `(nalType, numNalus, [nalLength:u16, NAL bytes]…)`. We convert this
+/// directly to a length-prefixed byte stream (NAL length is u16 in
+/// hvcC; widen to u32 for the standard length-prefixed encoding).
+fn hvcc_payload(hvcc: &[u8]) -> Vec<u8> {
+    if hvcc.len() < 23 {
+        return Vec::new();
+    }
+    let num_arrays = hvcc[22] as usize;
+    let mut out = Vec::new();
+    let mut offset = 23;
+    for _ in 0..num_arrays {
+        if offset + 3 > hvcc.len() {
+            break;
+        }
+        offset += 1;
+        let num_nalus = u16::from_be_bytes([hvcc[offset], hvcc[offset + 1]]) as usize;
+        offset += 2;
+        for _ in 0..num_nalus {
+            if offset + 2 > hvcc.len() {
+                break;
+            }
+            let nal_len = u16::from_be_bytes([hvcc[offset], hvcc[offset + 1]]) as usize;
+            offset += 2;
+            if offset + nal_len > hvcc.len() {
+                break;
+            }
+            out.extend_from_slice(&(nal_len as u32).to_be_bytes());
+            out.extend_from_slice(&hvcc[offset..offset + nal_len]);
+            offset += nal_len;
+        }
+    }
+    out
+}
+
+/// Build a PES packet for a video access unit.
+fn build_video_pes(pts_90k: u64, es: &[u8]) -> Vec<u8> {
+    build_pes_packet(0xE0, pts_90k, es, /* length_in_header */ false)
+}
+
+/// Build a PES packet for an audio access unit.
+fn build_audio_pes(pts_90k: u64, es: &[u8]) -> Vec<u8> {
+    // Audio PES: length is fillable when it fits in u16. We always
+    // write the length so receivers don't have to scan for the next
+    // start code.
+    build_pes_packet(0xBD, pts_90k, es, /* length_in_header */ true)
+}
+
+fn build_pes_packet(stream_id: u8, pts_90k: u64, es: &[u8], length_in_header: bool) -> Vec<u8> {
+    let mut out = Vec::with_capacity(es.len() + 14);
+    out.extend_from_slice(&[0x00, 0x00, 0x01, stream_id]);
+    // PES_packet_length: total bytes after this field. 3 flag bytes + 5
+    // PTS bytes + es.len(). Zero means "unbounded" — used for video
+    // where PES can exceed u16.
+    let pes_len = 8 + es.len();
+    if length_in_header && pes_len <= u16::MAX as usize {
+        out.extend_from_slice(&(pes_len as u16).to_be_bytes());
+    } else {
+        out.extend_from_slice(&[0x00, 0x00]);
+    }
+    // Flags byte 1: 10 = MPEG-2 marker, then scrambling/priority/etc 0.
+    out.push(0x80);
+    // Flags byte 2: PTS flag (bit 7).
+    out.push(0x80);
+    // PES_header_data_length = 5 (just PTS).
+    out.push(5);
+    // PTS bytes — 33-bit timestamp split across 5 bytes with marker bits.
+    let pts = pts_90k & 0x1_FFFF_FFFF;
+    out.push(0x21 | (((pts >> 29) & 0x0E) as u8));
+    out.push(((pts >> 22) & 0xFF) as u8);
+    out.push(0x01 | (((pts >> 14) & 0xFE) as u8));
+    out.push(((pts >> 7) & 0xFF) as u8);
+    out.push(0x01 | (((pts << 1) & 0xFE) as u8));
+    out.extend_from_slice(es);
+    out
+}
+
+/// Build the PAT payload (section, with pointer_field).
+fn build_pat(pmt_pid: u16) -> Vec<u8> {
+    let mut section = Vec::new();
+    section.push(0x00); // table_id = PAT
+    // section_syntax_indicator(1) | '0'(1) | reserved(2) | section_length(12)
+    // section_length covers from end of this field through CRC.
+    // Body: transport_stream_id(2) + version/cni(1) + section/last_section(2) + program(4) = 9 bytes,
+    // plus CRC(4) = 13. Encoded big-endian.
+    section.extend_from_slice(&[0xB0, 13]);
+    section.extend_from_slice(&[0x00, 0x01]); // transport_stream_id = 1
+    section.push(0xC1); // reserved | version=0 | current_next=1
+    section.push(0x00); // section_number
+    section.push(0x00); // last_section_number
+    section.extend_from_slice(&[0x00, 0x01]); // program_number = 1
+    // reserved(3) | network_PID/program_map_PID(13)
+    let pid_bytes = (0xE000u16 | (pmt_pid & 0x1FFF)).to_be_bytes();
+    section.extend_from_slice(&pid_bytes);
+    let crc = mpegts_crc32(&section);
+    section.extend_from_slice(&crc.to_be_bytes());
+    // Prepend pointer_field=0 (section starts immediately).
+    let mut payload = Vec::with_capacity(section.len() + 1);
+    payload.push(0x00);
+    payload.extend_from_slice(&section);
+    payload
+}
+
+/// Build the PMT payload (section, with pointer_field).
+fn build_pmt(audio: Option<AudioCodec>) -> Vec<u8> {
+    let mut section = Vec::new();
+    section.push(0x02); // table_id = PMT
+    // section_length filled in after we know the body size.
+    let len_placeholder = section.len();
+    section.extend_from_slice(&[0xB0, 0x00]);
+
+    section.extend_from_slice(&1u16.to_be_bytes()); // program_number
+    section.push(0xC1); // reserved | version=0 | current_next=1
+    section.push(0x00); // section_number
+    section.push(0x00); // last_section_number
+    // reserved(3) | PCR_PID(13)
+    let pcr_pid = (0xE000u16 | (PID_VIDEO & 0x1FFF)).to_be_bytes();
+    section.extend_from_slice(&pcr_pid);
+    // program_info_length = 0
+    section.extend_from_slice(&[0xF0, 0x00]);
+
+    // Video elementary stream entry.
+    section.push(STREAM_TYPE_HEVC);
+    let v_pid = (0xE000u16 | (PID_VIDEO & 0x1FFF)).to_be_bytes();
+    section.extend_from_slice(&v_pid);
+    section.extend_from_slice(&[0xF0, 0x00]); // ES_info_length = 0
+
+    if let Some(codec) = audio {
+        section.push(codec.stream_type());
+        let a_pid = (0xE000u16 | (PID_AUDIO & 0x1FFF)).to_be_bytes();
+        section.extend_from_slice(&a_pid);
+        section.extend_from_slice(&[0xF0, 0x00]);
+    }
+
+    // Now patch section_length: covers everything after the length field
+    // through the CRC, so (current body size - 3 bytes consumed by
+    // table_id + 2 length bytes) + 4 (CRC).
+    let section_len = section.len() - 3 + 4;
+    section[len_placeholder] = 0xB0 | ((section_len >> 8) as u8 & 0x0F);
+    section[len_placeholder + 1] = section_len as u8;
+
+    let crc = mpegts_crc32(&section);
+    section.extend_from_slice(&crc.to_be_bytes());
+
+    let mut payload = Vec::with_capacity(section.len() + 1);
+    payload.push(0x00);
+    payload.extend_from_slice(&section);
+    payload
+}
+
+/// Build the adaptation field carrying a PCR (no other flags).
+fn build_pcr_adaptation(pcr_90k: u64) -> Vec<u8> {
+    // adaptation_field_length is set by `Packet::append_adaptation`
+    // — this function returns just the field body.
+    //
+    // Layout: discontinuity_indicator(1) | random_access(1) |
+    // elementary_stream_priority(1) | PCR_flag(1) | OPCR_flag(1) |
+    // splicing_point_flag(1) | transport_private_data_flag(1) |
+    // adaptation_field_extension_flag(1) | PCR(48b).
+    let mut af = vec![0x50]; // PCR_flag=1, random_access_indicator=1
+    let pcr_base = pcr_90k & 0x1_FFFF_FFFF; // 33-bit
+    let pcr_ext: u16 = 0; // 9-bit, we keep it zero (no sub-tick precision)
+    // Encode PCR: 33b base | 6b reserved | 9b extension = 48b
+    af.push((pcr_base >> 25) as u8);
+    af.push((pcr_base >> 17) as u8);
+    af.push((pcr_base >> 9) as u8);
+    af.push((pcr_base >> 1) as u8);
+    af.push(((pcr_base << 7) as u8 & 0x80) | 0x7E | ((pcr_ext >> 8) as u8 & 0x01));
+    af.push(pcr_ext as u8);
+    af
+}
+
+/// MPEG-TS CRC-32 (poly 0x04C11DB7, init 0xFFFFFFFF, no reflection, no
+/// final XOR). Implementation: bitwise so we don't need a table —
+/// PSI sections are tiny, the cost is negligible.
+fn mpegts_crc32(data: &[u8]) -> u32 {
+    let mut crc: u32 = 0xFFFF_FFFF;
+    for &b in data {
+        crc ^= (b as u32) << 24;
+        for _ in 0..8 {
+            if crc & 0x8000_0000 != 0 {
+                crc = (crc << 1) ^ 0x04C1_1DB7;
+            } else {
+                crc <<= 1;
+            }
+        }
+    }
+    crc
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// All emitted bytes must align to 188-byte packet boundaries and
+    /// every packet must start with `0x47`.
+    fn assert_ts_well_formed(buf: &[u8]) {
+        assert_eq!(
+            buf.len() % 188,
+            0,
+            "stream not packet-aligned: {} bytes",
+            buf.len()
+        );
+        for (i, chunk) in buf.chunks(188).enumerate() {
+            assert_eq!(chunk[0], 0x47, "packet {} missing sync byte", i);
+        }
+    }
+
+    fn extract_pids(buf: &[u8]) -> Vec<u16> {
+        buf.chunks(188)
+            .map(|p| u16::from_be_bytes([p[1] & 0x1F, p[2]]))
+            .collect()
+    }
+
+    #[test]
+    fn crc32_is_self_validating() {
+        // The MPEG-TS CRC has the property that prepending a single
+        // bit-flip changes the output; running it over its own input +
+        // CRC yields a fixed magic constant (the CRC residue). Rather
+        // than hardcoding sample bytes, verify the underlying algorithm
+        // by checking that two distinct inputs produce distinct CRCs
+        // and that the same input is deterministic.
+        let a = [
+            0u8, 0xB0, 0x0D, 0x00, 0x01, 0xC1, 0x00, 0x00, 0x00, 0x01, 0xE1, 0x00,
+        ];
+        let mut b = a;
+        b[5] ^= 0x01; // flip one bit
+        let crc_a = mpegts_crc32(&a);
+        let crc_b = mpegts_crc32(&b);
+        assert_ne!(crc_a, crc_b);
+        assert_eq!(crc_a, mpegts_crc32(&a)); // deterministic
+        // Sanity: all-zero input ⇒ CRC = 0 (init XORs but the
+        // shift/feedback cancels for zero data after init drains).
+        // We don't assert exact value — that depends on poly choice —
+        // but check it's not the same as for non-zero data.
+        let crc_zero = mpegts_crc32(&[0u8; 12]);
+        assert_ne!(crc_zero, crc_a);
+    }
+
+    #[test]
+    fn video_only_mux_emits_pat_pmt_then_video() {
+        let mut sink: Vec<u8> = Vec::new();
+        let mut mux = M2tsMux::new(&mut sink);
+        // One small video frame, no codec_private (so no params NAL inline).
+        let mut frame = Vec::new();
+        frame.extend_from_slice(&4u32.to_be_bytes());
+        frame.extend_from_slice(&[0x40, 0x01, 0x0C, 0x01]);
+        mux.write_video(0, &frame).unwrap();
+        mux.finish().unwrap();
+        drop(mux);
+
+        assert_ts_well_formed(&sink);
+        let pids = extract_pids(&sink);
+        // First two packets: PAT, PMT. At least one video packet after.
+        assert_eq!(pids[0], PID_PAT);
+        assert_eq!(pids[1], PID_PMT);
+        assert!(pids.iter().any(|p| *p == PID_VIDEO));
+    }
+
+    #[test]
+    fn audio_track_appears_in_pmt_and_stream() {
+        let mut sink: Vec<u8> = Vec::new();
+        let mut mux = M2tsMux::new(&mut sink);
+        mux.set_audio(AudioCodec::Ac3);
+        // Video + audio frame pair.
+        let mut frame = Vec::new();
+        frame.extend_from_slice(&3u32.to_be_bytes());
+        frame.extend_from_slice(&[0x40, 0x01, 0x0C]);
+        mux.write_video(0, &frame).unwrap();
+        mux.write_audio(20_000_000, &[0x0B, 0x77, 0x12, 0x34])
+            .unwrap();
+        mux.finish().unwrap();
+        drop(mux);
+
+        assert_ts_well_formed(&sink);
+        let pids = extract_pids(&sink);
+        assert!(pids.iter().any(|p| *p == PID_VIDEO));
+        assert!(pids.iter().any(|p| *p == PID_AUDIO));
+    }
+
+    #[test]
+    fn psi_re_emits_at_interval() {
+        let mut sink: Vec<u8> = Vec::new();
+        let mut mux = M2tsMux::new(&mut sink);
+        // Build a frame large enough to span > PSI_INTERVAL_PACKETS TS packets.
+        // 184 B payload per packet ⇒ ~250 packets = ~46 KB elementary stream.
+        let big: Vec<u8> = (0..(60 * 1024)).map(|i| (i & 0xff) as u8).collect();
+        let mut frame = Vec::new();
+        frame.extend_from_slice(&(big.len() as u32).to_be_bytes());
+        frame.extend_from_slice(&big);
+        mux.write_video(0, &frame).unwrap();
+        mux.finish().unwrap();
+        drop(mux);
+
+        assert_ts_well_formed(&sink);
+        let pids = extract_pids(&sink);
+        // Count PAT/PMT pairs — must be at least 2 given the input size.
+        let pat_count = pids.iter().filter(|p| **p == PID_PAT).count();
+        let pmt_count = pids.iter().filter(|p| **p == PID_PMT).count();
+        assert!(pat_count >= 2, "expected ≥2 PAT, got {}", pat_count);
+        assert!(pmt_count >= 2, "expected ≥2 PMT, got {}", pmt_count);
+    }
+
+    #[test]
+    fn continuity_counter_increments_per_pid() {
+        let mut sink: Vec<u8> = Vec::new();
+        let mut mux = M2tsMux::new(&mut sink);
+        // Three small video frames to get a sequence of video TS packets.
+        for pts in [0i64, 40_000_000, 80_000_000] {
+            let mut frame = Vec::new();
+            frame.extend_from_slice(&3u32.to_be_bytes());
+            frame.extend_from_slice(&[0xAA, 0xBB, 0xCC]);
+            mux.write_video(pts, &frame).unwrap();
+        }
+        mux.finish().unwrap();
+        drop(mux);
+
+        // Collect CCs for video packets in order.
+        let ccs: Vec<u8> = sink
+            .chunks(188)
+            .filter(|p| u16::from_be_bytes([p[1] & 0x1F, p[2]]) == PID_VIDEO)
+            .map(|p| p[3] & 0x0F)
+            .collect();
+        for w in ccs.windows(2) {
+            assert_eq!(w[1], (w[0] + 1) & 0x0F);
+        }
+    }
+}

src/mux/m2ts_mux/packet.rs

@@ -0,0 +1,145 @@
+//! 188-byte MPEG-TS packet builder + writer.
+//!
+//! Internal helper for `super::M2tsMux`. Not public API — exposes raw
+//! byte layout so the parent module can compose PSI / PCR / PES bytes
+//! without each caller re-implementing the 188-byte boundary math.
+
+use std::io::{self, Write};
+
+const TS_PACKET_SIZE: usize = 188;
+const SYNC_BYTE: u8 = 0x47;
+const STUFF_BYTE: u8 = 0xFF;
+
+/// One TS packet under construction. Always emits 188 bytes when
+/// [`pad_to_188`](Self::pad_to_188) is called; if it's not called the
+/// caller is responsible for filling the packet exactly.
+pub(super) struct Packet {
+    buf: Vec<u8>,
+}
+
+impl Packet {
+    pub(super) fn new() -> Self {
+        Self {
+            buf: Vec::with_capacity(TS_PACKET_SIZE),
+        }
+    }
+
+    /// Write the 4-byte TS packet header.
+    ///
+    /// * `pid` — 13-bit PID
+    /// * `payload_unit_start` — first packet of a PES / PSI section
+    /// * `has_payload` — packet carries any payload bytes
+    /// * `has_adaptation` — packet carries an adaptation field
+    /// * `cc` — 4-bit continuity counter
+    pub(super) fn set_header(
+        &mut self,
+        pid: u16,
+        payload_unit_start: bool,
+        has_payload: bool,
+        has_adaptation: bool,
+        cc: u8,
+    ) {
+        self.buf.clear();
+        self.buf.push(SYNC_BYTE);
+        let pus_bit = if payload_unit_start { 0x40 } else { 0 };
+        // transport_error_indicator(1)=0 | payload_unit_start(1) | transport_priority(1)=0 | PID(5 high)
+        self.buf.push(pus_bit | ((pid >> 8) as u8 & 0x1F));
+        self.buf.push(pid as u8);
+        // transport_scrambling_control(2)=0 | adaptation_field_control(2) | continuity_counter(4)
+        let afc = match (has_adaptation, has_payload) {
+            (false, false) => 0b00, // reserved — should not happen
+            (false, true) => 0b01,  // payload only
+            (true, false) => 0b10,  // adaptation only
+            (true, true) => 0b11,   // both
+        };
+        self.buf.push((afc << 4) | (cc & 0x0F));
+    }
+
+    /// Append the adaptation field after the header.
+    ///
+    /// `body` is the adaptation field body (flags byte plus optional PCR
+    /// and so on). `stuffing` is the number of `0xFF` stuffing bytes to
+    /// append after the body. The first byte of the field
+    /// (`adaptation_field_length`) is computed here from
+    /// `body.len() + stuffing`.
+    pub(super) fn append_adaptation(&mut self, body: &[u8], stuffing: usize) {
+        let af_len = body.len() + stuffing;
+        debug_assert!(af_len <= 183, "adaptation field overflow");
+        self.buf.push(af_len as u8);
+        self.buf.extend_from_slice(body);
+        for _ in 0..stuffing {
+            self.buf.push(STUFF_BYTE);
+        }
+    }
+
+    /// Append payload bytes.
+    pub(super) fn append_payload(&mut self, payload: &[u8]) {
+        self.buf.extend_from_slice(payload);
+        debug_assert!(self.buf.len() <= TS_PACKET_SIZE, "packet overflow");
+    }
+
+    /// Pad the packet to exactly 188 bytes with `0xFF` bytes — used by
+    /// PSI emit paths where the section is much smaller than 184 bytes.
+    /// For PSI packets only — payload-carrying packets reserve room for
+    /// stuffing via `append_adaptation`.
+    pub(super) fn pad_to_188(&mut self) {
+        while self.buf.len() < TS_PACKET_SIZE {
+            self.buf.push(STUFF_BYTE);
+        }
+    }
+
+    pub(super) fn bytes(&self) -> &[u8] {
+        &self.buf
+    }
+
+    pub(super) fn len(&self) -> usize {
+        self.buf.len()
+    }
+}
+
+/// Buffered writer for assembled TS packets. Owns the underlying sink.
+pub(super) struct PacketWriter<W: Write> {
+    inner: W,
+}
+
+impl<W: Write> PacketWriter<W> {
+    pub(super) fn new(inner: W) -> Self {
+        Self { inner }
+    }
+
+    pub(super) fn write_packet(&mut self, packet: &Packet) -> io::Result<()> {
+        let bytes = packet.bytes();
+        debug_assert_eq!(bytes.len(), TS_PACKET_SIZE);
+        self.inner.write_all(bytes)
+    }
+
+    pub(super) fn flush(&mut self) -> io::Result<()> {
+        self.inner.flush()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn pad_fills_to_188() {
+        let mut p = Packet::new();
+        p.set_header(0x100, true, true, false, 0);
+        p.append_payload(&[1, 2, 3]);
+        p.pad_to_188();
+        assert_eq!(p.bytes().len(), 188);
+        assert_eq!(p.bytes()[0], SYNC_BYTE);
+        // After 4-byte header + 3 payload, byte 7 starts stuffing.
+        assert_eq!(p.bytes()[7], STUFF_BYTE);
+    }
+
+    #[test]
+    fn header_pid_round_trips() {
+        let mut p = Packet::new();
+        p.set_header(0x1ABC, false, true, false, 0xA);
+        let pid = u16::from_be_bytes([p.bytes()[1] & 0x1F, p.bytes()[2]]);
+        assert_eq!(pid, 0x1ABC);
+        assert_eq!(p.bytes()[3] & 0x0F, 0xA);
+    }
+}

src/mux/mod.rs

@@ -33,6 +33,17 @@ pub(crate) mod m2ts;
 /// to round-trip codec_privates that don't fit inside the underlying format).
 /// Exposed for integration tests that exercise the wire format directly.
 pub mod meta;
+
+// ── Phase 3 sequential muxers ──────────────────────────────────────────────
+//
+// New container muxers that consume PES frames and write to a
+// `SequentialSink`. They are NOT refactors of the existing `MkvStream` /
+// `M2tsStream` (which round-trip via the legacy `Stream` trait + the
+// BD-TS framing); they're sequential-only and target the Phase 2 sink
+// split end-to-end.
+pub mod fmp4;
+pub mod hevc;
+pub mod m2ts_mux;
 pub(crate) mod mkv;
 pub(crate) mod mkvstream;
 pub(crate) mod network;

src/platform/fs_type/linux.rs

@@ -0,0 +1,49 @@
+//! Linux `statfs64`-based filesystem type detection.
+//!
+//! Recognised local-FS magics: ext2/3/4, xfs, btrfs, tmpfs. NFS is the
+//! one network FS this layer cares about (the buffering decision keys
+//! off it). Anything else maps to [`FsType::Unknown`].
+
+use std::ffi::CString;
+use std::os::unix::ffi::OsStrExt;
+use std::path::Path;
+
+use super::FsType;
+
+// Magic numbers from `<linux/magic.h>`. Kept literal here so we don't
+// depend on libc exposing each one — only `NFS_SUPER_MAGIC` is
+// guaranteed to be present across libc / musl revisions.
+const EXT2_SUPER_MAGIC: i64 = 0xEF53;
+const XFS_SUPER_MAGIC: i64 = 0x5846_5342;
+const BTRFS_SUPER_MAGIC: i64 = 0x9123_683E;
+const TMPFS_MAGIC: i64 = 0x0102_1994;
+
+pub(super) fn detect_impl(path: &Path) -> FsType {
+    let cpath = match CString::new(path.as_os_str().as_bytes()) {
+        Ok(c) => c,
+        Err(_) => return FsType::Unknown,
+    };
+    // `statfs64` is repr(C); zeroing is the documented init pattern for
+    // the kernel uapi struct.
+    let mut buf: libc::statfs64 = unsafe { std::mem::zeroed() };
+    let rc = unsafe { libc::statfs64(cpath.as_ptr(), &mut buf) };
+    if rc != 0 {
+        return FsType::Unknown;
+    }
+    // `f_type` is signed (`__fsword_t`) on glibc and unsigned
+    // (`c_ulong`) on musl. Cast both sides to i64 for a portable
+    // comparison. On glibc x86_64 both already are i64 — clippy flags
+    // the cast as unnecessary on that target only, but we need it for
+    // musl, so silence the lint.
+    #[allow(clippy::unnecessary_cast)]
+    let f_type = buf.f_type as i64;
+    #[allow(clippy::unnecessary_cast)]
+    let nfs_magic = libc::NFS_SUPER_MAGIC as i64;
+    if f_type == nfs_magic {
+        return FsType::Nfs;
+    }
+    match f_type {
+        EXT2_SUPER_MAGIC | XFS_SUPER_MAGIC | BTRFS_SUPER_MAGIC | TMPFS_MAGIC => FsType::Local,
+        _ => FsType::Unknown,
+    }
+}

src/platform/fs_type/macos.rs

@@ -0,0 +1,41 @@
+//! macOS `statfs`-based filesystem type detection.
+//!
+//! macOS exposes a textual `f_fstypename` field (e.g. `"apfs"`, `"hfs"`,
+//! `"nfs"`, `"smbfs"`) on its `statfs` struct, which is far more
+//! reliable than chasing magic numbers. Anything starting with `"nfs"`
+//! counts as NFS; anything else recognised maps to `Local`; otherwise
+//! `Unknown`.
+
+use std::ffi::CString;
+use std::os::unix::ffi::OsStrExt;
+use std::path::Path;
+
+use super::FsType;
+
+pub(super) fn detect_impl(path: &Path) -> FsType {
+    let cpath = match CString::new(path.as_os_str().as_bytes()) {
+        Ok(c) => c,
+        Err(_) => return FsType::Unknown,
+    };
+    let mut buf: libc::statfs = unsafe { std::mem::zeroed() };
+    let rc = unsafe { libc::statfs(cpath.as_ptr(), &mut buf) };
+    if rc != 0 {
+        return FsType::Unknown;
+    }
+    // `f_fstypename` is a NUL-terminated C string of length MFSTYPENAMELEN.
+    // SAFETY: libc guarantees the field is initialised to a NUL-terminated
+    // string by a successful statfs.
+    let name_ptr = buf.f_fstypename.as_ptr();
+    let cstr = unsafe { std::ffi::CStr::from_ptr(name_ptr) };
+    let name = cstr.to_bytes();
+    if name.starts_with(b"nfs") {
+        return FsType::Nfs;
+    }
+    // Recognised local types. SMB is not NFS but the buffering-policy
+    // outcome on macOS is the same as for any other local FS — there's
+    // no `WritebackFile` machinery to opt out of on this OS.
+    match name {
+        b"apfs" | b"hfs" | b"exfat" | b"msdos" | b"tmpfs" | b"smbfs" | b"webdav" => FsType::Local,
+        _ => FsType::Unknown,
+    }
+}

src/platform/fs_type/mod.rs

@@ -0,0 +1,112 @@
+//! Filesystem-type detection.
+//!
+//! The buffering architecture (Phase 2) selects different output sinks
+//! for local vs network filesystems: NFS gets the adaptive
+//! `WritebackFile` machinery on Linux; local disks get `LocalFileSink`
+//! and rely on the kernel's default writeback policy. This module
+//! provides the construction-site primitive that picks which one.
+//!
+//! Per the per-OS file-split convention, the actual `statfs` call lives
+//! in the matching platform file (`linux.rs`, `macos.rs`, `windows.rs`,
+//! `other.rs`); this `mod.rs` exposes only the cross-platform enum and
+//! the `detect` entry point.
+
+use std::path::Path;
+
+/// What kind of filesystem a path lives on, to the extent we can tell
+/// cheaply at construction time.
+///
+/// `Unknown` is the fail-open default: a misdetection here should not
+/// be load-bearing for correctness, only for the choice of sink (and
+/// hence buffering policy). Callers that need a binary local/non-local
+/// answer should treat `Unknown` as "probably local".
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum FsType {
+    /// A local on-disk filesystem (ext4, xfs, btrfs, apfs, ntfs, …).
+    Local,
+    /// A network filesystem with NFS-like semantics. The current
+    /// detector lumps SMB / UNC into this on Windows because the
+    /// buffering policy outcome is the same.
+    Nfs,
+    /// `statfs` failed, the filesystem type is not on our recognised
+    /// list, or we are on an OS without a real implementation.
+    Unknown,
+}
+
+#[cfg(target_os = "linux")]
+mod linux;
+#[cfg(target_os = "macos")]
+mod macos;
+#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))]
+mod other;
+#[cfg(target_os = "windows")]
+mod windows;
+
+#[cfg(target_os = "linux")]
+use linux::detect_impl;
+#[cfg(target_os = "macos")]
+use macos::detect_impl;
+#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))]
+use other::detect_impl;
+#[cfg(target_os = "windows")]
+use windows::detect_impl;
+
+/// Best-effort classification of the filesystem under `path`.
+///
+/// Falls back to [`FsType::Unknown`] on any syscall error or unrecognised
+/// filesystem signature. Never panics. Never blocks beyond the cost of
+/// a single `statfs(2)` (Unix) or a string check (Windows).
+pub fn detect(path: &Path) -> FsType {
+    detect_impl(path)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn detect_does_not_panic_on_missing_path() {
+        // Non-existent path should fall through to Unknown, not panic.
+        let p = std::path::Path::new("/this/path/should/not/exist/freemkv-test");
+        let r = detect(p);
+        // We don't assert == Unknown because Windows' heuristic looks at
+        // the leading bytes and might still classify; just confirm the
+        // call returns rather than panicking.
+        let _ = r;
+    }
+
+    #[cfg(target_os = "macos")]
+    #[test]
+    fn macos_tmp_is_local() {
+        // `/tmp` on macOS dev rigs is APFS via the symlink to
+        // `/private/tmp`. Either way, never NFS.
+        let r = detect(std::path::Path::new("/tmp"));
+        assert!(
+            matches!(r, FsType::Local | FsType::Unknown),
+            "expected Local or Unknown for /tmp on macOS, got {r:?}",
+        );
+    }
+
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn linux_tmp_is_local_or_unknown() {
+        // `/tmp` is tmpfs on most distros (which we recognise) but
+        // could be ext4 on others. NFS would be unusual.
+        let r = detect(std::path::Path::new("/tmp"));
+        assert!(
+            matches!(r, FsType::Local | FsType::Unknown),
+            "expected Local or Unknown for /tmp on Linux, got {r:?}",
+        );
+    }
+
+    /// Real NFS exercise needs an actual NFS mount and isn't available
+    /// in CI. Kept here as a manual probe.
+    #[test]
+    #[ignore = "needs an NFS mount path (e.g. /mnt/nfs/...) to validate live"]
+    fn nfs_path_classified_as_nfs() {
+        // Operator passes the mount as FREEMKV_NFS_PROBE; gated behind
+        // `--ignored` because there's no portable NFS path.
+        let p = std::env::var("FREEMKV_NFS_PROBE").expect("set FREEMKV_NFS_PROBE");
+        assert_eq!(detect(std::path::Path::new(&p)), FsType::Nfs);
+    }
+}

src/platform/fs_type/other.rs

@@ -0,0 +1,12 @@
+//! Fallback fs-type detection for platforms without a specific impl.
+//!
+//! Always returns [`FsType::Unknown`]. Callers treat that as
+//! "probably local" and pick `LocalFileSink`.
+
+use std::path::Path;
+
+use super::FsType;
+
+pub(super) fn detect_impl(_path: &Path) -> FsType {
+    FsType::Unknown
+}

src/platform/fs_type/windows.rs

@@ -0,0 +1,26 @@
+//! Windows filesystem-type detection.
+//!
+//! Heuristic-only: any UNC path (`\\server\share\...`) is treated as a
+//! network mount and bucketed into `Nfs`. Strictly, SMB is not NFS, but
+//! the buffering-policy outcome for our purposes is the same — there is
+//! no platform `WritebackFile` machinery on Windows yet, so the worst
+//! case of a false positive is using `LocalFileSink` regardless. A
+//! proper `GetVolumeInformation` query is a Phase 4 concern.
+
+use std::path::Path;
+
+use super::FsType;
+
+pub(super) fn detect_impl(path: &Path) -> FsType {
+    // `Path::starts_with("\\\\")` won't match because component-wise
+    // matching strips the prefix. Look at the raw OsStr instead.
+    let s = path.as_os_str();
+    // OsStr -> [u16] is the platform-correct way on Windows, but
+    // checking the leading bytes via `to_string_lossy` is good enough
+    // for a UNC prefix probe and works on any encoding.
+    let lossy = s.to_string_lossy();
+    if lossy.starts_with("\\\\") || lossy.starts_with("//") {
+        return FsType::Nfs;
+    }
+    FsType::Local
+}

src/platform/mod.rs

@@ -1,5 +1,6 @@
 //! Platform-specific drive initialization and disc probing.
 
+pub mod fs_type;
 pub mod mt1959;
 
 use crate::error::Result;