multimux 0.5.0

//! Issue #805 task 3: end-to-end coverage of the *route-dispatch* path
//! itself -- `multimux::serve_with_registry` / `spawn_ingest` -- rather than
//! any one source driven directly.
//!
//! # The blind spot this file closes
//!
//! Before task 3, no test in this workspace entered through
//! `serve_with_registry`/`spawn_ingest` with real bytes flowing all the way
//! to a served HTTP response: `multimux/tests/rtsp_ingest.rs` drives
//! `RtspDialer`/`RtspIngestSession` directly, and `origin_llhls.rs`/
//! `lldash_dashjs.rs` build a `RouteHandle` and feed it by hand, never
//! referencing `InputSpec` or `serve_with_registry` at all. That is exactly
//! the shape of gap that let eight of nine `InputSpec` variants dispatch to
//! a stubbed no-op arm for a long time while every other gate (build/
//! clippy/doc, thousands of passing tests) stayed green -- see
//! `multimux/src/origin/mod.rs`'s own
//! `every_input_spec_variant_dispatches_to_real_ingest_not_a_stub` (the
//! cheap, exhaustive, in-crate regression net for *that*) for the full
//! story. This file is the deeper, representative half: real bytes in
//! through `serve_with_registry`, real media out over a real HTTP `GET`.
//!
//! # Real fixture, not synthetic bytes
//!
//! Streams the workspace's real, ffmpeg-encoded `fixtures/ts/h264_aac.ts`
//! capture (320x240 Main-profile H.264 @ 25 fps + AAC, ~3.0 s / 75 real
//! video frames with keyframes roughly every second -- see
//! `fixtures/ts/CODEC-ORACLE.md`) verbatim over the wire, exactly as a real
//! camera/HTTP origin would -- not `ts_program::test_support::build_ts_bytes`
//! (a muxed-but-hand-faked NAL payload used by this crate's own unit tests).
//!
//! # Driver-backed kinds
//!
//! `TsUdp` (a UDP socket) and `TsHttp` (a small loopback HTTP server) are the
//! cheapest of the nine to drive with a real fixture -- no RTSP/SRT
//! handshake, no out-of-band SDP, no HLS/DASH/Smooth manifest to author.
//! `Rtmp` (issue #805 task 4) is covered separately below with its own real
//! ffmpeg-captured publish, since it needs the RTMP handshake/`publish`
//! dance rather than a bare byte stream.
//!
//! # `InputSpec::Custom` driver-backed coverage (issue #805 task 5)
//!
//! Before task 5, the `Custom` path drove `crate::pipeline::run_pipeline` (a
//! `SampleSource`-fed segmenter loop) which was itself silently broken for a
//! time: it never published its `Trunk` into `RouteHandle`'s program
//! registry, so every consumer would hang on
//! `ProgramResolution::NotYetAnnounced` (see `RouteHandle::new`'s own doc,
//! "A producer writing the owned `Trunk` must publish it") -- fixed on this
//! branch (`fix(multimux): a producer writing the owned Trunk must publish
//! it, or egress hangs`) before this file existed. Task 5 deleted
//! `run_pipeline`/`SampleSource`/`MockSource` outright (the `Custom` path was
//! their last caller once RTMP left at task 4): a `Custom` factory now spawns
//! `multimux::supervise_driver` over its own `media_plane::ingress::Dialer`/
//! `IngestSession`, exactly like every built-in source (see
//! `examples/custom_scheme.rs`).
//! `custom_dispatch_drives_a_driver_backed_source_and_serves_real_media`
//! below covers that shape through the *exact* dispatch path a real `Custom`
//! route uses (`InputSpec::Custom` -> `SchemeRegistry` -> `InputCtx` -> a
//! factory that spawns `supervise_driver`), replaying the real
//! `h264_aac.ts` fixture (demuxed, not synthetic) through a small
//! `IngestSession` of its own -- so a regression of
//! `crate::source::report_driver_progress`'s registry-publish call (or
//! `crate::source::segment::drive_program_segmenters`'s segmenting) is
//! caught here too for the `Custom` dispatch path specifically, not just by
//! every built-in source's own loopback tests.

use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use multimux::config::{Config, InputSpec, Route};
use multimux::output::OutputKind;
use multimux::registry::SchemeRegistry;
use multimux::serve_with_registry;

