Fix batch overflow in Disc::copy(), DVD PGC parsing, demuxer flush at EOF

- Disc::copy() hardcoded batch=64 sectors, exceeding BU40N's 60-sector
  hw limit. Now accepts batch_sectors param, defaults to 60.
- IFO PGC: playback time at offset 0x04 not 0x02, cell time at cell+4
- DiscStream: set demuxer from content_format (TS for BD, PS for DVD)
- Flush TS/PS demuxers at EOF to avoid losing last PES frame
- M2tsStream: flush demuxer at EOF
- StdioStream: FMKV metadata header for roundtrip compatibility

src/mux/disc.rs

@@ -75,9 +75,17 @@ impl DiscStream {
 
         let mut stream = Self::from_reader(Box::new(drive), title, keys, max_batch);
 
-        if content_format == crate::disc::ContentFormat::MpegPs {
-            stream.ts_demuxer = None;
-            stream.ps_demuxer = Some(super::ps::PsDemuxer::new());
+        // Set demuxer based on content format
+        match content_format {
+            crate::disc::ContentFormat::MpegPs => {
+                stream.ps_demuxer = Some(super::ps::PsDemuxer::new());
+            }
+            crate::disc::ContentFormat::BdTs => {
+                let pids: Vec<u16> = stream.pid_to_track.iter().map(|(pid, _)| *pid).collect();
+                if !pids.is_empty() {
+                    stream.ts_demuxer = Some(super::ts::TsDemuxer::new(&pids));
+                }
+            }
         }
         Ok((stream, disc))
     }
@@ -110,6 +118,18 @@ impl DiscStream {
         let batch: u16 = 64;
 
         let mut stream = Self::from_reader(Box::new(reader), title, keys, batch);
+        // Set demuxer based on content format
+        match disc.content_format {
+            crate::disc::ContentFormat::MpegPs => {
+                stream.ps_demuxer = Some(super::ps::PsDemuxer::new());
+            }
+            crate::disc::ContentFormat::BdTs => {
+                let pids: Vec<u16> = stream.pid_to_track.iter().map(|(pid, _)| *pid).collect();
+                if !pids.is_empty() {
+                    stream.ts_demuxer = Some(super::ts::TsDemuxer::new(&pids));
+                }
+            }
+        }
         stream.disc = Some(disc);
         Ok(stream)
     }
@@ -150,7 +170,9 @@ impl DiscStream {
             batch_sectors,
             errors: 0,
             eof: false,
-            ts_demuxer: if pids.is_empty() { None } else { Some(super::ts::TsDemuxer::new(&pids)) },
+            // Demuxer set by caller — open_drive() checks content_format,
+            // open_iso() always uses TS (Blu-ray ISO).
+            ts_demuxer: None,
             ps_demuxer: None,
             parsers,
             pending_frames: std::collections::VecDeque::new(),
@@ -216,7 +238,41 @@ impl crate::pes::Stream for DiscStream {
         loop {
             if !self.fill_extents() {
                 self.eof = true;
-                return Ok(None);
+                // Flush demuxer — last PES packet may still be in the assembler
+                if let Some(ref mut demuxer) = self.ts_demuxer {
+                    for pes in &demuxer.flush() {
+                        if let Some((_, track)) = self.pid_to_track.iter().find(|(pid, _)| *pid == pes.pid) {
+                            if let Some((_, parser)) = self.parsers.iter_mut().find(|(pid, _)| *pid == pes.pid) {
+                                for frame in parser.parse(pes) {
+                                    self.pending_frames.push_back(
+                                        crate::pes::PesFrame::from_codec_frame(*track, frame)
+                                    );
+                                }
+                            }
+                        }
+                    }
+                }
+                // PS demuxer flush (DVD)
+                if let Some(ref mut demuxer) = self.ps_demuxer {
+                    for ps in &demuxer.flush() {
+                        let track = match ps.stream_id {
+                            0xE0..=0xEF => 0,
+                            0xC0..=0xDF => 1,
+                            0xBD => ps.sub_stream_id.map(|s| (s & 0x1F) as usize + 1).unwrap_or(1),
+                            _ => continue,
+                        };
+                        if track < self.title.streams.len() {
+                            let pts_ns = ps.pts.map(|p| (p as i64) * 1_000_000_000 / 90_000).unwrap_or(0);
+                            self.pending_frames.push_back(crate::pes::PesFrame {
+                                track,
+                                pts: pts_ns,
+                                keyframe: true,
+                                data: ps.data.clone(),
+                            });
+                        }
+                    }
+                }
+                return Ok(self.pending_frames.pop_front());
             }
 
             let bytes = self.buf_valid;

src/mux/m2ts.rs

@@ -172,7 +172,21 @@ impl crate::pes::Stream for M2tsStream {
             let n = reader.read(&mut buf)?;
             if n == 0 {
                 self.pes_eof = true;
-                return Ok(None);
+                // Flush demuxer — last PES packet may still be in the assembler
+                if let Some(ref mut demuxer) = self.demuxer {
+                    for pes in &demuxer.flush() {
+                        if let Some((_, track)) = self.pid_to_track.iter().find(|(pid, _)| *pid == pes.pid) {
+                            if let Some((_, parser)) = self.parsers.iter_mut().find(|(pid, _)| *pid == pes.pid) {
+                                for frame in parser.parse(pes) {
+                                    self.pending_frames.push_back(
+                                        crate::pes::PesFrame::from_codec_frame(*track, frame)
+                                    );
+                                }
+                            }
+                        }
+                    }
+                }
+                return Ok(self.pending_frames.pop_front());
             }
 
             if let Some(ref mut demuxer) = self.demuxer {

src/mux/ps.rs

@@ -163,6 +163,7 @@ impl PsDemuxer {
 
         packets
     }
+
 }
 
 /// Check whether a start code byte is a valid PES stream ID that carries payload.

src/mux/stdio.rs

@@ -1,13 +1,21 @@
-//! StdioStream — PES frames via stdin/stdout.
+//! StdioStream — PES frames via stdin/stdout with FMKV metadata header.
+//!
+//! The FMKV header carries stream metadata (PIDs, codecs, languages, codec_privates)
+//! so the receiving end can set up muxing without scanning the content.
 
+use super::meta;
 use crate::disc::DiscTitle;
 use std::io::{self, Write};
 
 /// Stdio stream — reads PES from stdin, writes PES to stdout.
+/// FMKV metadata header is written/read automatically.
 pub struct StdioStream {
     disc_title: DiscTitle,
     reader: Option<io::Stdin>,
     writer: Option<io::BufWriter<io::Stdout>>,
+    header_written: bool,
+    header_read: bool,
+    stored_codec_privates: Vec<Option<Vec<u8>>>,
 }
 
 impl StdioStream {
@@ -17,6 +25,9 @@ impl StdioStream {
             disc_title: DiscTitle::empty(),
             reader: Some(io::stdin()),
             writer: None,
+            header_written: false,
+            header_read: false,
+            stored_codec_privates: Vec::new(),
         }
     }
 
@@ -26,12 +37,32 @@ impl StdioStream {
             disc_title: title.clone(),
             reader: None,
             writer: Some(io::BufWriter::new(io::stdout())),
+            header_written: false,
+            header_read: false,
+            stored_codec_privates: Vec::new(),
         }
     }
+
+    /// Read the FMKV metadata header from stdin on first read.
+    fn ensure_header_read(&mut self) -> io::Result<()> {
+        if self.header_read {
+            return Ok(());
+        }
+        self.header_read = true;
+        if let Some(ref mut r) = self.reader {
+            if let Ok(Some(m)) = meta::read_header(r) {
+                let title = m.to_title();
+                self.stored_codec_privates = title.codec_privates.clone();
+                self.disc_title = title;
+            }
+        }
+        Ok(())
+    }
 }
 
 impl crate::pes::Stream for StdioStream {
     fn read(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
+        self.ensure_header_read()?;
         match &mut self.reader {
             Some(r) => crate::pes::PesFrame::deserialize(r),
             None => Err(crate::error::Error::StreamWriteOnly.into()),
@@ -39,7 +70,16 @@ impl crate::pes::Stream for StdioStream {
     }
     fn write(&mut self, frame: &crate::pes::PesFrame) -> io::Result<()> {
         match &mut self.writer {
-            Some(w) => frame.serialize(w),
+            Some(ref mut w) => {
+                if !self.header_written {
+                    if !self.disc_title.streams.is_empty() {
+                        let m = meta::M2tsMeta::from_title(&self.disc_title);
+                        meta::write_header(w, &m)?;
+                    }
+                    self.header_written = true;
+                }
+                frame.serialize(w)
+            }
             None => Err(crate::error::Error::StreamReadOnly.into()),
         }
     }
@@ -48,4 +88,13 @@ impl crate::pes::Stream for StdioStream {
         Ok(())
     }
     fn info(&self) -> &DiscTitle { &self.disc_title }
+
+    fn codec_private(&self, track: usize) -> Option<Vec<u8>> {
+        self.stored_codec_privates.get(track).and_then(|c| c.clone())
+    }
+
+    fn headers_ready(&self) -> bool {
+        // After first read(), header is parsed and codec_privates populated
+        self.header_read || self.writer.is_some()
+    }
 }