fn fixture_path() -> PathBuf {
    PathBuf::from(concat!(
        env!("CARGO_MANIFEST_DIR"),
        "/../fixtures/ts/h264_aac.ts"
    ))
}

/// The real ffmpeg-captured RTMP publish (`app=live`, `stream_key=testkey`,
/// H.264+AAC) -- see `tests/fixtures/PROVENANCE.md`.
fn rtmp_fixture_path() -> PathBuf {
    PathBuf::from(concat!(
        env!("CARGO_MANIFEST_DIR"),
        "/tests/fixtures/rtmp-obs-publish.bin"
    ))
}

/// Reserves a free TCP port, then immediately releases it -- the same
/// "reserve then drop, hand the exact address to the thing that binds it"
/// pattern `multimux/src/source/ts_udp.rs`'s own loopback test uses, just
/// for TCP (this crate's HTTP origin bind address).
fn reserve_tcp_addr() -> SocketAddr {
    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("reserve tcp port");
    let addr = listener.local_addr().expect("local addr");
    drop(listener);
    addr
}

/// Same as [`reserve_tcp_addr`], for a UDP port (the `TsUdp` route's own
/// bind address).
fn reserve_udp_addr() -> SocketAddr {
    let socket = std::net::UdpSocket::bind("127.0.0.1:0").expect("reserve udp port");
    let addr = socket.local_addr().expect("local addr");
    drop(socket);
    addr
}

/// One LL-HLS-only route named `"cam"`, bound at `bind`, ingesting `input` --
/// short segments/parts (0.5 s / 100 ms) so the ~3 s real fixture closes at
/// least one real segment (two keyframes comfortably land inside the fed
/// data) well within this file's polling hang guards.
fn base_config(bind: SocketAddr, input: InputSpec) -> Config {
    Config {
        bind: bind.to_string(),
        target_duration_secs: 0.5,
        part_target_ms: 100,
        window_segments: 8,
        routes: vec![Route {
            name: "cam".to_string(),
            input,
            outputs: vec![OutputKind::LlHls],
        }],
        ..Config::default()
    }
}

/// Polls `playlist_url` until its body carries a real closed-segment
/// `#EXTINF:` line -- deliberately **not** satisfied by
/// `#EXT-X-PART-INF`/`#EXT-X-MAP`, which `ll_hls_runtime`'s engine renders
/// unconditionally even for a route with zero closed segments (see
/// `ll-hls-runtime/src/server/engine.rs`'s `render_playlist`), so a
/// zero-segment route cannot accidentally pass this check.
///
/// A generous hang guard, not a latency assertion (issue #807): real
/// loopback ingest + demux + segmentation of this ~80 KiB fixture is
/// comfortably faster than this bound in practice; the bound exists only so
/// a genuinely broken/dead dispatch path fails the test instead of hanging
/// the suite forever.
async fn poll_until_extinf(client: &reqwest::Client, playlist_url: &str) -> String {
    let deadline = tokio::time::Instant::now() + Duration::from_secs(20);
    loop {
        if let Ok(resp) = client.get(playlist_url).send().await {
            if resp.status().is_success() {
                if let Ok(body) = resp.text().await {
                    if body.contains("#EXTINF:") {
                        return body;
                    }
                }
            }
        }
        if tokio::time::Instant::now() >= deadline {
            panic!(
                "no #EXTINF: line appeared in {playlist_url} within the hang guard -- \
                 dispatched ingest never produced a closed segment"
            );
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
}

/// Extracts the first `seg-{track}-{seq}.m4s` URI out of a rendered media
/// playlist body (mirrors `multimux/tests/origin_llhls.rs`'s own helper).
fn first_segment_uri(playlist: &str) -> &str {
    let start = playlist
        .find("seg-")
        .unwrap_or_else(|| panic!("no seg-*.m4s URI in playlist: {playlist}"));
    let rest = &playlist[start..];
    let end = rest
        .find(".m4s")
        .unwrap_or_else(|| panic!("no .m4s in playlist: {playlist}"))
        + ".m4s".len();
    &rest[..end]
}

/// Fetches `url`, asserting `200 OK` and a non-empty body -- returns the
/// bytes so a caller can assert further (e.g. structural conformance).
async fn get_non_empty(client: &reqwest::Client, url: &str) -> bytes::Bytes {
    let resp = client
        .get(url)
        .send()
        .await
        .unwrap_or_else(|e| panic!("GET {url} failed: {e}"));
    assert_eq!(resp.status(), reqwest::StatusCode::OK, "GET {url}");
    let body = resp
        .bytes()
        .await
        .unwrap_or_else(|e| panic!("reading body of {url} failed: {e}"));
    assert!(!body.is_empty(), "GET {url}: body must be non-empty");
    body
}

/// Real fixture bytes in through `InputSpec::TsUdp` (a real UDP socket
/// `serve_with_registry` binds), real media out over a real HTTP `GET` of
/// the resulting LL-HLS media playlist/init/segment.
///
/// MUTATION VERIFIED: reverting `spawn_ingest`'s `InputSpec::TsUdp` arm (in
/// `multimux/src/origin/mod.rs`) to the pre-#805 combined stub (`{
/// tokio::spawn(async move { tracing::error!(..); }) }`) makes this test
/// fail: `poll_until_extinf` never sees `#EXTINF:` within its 20 s hang
/// guard (the stub never binds a socket, let alone ingests anything), and
/// the `panic!("no #EXTINF: line appeared ... dispatched ingest never
/// produced a closed segment")` inside it fires. Confirmed by applying the
/// same mutation this file's sibling regression net in
/// `multimux/src/origin/mod.rs` already mutation-verifies, rebuilding, and
/// re-running this test to see that exact panic; reverted afterwards.
#[tokio::test]
async fn ts_udp_dispatch_serves_real_media_end_to_end() {
    let bind_addr = reserve_tcp_addr();
    let udp_addr = reserve_udp_addr();
    let config = base_config(
        bind_addr,
        InputSpec::TsUdp {
            addr: udp_addr.to_string(),
            multicast_group: None,
        },
    );

    let server = tokio::spawn(serve_with_registry(config, SchemeRegistry::new()));

    let ts_bytes = std::fs::read(fixture_path()).expect("h264_aac.ts fixture must exist");
    // Resend the whole fixture on a loop (rather than once) so the (async,
    // therefore not synchronized with this test) socket bind inside
    // `serve_with_registry`'s spawned ingest task has ample opportunity to
    // land before a datagram that matters arrives -- exactly the pattern
    // `multimux/src/origin/mod.rs`'s own
    // `ts_udp_input_ingests_and_becomes_resolvable_through_the_registry` test
    // uses, just with real fixture bytes instead of
    // `ts_program::test_support::build_ts_bytes`.
    let stop = Arc::new(AtomicBool::new(false));
    let sender_stop = Arc::clone(&stop);
    let sender = tokio::net::UdpSocket::bind("127.0.0.1:0")
        .await
        .expect("bind sender");
    let send_task = tokio::spawn(async move {
        while !sender_stop.load(Ordering::Relaxed) {
            for chunk in ts_bytes.chunks(7 * 188) {
                let _ = sender.send_to(chunk, udp_addr).await;
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        }
    });

    let client = reqwest::Client::new();
    let playlist_url = format!("http://{bind_addr}/cam/media.m3u8");
    let playlist = poll_until_extinf(&client, &playlist_url).await;
    stop.store(true, Ordering::Relaxed);

    assert!(
        playlist.contains("#EXTINF:"),
        "media playlist must carry a real closed-segment #EXTINF line: {playlist}"
    );

    let init_bytes = get_non_empty(&client, &format!("http://{bind_addr}/cam/init-1.mp4")).await;
    let _ = init_bytes; // non-emptiness already asserted by get_non_empty

    let seg_uri = first_segment_uri(&playlist).to_string();
    let _seg_bytes = get_non_empty(&client, &format!("http://{bind_addr}/cam/{seg_uri}")).await;

    send_task.abort();
    server.abort();
}

/// Same property as [`ts_udp_dispatch_serves_real_media_end_to_end`], for
/// `InputSpec::TsHttp` (a small loopback HTTP server streaming the fixture
/// over chunked transfer-encoding, mirroring
/// `multimux/src/source/ts_http.rs`'s own
/// `start_chunked_ts_server`/`loopback_http_ts_yields_samples_after_pmt_resolves`
/// test) instead of a UDP socket.
///
/// MUTATION VERIFIED: reverting `spawn_ingest`'s `InputSpec::TsHttp` arm to
/// the pre-#805 combined stub (`{ tokio::spawn(async move {
/// tracing::error!(..); }) }`) makes this test fail: the exact same
/// `poll_until_extinf` 20 s hang-guard panic as
/// `ts_udp_dispatch_serves_real_media_end_to_end`'s own mutation-verify
/// above -- `"no #EXTINF: line appeared in http://127.0.0.1:<port>/cam/media.m3u8
/// within the hang guard -- dispatched ingest never produced a closed
/// segment"` (the stub never opens the GET at all). Rebuilt and re-ran to
/// confirm this exact panic, then reverted.
#[tokio::test]
async fn ts_http_dispatch_serves_real_media_end_to_end() {
    use axum::Router;
    use axum::body::Body;
    use axum::response::IntoResponse;
    use axum::routing::get;

    let ts_bytes = std::fs::read(fixture_path()).expect("h264_aac.ts fixture must exist");

    async fn handler(body: axum::extract::State<Vec<u8>>) -> axum::response::Response {
        let chunks: Vec<std::result::Result<Vec<u8>, std::io::Error>> =
            body.0.chunks(7 * 188).map(|c| Ok(c.to_vec())).collect();
        let stream = futures_util::stream::iter(chunks);
        let body = Body::from_stream(stream);
        ([(axum::http::header::CONTENT_TYPE, "video/mp2t")], body).into_response()
    }
    let app = Router::new()
        .route("/stream.ts", get(handler))
        .with_state(ts_bytes);
    let ts_listener = tokio::net::TcpListener::bind("127.0.0.1:0")
        .await
        .expect("bind ephemeral loopback port for the ts source");
    let ts_addr = ts_listener.local_addr().expect("local addr");
    let ts_server = tokio::spawn(async move {
        axum::serve(ts_listener, app).await.expect("axum ts server");
    });

    let bind_addr = reserve_tcp_addr();
    let config = base_config(
        bind_addr,
        InputSpec::TsHttp {
            url: format!("http://{ts_addr}/stream.ts"),
            auth: None,
        },
    );
    let server = tokio::spawn(serve_with_registry(config, SchemeRegistry::new()));

    let client = reqwest::Client::new();
    let playlist_url = format!("http://{bind_addr}/cam/media.m3u8");
    let playlist = poll_until_extinf(&client, &playlist_url).await;

    assert!(
        playlist.contains("#EXTINF:"),
        "media playlist must carry a real closed-segment #EXTINF line: {playlist}"
    );

    let _init_bytes = get_non_empty(&client, &format!("http://{bind_addr}/cam/init-1.mp4")).await;
    let seg_uri = first_segment_uri(&playlist).to_string();
    let _seg_bytes = get_non_empty(&client, &format!("http://{bind_addr}/cam/{seg_uri}")).await;

    server.abort();
    ts_server.abort();
}

/// Real fixture bytes in through `InputSpec::Rtmp` (issue #805 task 4 -- a
/// real TCP client playing back the captured ffmpeg publish against the
/// listen socket `serve_with_registry` binds), real media out over a real
/// HTTP `GET` of the resulting LL-HLS media playlist/init/segment -- the
/// same property as `ts_udp_dispatch_serves_real_media_end_to_end`, for the
/// one push-based (`Listener`-backed) input kind.
///
/// MUTATION VERIFIED: stubbing `spawn_ingest`'s `InputSpec::Rtmp` arm (in
/// `multimux/src/origin/mod.rs`) to the same dead-arm shape the `ts_udp`/
/// `ts_http` siblings' own mutation-verify uses (`tokio::spawn(async move {
/// tracing::error!(..); })`, never binding a listen socket) makes this test
/// fail exactly the same way: `poll_until_extinf`'s `panic!("no #EXTINF:
/// line appeared in http://127.0.0.1:<port>/cam/media.m3u8 within the hang
/// guard -- dispatched ingest never produced a closed segment")` fires after
/// its full 20 s hang guard elapses (confirmed: rebuilt with the stub in
/// place, ran `cargo test -p multimux --test dispatch_ingest
/// rtmp_dispatch_serves_real_media_end_to_end`, saw that exact panic at
/// `multimux/tests/dispatch_ingest.rs:141`, then reverted). This test proves
/// the *dispatch wiring* (`InputSpec::Rtmp` -> `run_rtmp` -> a real HTTP
/// response); `multimux/src/source/rtmp.rs`'s own tests separately
/// mutation-verify the concurrency fix and the first-sample-not-dropped
/// invariant *inside* `run_rtmp`/`RtmpIngestSession`, which this dispatch
/// test does not re-derive (RTMP already served media before issue #805
/// task 4 too -- this test's contribution is pinning the dispatch arm, not
/// distinguishing old from new architecture).
#[tokio::test]
async fn rtmp_dispatch_serves_real_media_end_to_end() {
    use tokio::io::AsyncReadExt;
    use tokio::io::AsyncWriteExt;
    use tokio::net::TcpStream;

    let bind_addr = reserve_tcp_addr();
    let rtmp_addr = reserve_tcp_addr();
    let config = base_config(
        bind_addr,
        InputSpec::Rtmp {
            listen: rtmp_addr.to_string(),
            app: None,
            stream_key: None,
        },
    );
    let server = tokio::spawn(serve_with_registry(config, SchemeRegistry::new()));

    let fixture = std::fs::read(rtmp_fixture_path()).expect("rtmp fixture must exist");
    let publisher = tokio::spawn(async move {
        let mut stream = None;
        for _ in 0..200 {
            match TcpStream::connect(rtmp_addr).await {
                Ok(s) => {
                    stream = Some(s);
                    break;
                }
                Err(_) => tokio::time::sleep(Duration::from_millis(10)).await,
            }
        }
        let mut stream = stream.expect("connect to the RTMP listener");
        stream
            .write_all(&fixture)
            .await
            .expect("write rtmp publish bytes");
        let mut sink = [0u8; 8192];
        loop {
            match stream.read(&mut sink).await {
                Ok(0) | Err(_) => break,
                Ok(_) => {}
            }
        }
    });

    let client = reqwest::Client::new();
    let playlist_url = format!("http://{bind_addr}/cam/media.m3u8");
    let playlist = poll_until_extinf(&client, &playlist_url).await;

    assert!(
        playlist.contains("#EXTINF:"),
        "media playlist must carry a real closed-segment #EXTINF line: {playlist}"
    );

    let _init_bytes = get_non_empty(&client, &format!("http://{bind_addr}/cam/init-1.mp4")).await;
    let seg_uri = first_segment_uri(&playlist).to_string();
    let _seg_bytes = get_non_empty(&client, &format!("http://{bind_addr}/cam/{seg_uri}")).await;

    publisher.abort();
    server.abort();
}

// --- `InputSpec::Custom` driver-backed coverage (issue #805 task 5) ---

mod custom_dispatch_driver_backed {
    use super::*;
    use std::collections::VecDeque;
    use std::convert::Infallible;
    use std::num::NonZeroUsize;

    use broadcast_common::{Demand, Stage, Timestamp};
    use media_plane::ingress::{
        Dialer, HandshakePolicy, IngestDriver, IngestSession, ProgramId, SessionEvent,
    };
    use media_plane::trunk::{RetentionClass, TrunkConfig};
    use multimux::registry::{InputCtx, InputFactory};
    use multimux::route::RouteHandle;
    use multimux::source::{DriverProgress, advance_route};
    use multimux::{Backoff, supervise_driver};
    use transmux::TsDemux;
    use transmux::pipeline::{CodecConfig, Sample, TrackSpec};

    fn nz(n: usize) -> NonZeroUsize {
        NonZeroUsize::new(n).expect("non-zero capacity")
    }

    /// Demux the real `h264_aac.ts` fixture's AVC video track into a
    /// `TrackSpec` + its real, decoded `Sample`s -- mirrors
    /// `tests/lldash_dashjs.rs`'s own `real_video_track_and_samples`: real
    /// bytes in, not `ts_program::test_support::build_ts_bytes`'s
    /// hand-faked NAL payload.
    fn real_video_track_and_samples() -> (TrackSpec, Vec<Sample>) {
        let ts = std::fs::read(fixture_path()).expect("h264_aac.ts fixture must exist");
        let media = TsDemux::new().demux(&ts).expect("demux h264_aac.ts");
        let video = media
            .tracks
            .into_iter()
            .find(|t| matches!(t.spec.config, CodecConfig::Avc { .. }))
            .expect("h264_aac.ts must carry an AVC video track");
        (video.spec, video.samples)
    }

    /// A small `IngestSession` carrying real, pre-demuxed samples -- exactly
    /// the "small `Dialer`/`IngestSession` of its own" shape
    /// `examples/custom_scheme.rs` documents, just fed real fixture-derived
    /// media instead of synthetic frames.
    ///
    /// Its **first** `feed` call announces the one program (the real track)
    /// *and* queues every one of its real samples, in that same call -- the
    /// ordinary shape a real transport produces (a single MPEG-TS feed batch
    /// commonly carries the PMT and the first PES samples together), and
    /// exactly the shape `ProgramSegmenter`'s `subscribe_from_backlog`
    /// cursor (issue #808) exists to handle: it replays whatever backlog is
    /// already resident in the ring by the time `drive_program_segmenters`
    /// builds the segmenter, so samples published in the same `feed` call
    /// that announced the program are not lost.
    struct RealTsSession {
        pending: VecDeque<SessionEvent>,
        sent: bool,
        spec: TrackSpec,
        samples: Vec<Sample>,
    }

    impl Stage for RealTsSession {
        type In<'a> = &'a [u8];
        type Out = SessionEvent;
        type Error = Infallible;

        fn demand(&self) -> Demand {
            Demand::new(1)
        }

        fn feed(&mut self, _input: &[u8], _now: Timestamp) -> Result<(), Infallible> {
            if !self.sent {
                self.sent = true;
                self.pending.push_back(SessionEvent::NewProgram {
                    program: ProgramId(0),
                    tracks: vec![self.spec.clone()],
                });
                let track_id = self.spec.track_id;
                for sample in self.samples.drain(..) {
                    self.pending.push_back(SessionEvent::Sample {
                        program: ProgramId(0),
                        track_id,
                        retention: RetentionClass::Timed,
                        sample,
                    });
                }
            }
            Ok(())
        }

        fn poll(&mut self) -> Option<SessionEvent> {
            self.pending.pop_front()
        }

        fn next_deadline(&self) -> Option<Timestamp> {
            None
        }

        fn on_deadline(&mut self, _now: Timestamp) {}

        fn finish(&mut self) -> Result<(), Infallible> {
            Ok(())
        }
    }

    impl IngestSession for RealTsSession {
        type Request = Infallible;
    }

    /// Constructs a [`RealTsSession`] carrying `spec`/`samples` -- performs
    /// no I/O, exactly like every other `Dialer::dial` in this crate.
    struct RealTsDialer {
        spec: TrackSpec,
        samples: Vec<Sample>,
    }

    impl Dialer for RealTsDialer {
        type Session = RealTsSession;
        type Error = Infallible;

        fn dial(&mut self) -> Result<RealTsSession, Infallible> {
            let mut pending = VecDeque::new();
            pending.push_back(SessionEvent::Established);
            Ok(RealTsSession {
                pending,
                sent: false,
                spec: self.spec.clone(),
                samples: self.samples.clone(),
            })
        }
    }

    /// One dial-through-disconnect attempt -- the `supervise_driver`
    /// `attempt` closure. Mirrors every in-tree `run_*`: dial, wrap in an
    /// `IngestDriver`, feed, and after every feed call [`advance_route`] --
    /// the one facade call `examples/custom_scheme.rs` documents as the whole
    /// extension contract (registry publish + health flip, then turning
    /// samples into servable segments/parts).
    async fn run_real_ts(
        route_handle: Arc<RouteHandle>,
        spec: TrackSpec,
        samples: Vec<Sample>,
    ) -> multimux::Result<()> {
        let mut dialer = RealTsDialer { spec, samples };
        let session = dialer
            .dial()
            .unwrap_or_else(|never: Infallible| match never {});
        let trunk_config = TrunkConfig::new(nz(64), nz(16), nz(8), nz(64), nz(64));
        let handshake = HandshakePolicy::establish_by(Timestamp::from_nanos(u64::MAX));
        let mut driver: IngestDriver<RealTsSession> = IngestDriver::new(
            session,
            trunk_config,
            handshake,
            media_plane::DEFAULT_MAX_PROGRAMS,
        );
        let mut progress = DriverProgress::new();

        // One feed: announces NewProgram AND queues every real sample --
        // mints the driver-side Trunk and publishes its samples in the same
        // batch. The ProgramSegmenter `advance_route`'s own segmenting step
        // builds is subscribed via `subscribe_from_backlog` (issue #808),
        // which replays whatever backlog is already resident in the ring
        // rather than starting from "now" -- so this single feed's samples
        // are not lost.
        driver.feed(&[], Timestamp::from_nanos(0));
        advance_route(&driver, &route_handle, &mut progress);

        driver.finish();
        advance_route(&driver, &route_handle, &mut progress);

        Ok(())
    }

    /// Drives a driver-backed `Custom` factory through the **exact**
    /// dispatch path a real `InputSpec::Custom` route uses:
    /// `serve_with_registry` resolves `Custom`'s `type_tag` through a
    /// `SchemeRegistry`, builds an `InputCtx`, and calls the registered
    /// factory -- this factory spawns `multimux::supervise_driver` over its
    /// own small `Dialer`/`IngestSession` fed the real `h264_aac.ts`
    /// fixture, exactly as a real embedding application's factory would
    /// spawn its own transport-fed driver loop (see
    /// `examples/custom_scheme.rs`). If [`advance_route`]'s internal
    /// registry-publish step were ever skipped, every request below would
    /// hang on `ProgramResolution::NotYetAnnounced` instead of erroring, and
    /// `poll_until_extinf` would time out; if its segmenting step were
    /// skipped, the route would resolve (health `Live`) but never carry a
    /// single `#EXTINF:` line, since nothing would ever turn the ingested
    /// samples into closed segments.
    ///
    /// MUTATION VERIFIED: commenting out `advance_route`'s
    /// `report_driver_progress(driver, route_handle, &mut state.published);`
    /// line in `multimux/src/source/mod.rs` (i.e. simulating a bug in the one
    /// facade every `Custom` factory author now relies on, rather than
    /// hand-assembling the two steps itself) makes this test fail:
    /// `poll_until_extinf`'s 20 s hang guard elapses and its own
    /// `panic!("no #EXTINF: line appeared ... dispatched ingest never
    /// produced a closed segment")` fires at
    /// `multimux/tests/dispatch_ingest.rs:157:13` -- the session demuxes and
    /// segments the real fixture correctly (nothing else changed), but every
    /// HTTP request resolves `ProgramResolution::NotYetAnnounced` forever
    /// since nothing ever published `SPTS_PROGRAM_ID` into the registry, so
    /// the LL-HLS engine never even gets a `Trunk` to render from. Rebuilt
    /// and re-ran to confirm this exact panic, then reverted.
    #[tokio::test]
    async fn custom_dispatch_drives_a_driver_backed_source_and_serves_real_media() {
        let (spec, samples) = real_video_track_and_samples();
        assert!(
            !samples.is_empty(),
            "h264_aac.ts must demux to at least one video sample"
        );

        let mut registry = SchemeRegistry::new();
        registry.register_input(
            "mock-driver-backed",
            Arc::new(move |ctx: InputCtx| {
                let spec = spec.clone();
                let samples = samples.clone();
                Ok(tokio::spawn(supervise_driver(
                    move |route_handle| run_real_ts(route_handle, spec.clone(), samples.clone()),
                    ctx.store,
                    Backoff::production_default(),
                    ctx.name,
                    ctx.shutdown_rx,
                )))
            }) as InputFactory,
        );

        let bind_addr = reserve_tcp_addr();
        let config = base_config(
            bind_addr,
            InputSpec::Custom {
                type_tag: "mock-driver-backed".to_string(),
                params: serde_json::Value::Null,
            },
        );
        let server = tokio::spawn(serve_with_registry(config, registry));

        let client = reqwest::Client::new();
        let playlist_url = format!("http://{bind_addr}/cam/media.m3u8");
        let playlist = poll_until_extinf(&client, &playlist_url).await;

        assert!(
            playlist.contains("#EXTINF:"),
            "media playlist must carry a real closed-segment #EXTINF line: {playlist}"
        );

        let _init_bytes =
            get_non_empty(&client, &format!("http://{bind_addr}/cam/init-1.mp4")).await;
        let seg_uri = first_segment_uri(&playlist).to_string();
        let _seg_bytes = get_non_empty(&client, &format!("http://{bind_addr}/cam/{seg_uri}")).await;

        server.abort();
    }
}