media_plane/

ingress.rs

//! `Dialer`/`Listener`/`IngestSession` — the ingress traits, and the generic
//! `run_dial`/`run_listen` drivers that pump them into a [`crate::Trunk`]
//! (plan step 3c;
//! `docs/superpowers/specs/2026-07-26-media-plane-architecture.md` §2).
//!
//! Traits and generic drivers only — no real protocol is ported here (that
//! is plan step 5, "port the 9 sources"). This module's job is to define a
//! shape those nine `multimux::source::*` implementations can all actually
//! be squeezed into, and to own the feed/poll/deadline/dispatch loop so no
//! protocol has to reimplement it (every one of the nine today hand-rolls
//! its own `while let Some(event) = demux.poll_event() { match event { .. } }`
//! drain — see e.g. `multimux::source::ts_udp::TsUdpSession::next_samples`).
//!
//! `#[cfg(feature = "std")]`: like [`crate::trunk`], this module wires
//! straight to [`crate::Trunk`] (`Arc`/`HashMap`), and every real consumer is
//! `std`+`tokio` per the architecture (see `crate::trunk`'s module docs).
//!
//! # `IngestSession` is a `Stage`, matching `ByteStage`'s precedent
//!
//! Like [`crate::ByteStage`], `IngestSession` builds on
//! [`broadcast_common::Stage`] — but, unlike `ByteStage`, it is not a bare
//! blanket alias over it. Only the output is pinned:
//!
//! ```text
//! pub trait IngestSession: for<'a> Stage<Out = SessionEvent> + Send {
//!     type Request: Send;
//!     fn poll_transmit(&mut self) -> Option<Self::Request> { None }
//! }
//! ```
//!
//! `Stage::In<'a>` is deliberately **not** pinned to `&'a [u8]` (round 3;
//! it was through round 2) — see
//! [Pull sources need a typed request/response identity](#pull-sources-need-a-typed-requestresponse-identity-round-3)
//! below for why. This still buys the same things `ByteStage` documents for
//! the byte-stream sources that make up most of the plane: one drive model,
//! `finish()` for a clean end-of-input flush, and `demand()` for
//! back-pressure — [`run_dial`]/[`run_listen`] drive any `IngestSession` with
//! the same "feed, drain `poll()`, repeat" loop
//! [`crate::byte_stage`]'s own tests already validate against a real `Stage`,
//! whatever `In<'a>` an implementor chooses. The two things every
//! `IngestSession` adds over a bare `Stage` are
//! [`poll_transmit`](IngestSession::poll_transmit) (with a `None`-returning default —
//! see
//! [Why `poll_transmit` exists](#why-poll_transmit-exists-and-most-sources-will-never-override-it)
//! below) and [`Request`](IngestSession::Request) (with **no** default — every
//! implementor names its own request type explicitly; see the pull-sources
//! section below for why it has none).
//!
//! [`ByteStage`]: crate::ByteStage
//!
//! ## Why `poll_transmit` exists, and most sources will never override it
//!
//! `RtspSession` (`multimux/src/source/rtsp.rs`) is driven over an
//! interleaved TCP connection where the client is also expected to send
//! (RTCP receiver reports, periodic keepalive) — a pure `feed(bytes) ->
//! poll() -> SessionEvent` consumer has nowhere to hand that back. Rather
//! than invent a second trait for the two or three sources that need it,
//! [`IngestSession::poll_transmit`] is a plain method with a `None` default:
//! a driver (this module's [`IngestDriver`]/[`ListenDriver`], or a future
//! Step 5 adapter) drains it after every `feed`/`on_deadline` exactly like
//! [`Stage::poll`], and a session with nothing to send simply never
//! overrides it.
//!
//! # The program dimension (B5) — `SessionEvent::NewProgram` at any time
//!
//! Rev 1 of the architecture assumed one connection maps to exactly one
//! timeline; the audit's finding B5 is that MPTS and T2-MI multi-PLP break
//! this outright (`parse_pat` flattens every program; `program_number`
//! appears nowhere in the demuxed IR). [`SessionEvent::NewProgram`] is how an
//! `IngestSession` announces one: [`IngestDriver`]/[`ListenDriver`] mint a
//! **fresh [`Trunk`]** for every [`ProgramId`] the moment it is announced —
//! including the second, third, ... program on the *same* connection, and
//! including one announced only after other programs (or samples for them)
//! have already been flowing. There is no "known programs" list supplied up
//! front and no special first-poll path: `NewProgram` is just another
//! [`SessionEvent`] variant, driven through the exact same `poll()` drain as
//! every `Sample`, so "a program appears mid-session" is not a distinct code
//! path from "a program was there from the start" — it is the *only* path.
//!
//! This is deliberately more general than architecture §1.3's steady-state
//! design (program-splitting as a `ByteStage` upstream of demux, so each
//! `IngestSession` only ever sees one already-known program). That design is
//! still the right Step 5 target for MPTS — it lets each program's demux run
//! independently — but it presupposes the program table has already been
//! read once to know how many `ByteStage`s to build, which is exactly the
//! chicken-and-egg `multimux::source::ts_udp::TsUdpSession::next_samples`
//! hits *today*: a PMT version bump that adds a **track** after `connect()`'s
//! one-shot `track_specs()` snapshot is already only handled by logging a
//! warning and dropping it (see that function's `DemuxEvent::TrackAdded`
//! arm) — there is no live wiring for "new track" today, let alone "new
//! program". `SessionEvent::NewProgram` is what closes the "new program"
//! half of that gap generically: whether a program is split upstream (known
//! before the session starts) or discovered while demuxing an MPTS in one
//! session, the driver's reaction is identical — mint a `Trunk`, keep going.
//! [`SessionEvent::TracksChanged`] (issue #781) closes the other half — the
//! `ts_udp` warn-and-drop case cited above is exactly what an `IngestSession`
//! now has a real event to emit instead of logging and discarding.
//!
//! # Supervision: EOF is not an error (the `HealthState` fix)
//!
//! Today's bug, concretely: `multimux::origin::supervisor::supervise` treats
//! `run_pipeline`'s `Ok(())` (clean source EOF) and `Err(_)` (a real failure)
//! identically — both fall into the same `set_health(HealthState::Reconnecting)`
//! arm (`multimux/src/origin/supervisor.rs`) — and
//! `hls_runtime::server::store::HealthState::Failed`'s own doc comment
//! admits *"the loop here does not currently produce it"*. Nothing
//! distinguishes "the stream ended" from "the stream broke" because both are
//! folded into one `Result<(), Error>` before the health state is even set.
//!
//! [`HealthState`] here is not a copy of that enum — it is what the fold
//! above should have been: an `IngestSession`'s [`Stage::finish`] returning
//! `Ok(())` (clean end of input, no error ever raised) drives
//! [`HealthState::Ended`]; its [`Stage::feed`]/[`finish`](Stage::finish)
//! returning `Err` drives [`HealthState::Failed`] carrying that concrete error, generically
//! (`HealthState<E>`, `E = S::Error`) rather than losing it to a formatted
//! string. Both are reachable and observed via [`IngestDriver::health`]/
//! [`ListenDriver::health`] — see this module's tests for a mutation-checked
//! proof that ending cleanly is never mistaken for failing.
//!
//! # `Listener` and `max_sessions`: enforced by the driver, not by convention
//!
//! `max_sessions` lives on the [`Listener`] trait as a fixed accessor, but
//! **[`ListenDriver::poll_accept`] is the only place it is checked** — a
//! concrete `Listener` cannot forget to enforce it (there is nothing for it
//! to enforce; `poll_accept` just hands back whatever the transport
//! accepted). Once `max_sessions` live sessions are admitted, every further
//! accepted connection is dropped **immediately, before being fed a single
//! byte** — this project has already shipped four unbounded-allocation
//! vectors, and an unbounded listener is exactly that class of bug, so the
//! bound is structural (checked in one generic place) rather than a
//! per-protocol discipline.
//!
//! # `max_programs`: the fifth unbounded-allocation vector, and why it needs
//! its own bound rather than reusing `max_sessions` or a `Trunk` capacity
//!
//! [`SessionEvent::NewProgram`] (above) mints a **fresh [`Trunk`]** — five
//! bounded rings — per distinct [`ProgramId`] a session reports, with
//! nothing capping how many distinct ids one session may report. Every
//! individual ring is bounded ([`TrunkConfig`]'s five [`NonZeroUsize`]
//! capacities), which is exactly what makes this easy to miss: the unbounded
//! quantity is the *number of `Trunk`s*, not anything inside one, so no
//! per-ring capacity — however carefully chosen — helps at all. A malformed
//! or hostile multiplex announcing thousands of `program_number`s allocates
//! thousands of trunks. This is the fifth vector of this class this project
//! has shipped, all in code consuming remote input; the other two knobs
//! already documented in this module do not cover it:
//!
//! - **Not `max_sessions`.** That bounds concurrent *connections*; this is a
//!   count of *programs announced within one already-admitted connection* —
//!   a different axis entirely, and B5's whole premise is that one session
//!   can legitimately report many programs.
//! - **Not a `TrunkConfig` capacity.** Those bound *entries within one
//!   program's rings*; none of them says anything about how many programs
//!   may exist.
//!
//! So [`IngestDriver`] (and [`ListenDriver`], which embeds one per admitted
//! session) takes its own `max_programs: NonZeroUsize` — [`NonZeroUsize`] for
//! the same reason [`TrunkConfig`]'s five capacities are: zero is
//! unrepresentable rather than merely rejected, so there is no fallible
//! constructor to remember to call. It is enforced in exactly one place,
//! [`IngestDriver`]'s internal `drain()`, mirroring `max_sessions`'
//! placement: a `NewProgram` is checked against the bound **inside the
//! driver that owns the `programs`/`writers` maps**, not by any convention an
//! `IngestSession` implementor could forget — indeed an `IngestSession` has
//! no visibility into those maps at all, so there is nothing for it to
//! bypass even in principle.
//!
//! **The (N+1)th program is refused, not fatal — the admitted programs keep
//! flowing.** The alternative (failing the whole session once its program
//! count exceeds the bound) was rejected: a 200-program hostile or malformed
//! multiplex would then take down ingest for the 8 programs a real caller
//! asked for, which is a worse outcome than simply not admitting the extra
//! 192. Concretely: a `NewProgram` past the bound gets no [`Trunk`] — no
//! [`Trunk::new`] call happens for it at all, not merely an unstored one —
//! and any later `Sample` for it is dropped by the *already-existing,
//! already-tested* "sample for an unannounced program" path (see
//! [`SessionEvent::Sample`]'s docs), because a refused program never gets a
//! `writers` entry either. No new drop path was invented; refusal reuses the
//! one this module already had to have.
//!
//! **Refusal is reported, not silent** — this project's own #781 postmortem
//! (a silently dropped item that stayed invisible for a long time) is the
//! reason a bare `if len >= max { return; }` is not acceptable here.
//! [`IngestDriver::refused_program_count`] is a monotonically increasing
//! counter, incremented once per refused `NewProgram`, queryable at any time
//! — the same shape as [`DialSupervisor::attempts`]/[`DialAttempt::Exhausted`]
//! (a bounded count of "how many times has this happened", not a stored list
//! of each occurrence). A list of every refused [`ProgramId`] was considered
//! and rejected: retaining one entry per refusal is the *exact same*
//! unbounded-growth shape this fix exists to close, just moved from `Trunk`s
//! to a `Vec<ProgramId>` — a counter is `O(1)` in memory regardless of how
//! many programs a flood announces, which the accompanying flood test
//! proves directly.
//!
//! **Default: [`DEFAULT_MAX_PROGRAMS`].** A real DVB MPTS typically carries a
//! single-digit to low-tens program count; ATSC and cable multiplexes can run
//! somewhat higher (a handful of dozens is a realistic outer bound for a
//! legitimate stream). [`DEFAULT_MAX_PROGRAMS`]`= 64` sits comfortably above
//! any legitimate multiplex this project's fixtures or docs describe, while
//! still capping a hostile "thousands of programs" stream to 64 trunks (320
//! rings) rather than an unbounded count.
//!
//! # Establishment is ordinary driving — `dial()` performs no I/O
//!
//! **[`Dialer::dial`] does not connect anything.** It *constructs* a session
//! in a not-yet-established state, along with whatever first bytes that
//! session wants sent (queued for [`IngestSession::poll_transmit`]). The
//! handshake then completes through the **same feed/poll pump as everything
//! else**: the driver writes [`IngestSession::poll_transmit`]'s bytes to the
//! socket, reads the peer's reply, hands it to [`Stage::feed`], and the
//! session either queues the next request or announces
//! [`SessionEvent::Established`]. No I/O happens inside any trait method, so
//! the plane stays genuinely sans-IO and tokio stays out of this layer.
//!
//! This is deliberately **the same pattern `rtsp-runtime` already uses**, not
//! a second invention: `rtsp_runtime::client::ClientSession` is a sans-IO
//! engine whose request builders (`describe`/`setup`/`play`) *return the
//! outbound bytes to send* and whose `handle_data` consumes inbound bytes and
//! returns typed `ClientEvent`s, with the RFC 2326 Appendix A.1 state machine
//! held internally and exposed via `state()`. `IngestSession` is that shape
//! expressed through `Stage`: `poll_transmit` ≙ "the bytes to send",
//! `feed` ≙ `handle_data`, `poll` ≙ the returned events, and
//! [`IngestDriver::health`] ≙ `state()`. `hls-runtime` splits its client
//! and server engines the same way.
//!
//! An earlier revision of this module had `dial()` "perform the whole
//! connect/handshake" and return an already-live session. That was wrong and
//! is recorded here rather than quietly changed: a sans-IO trait cannot do
//! I/O, so such a `dial()` only ever fits sources whose "connect" is a purely
//! local operation (binding a UDP socket). Every genuinely multi-round-trip
//! source — RTSP (DESCRIBE → SETUP × N → PLAY), an SRT caller handshake,
//! TS-UDP's read-until-PMT-resolves — would have needed an executor bridge
//! (`block_on`, or a handshake thread) to be callable at all, dragging tokio
//! back into the layer that was kept free of it on purpose.
//!
//! ## One session type, not a separate `PendingSession`
//!
//! A distinct `PendingSession` type that `poll()`s into an `IngestSession`
//! was considered and rejected. It would have to duplicate
//! `feed`/`poll_transmit`/`next_deadline`/`on_deadline` (a handshake needs
//! every one of them — that is the whole point), doubling the trait surface
//! for one bit of state; it would force each driver to hold a
//! `Pending | Established` enum and re-dispatch every call through it; and
//! `rtsp-runtime` — the in-repo precedent this mirrors — deliberately does
//! *not* do it either: `ClientSession` is one type from `Init` through
//! `Playing`, with the phase readable via `state()`. One type, with the phase
//! visible as [`HealthState::Establishing`] vs [`HealthState::Live`], keeps
//! establishment on exactly the code path everything else already uses, which
//! was the goal.
//!
//! # The handshake is bounded by a caller-supplied deadline
//!
//! A peer that opens a connection and then goes quiet must not pin a session
//! forever. [`HandshakePolicy::establish_by`] is an **absolute
//! [`Timestamp`]** by which [`SessionEvent::Established`] must have arrived;
//! past it, a still-establishing session terminates as
//! [`HealthState::HandshakeTimedOut`] — and in a [`ListenDriver`] is reaped,
//! freeing its `max_sessions` slot, so a flood of half-open connections
//! cannot squat the bound.
//!
//! **Why a deadline rather than an attempt or pump-iteration cap** (the two
//! alternatives, both rejected): the failure mode is wall-clock — "the peer
//! stopped talking" — which is exactly what `multimux`'s already-proven
//! `IngestTimeouts::connect` (`DEFAULT_CONNECT_TIMEOUT`, 10 s) bounds today,
//! so this matches a shape known to work on real cameras and encoders. An
//! iteration cap would be a proxy that misfires in both directions: a real
//! RTSP handshake is DESCRIBE + SETUP × N + PLAY where **N comes from the
//! SDP and is not knowable when the cap would have to be chosen**, so any
//! fixed number is either too small for an 8-track presentation (breaking a
//! legitimate handshake) or too large to bound a stalled one usefully. The
//! deadline also costs no new parameter — [`Timestamp`] is already threaded
//! through [`Stage::feed`]/[`Stage::on_deadline`] — and [`IngestDriver::next_deadline`]
//! surfaces it, so a real driver knows when to fire the check without
//! polling. Per this crate's sans-IO rule there is no internal timer: the
//! deadline is only observed on a `feed`/`on_deadline` the caller makes,
//! exactly like [`crate::byte_merge::MergePolicy::Failover`]'s
//! `silence_timeout`.
//!
//! *Memory* during a handshake is bounded separately and deliberately not
//! here: it is the session's own `demand()`/internal-buffer bound (the
//! `FixedFramer` precedent in [`crate::byte_stage`]'s tests), because only
//! the session knows how much partial handshake state it is legitimately
//! holding.
//!
//! # Pull sources need a typed request/response identity (round 3)
//!
//! Round 2 recorded this as a seam and stopped, deliberately, with no caller
//! yet to design against. Round 3 has the caller —
//! `multimux::source::{hls_pull, dash_pull, smooth_pull}` — and resolves it.
//!
//! The seam was correctly diagnosed back then: **`feed` was never the
//! problem.** `Stage`'s contract says nothing about chunk size and explicitly
//! decouples `poll` from `feed`, so handing one whole downloaded segment body
//! to `feed` is entirely within contract — no different from a 1316-byte UDP
//! datagram except in size. **`poll_transmit() -> Option<Bytes>` was the real
//! gap**: it expresses "send these bytes on the connection you already have",
//! right for RTSP/RTMP/SRT, but it cannot express "issue a GET for *this
//! URL*", and there is no way to route an arriving response back to the
//! request it answers when several are outstanding at once and they can
//! complete out of order.
//!
//! Two shapes were considered for closing it:
//!
//! 1. **A per-source pull method**, sitting next to `feed` rather than
//!    replacing it (e.g. `IngestSession::feed_response(id, bytes)`).
//!    Rejected: it would make `feed(&[u8])` itself unreachable for a pull
//!    session (nothing ever calls it — every real input arrives through the
//!    new method instead) while `IngestSession: Stage<In<'a> = &'a [u8]>`
//!    still advertises that `feed` as the way in. **A session whose `feed`
//!    is never called, with every real input arriving out-of-band, is a type
//!    that lies about the contract it implements.** That is not a style
//!    objection — it means the trait bound stops meaning what it says for
//!    exactly the implementors that need the escape hatch, which defeats the
//!    point of having one drive contract for the whole plane.
//! 2. **An `Inbound` enum** (`enum Inbound<'a> { Bytes(&'a [u8]), Response {
//!    id: ResourceId, bytes: &'a [u8] } }`) as the one fixed `In<'a>` every
//!    `IngestSession` uses. Rejected too: it forces every stream source
//!    (RTSP/RTMP/SRT/TS-*) to match a `Response` variant that can never occur
//!    for it, and it bakes pull vocabulary (`ResourceId`) into `media-plane`
//!    itself — this crate would then need to know what a *resource* is, which
//!    is exactly the kind of protocol knowledge the plane exists to stay free
//!    of (`ResourceId`/`Action` belong to `hls-runtime`, and the analogous
//!    DASH/Smooth identities belong to `multimux`, not here).
//!
//! What round 3 actually did: relax `Stage::In<'a>`'s pin (it is no longer
//! `&'a [u8]`, only `Out = SessionEvent` is pinned) and add
//! [`IngestSession::Request`], an opaque associated type with **no default**.
//! A byte-stream source states `type In<'a> = &'a [u8]; type Request =
//! Bytes;` — one extra line, no behaviour change (see
//! `multimux::source::ts_program::TsIngestSession`). A pull source states its
//! own honest shape instead — e.g. `type In<'a> = (HlsResourceId, &'a [u8]);
//! type Request = hls_runtime::client::Action;` — and correlates an
//! arriving response to the request that caused it via whatever identity type
//! it chose, entirely inside its own `feed`. The plane never sees a
//! `ResourceId` or an `Action`; it only ever sees "some `S::In<'_>` went in,
//! some `Option<S::Request>` came out", which is the same shape `feed`/
//! `poll_transmit` always had, just no longer forced to agree on `&[u8]`
//! across every implementor.
//!
//! # Reconnect: caller-chosen backoff, never a hardcoded sleep
//!
//! [`DialSupervisor`] bounds *how many times* [`Dialer::dial`] is retried
//! (`ReconnectPolicy::max_attempts`) but never sleeps, blocks, or otherwise
//! decides *how long* to wait between attempts — that stays entirely with
//! the caller (an async `sleep`, a `tokio::time::sleep`, nothing at all in a
//! test), matching every other bounded-but-caller-driven knob in this crate
//! ([`crate::byte_merge::MergePolicy::Failover`]'s `silence_timeout`, driven
//! by the caller's own `on_deadline` calls, not an internal timer). Once
//! [`ReconnectPolicy::max_attempts`] is exhausted, every further
//! [`DialSupervisor::try_dial`] call is an `O(1)` no-op
//! ([`DialAttempt::Exhausted`]) — it does not call [`Dialer::dial`] again,
//! so a permanently-failing dialer cannot spin the attempt count or allocate
//! per call, however many times it is polled.

use std::collections::HashMap;
use std::num::NonZeroUsize;
use std::sync::Arc;

use broadcast_common::{Stage, Timestamp};
// Only the test-only `Bytes`-`Request` sessions below reference this type
// directly now: since round 3, `IngestSession::Request` is generic, so
// production code in this module no longer names `Bytes` itself.
#[cfg(test)]
use bytes::Bytes;
use transmux::{Sample, TrackSpec};

use crate::trunk::{RetentionClass, Trunk, TrunkConfig, TrunkWriter};

/// Identifies one program within one ingest connection — see
/// [the program dimension](self#the-program-dimension-b5-sessionevent-newprogram-at-any-time).
///
/// Meaningless outside the [`IngestSession`] that assigned it: two sessions
/// each reporting `ProgramId(1)` are two unrelated programs, each getting
/// its own [`Trunk`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ProgramId(pub u32);

/// Suggested [`IngestDriver`]/[`ListenDriver`] `max_programs` bound — see
/// [the module docs](self#max_programs-the-fifth-unbounded-allocation-vector-and-why-it-needs-its-own-bound-rather-than-reusing-max_sessions-or-a-trunk-capacity)
/// for the justification. Not applied automatically (there is no default
/// constructor for `max_programs`, matching [`TrunkConfig::new`]'s and
/// [`HandshakePolicy::establish_by`]'s own all-explicit-arguments shape) —
/// a caller passes it like any other driver parameter.
///
/// Typed [`NonZeroUsize`], not `usize`: every consumer of this constant feeds
/// it to a `max_programs: NonZeroUsize` parameter, so handing back a plain
/// `usize` made every call site re-wrap it and made a forgotten wrap a
/// compile error at the *call*, not here. The default for a `NonZeroUsize`
/// parameter should already be one.
pub const DEFAULT_MAX_PROGRAMS: NonZeroUsize = match NonZeroUsize::new(64) {
    Some(n) => n,
    // Unreachable: 64 is a non-zero literal. `match` rather than `expect`
    // keeps this a `const` on the 1.86 MSRV.
    None => panic!("64 is non-zero"),
};

/// What an [`IngestSession`]'s [`Stage::poll`] hands back to a driver.
///
/// `#[non_exhaustive]`: a later step (egress track-set negotiation, §1.3's
/// upstream program-split) may still need a `ProgramEnded` variant —
/// [`SessionEvent::TracksChanged`] (issue #781) closed the mid-stream
/// track-set half of this gap, but this step does not add a variant it has
/// no correct producer for yet (this crate's own precedent —
/// [`crate::byte_merge`]'s `Hitless2022_7` note).
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum SessionEvent {
    /// The handshake finished: this session's transport is usable and it is
    /// now live. Exactly one of these per session lifetime — see
    /// [Establishment is ordinary driving](self#establishment-is-ordinary-driving--dial-performs-no-io).
    /// Until it arrives the driver reports [`HealthState::Establishing`];
    /// after it, [`HealthState::Live`].
    ///
    /// # Why not called `TracksResolved`
    ///
    /// `transmux::DemuxEvent::TracksResolved { generation }` already owns that
    /// name for the analogous *demux-side* question, and this is deliberately
    /// not a synonym for it: **tracks here are a per-program fact** carried by
    /// [`SessionEvent::NewProgram`] (finding B5 — one connection, N programs),
    /// whereas establishment is a per-*connection* fact. Folding tracks into
    /// this variant would force a session ingesting an MPTS to nominate some
    /// arbitrary program as "the one that resolved the connection", and would
    /// leave a session that has finished its handshake but not yet seen a PAT
    /// with no way to say so. It carries no payload for the same
    /// "no field without a correct producer" reason.
    ///
    /// The two events do line up on the awkward case, though, and this is
    /// where that lands: `DemuxEvent::TracksResolved`'s own docs note that a
    /// container with no up-front track declaration (FLV/RTMP) legitimately
    /// never emits it, and that the asymmetry "is for the media plane's
    /// ingress layer to handle explicitly (e.g. gating on the first
    /// `DemuxEvent::Sample`)". This variant is that explicit handling: an
    /// RTMP session decides for itself when it is ready — gating on the first
    /// sample, exactly as those docs suggest — and says so here.
    Established,
    /// A new program was discovered — mint a fresh [`Trunk`] for it. May be
    /// the very first event a session ever produces, or arrive after many
    /// samples for other programs already have — both are the same case;
    /// see [the module docs](self#the-program-dimension-b5-sessionevent-newprogram-at-any-time).
    NewProgram {
        /// Identifies this program for every subsequent
        /// [`SessionEvent::Sample`] carrying it.
        program: ProgramId,
        /// The demuxed track specs for this program, so far. Reuses
        /// [`transmux::TrackSpec`] rather than inventing a parallel type —
        /// this crate's established pattern (see `crate::trunk::SegmentEntry`'s
        /// module doc for the same reuse-don't-duplicate reasoning).
        tracks: Vec<TrackSpec>,
    },
    /// One decoded sample for `track_id`, belonging to `program` (must have
    /// been announced via a prior `NewProgram` with this `program` — a
    /// `Sample` for an unannounced program is a contract violation by the
    /// `IngestSession` implementor, and a driver drops it rather than
    /// panicking; see [`IngestDriver`]'s docs).
    Sample {
        /// Which program this sample belongs to.
        program: ProgramId,
        /// Track id within that program, matching
        /// [`TrunkWriter::publish`]'s own `track_id`.
        track_id: u32,
        /// Which ring this sample's track publishes into — the publisher
        /// (ultimately, the `IngestSession` implementor) decides this, same
        /// as any other [`TrunkWriter::publish`] caller.
        retention: RetentionClass,
        /// The decoded sample itself.
        sample: Sample,
    },
    /// `program`'s track set changed mid-stream (issue #781) — e.g.
    /// transmux's PMT version diffing detects a broadcaster adding an audio
    /// language. Must have been announced via a prior `NewProgram` with this
    /// `program`, exactly like [`SessionEvent::Sample`] — a `TracksChanged`
    /// for an unannounced program is the identical contract violation, and a
    /// driver drops it the identical way (see [`IngestDriver`]'s docs); it
    /// never mints a `Trunk` on its own.
    ///
    /// # Why `tracks` is the complete replacement set, not a delta
    ///
    /// A PMT carries the **whole** elementary-stream list on every version
    /// bump, not just what changed — there is no "here is the one track
    /// that was added" signal at that layer, only "here is the program's
    /// full track list, as of now". Carrying the complete set here mirrors
    /// that fact rather than fighting it, and buys two things a delta
    /// encoding cannot:
    ///
    /// - **Idempotence.** Re-delivering the same `TracksChanged` twice (a
    ///   retried demux pass, a duplicate event) leaves the trunk's track set
    ///   unchanged in content — replacing a set with an identical set is a
    ///   no-op in substance, whereas replaying an "add track" delta twice
    ///   would double-add it.
    /// - **Immunity to delta-ordering bugs.** A dropped or reordered
    ///   `TrackAdded`/`TrackRemoved` pair (exactly the demux-layer events
    ///   `transmux` already emits — see below) can never leave a consumer's
    ///   view of the track set permanently wrong: the next `TracksChanged`
    ///   is a fresh, authoritative snapshot, not an increment on top of
    ///   whatever state happened to accumulate.
    ///
    /// A consumer that cares *which* track appeared or vanished diffs this
    /// snapshot against the previous one it already holds (or against
    /// [`crate::Trunk::tracks`], which this event's application updates) —
    /// that comparison is the consumer's to make, not this event's to
    /// pre-compute.
    ///
    /// # Why this layer does not mirror `transmux::DemuxEvent`'s three events
    ///
    /// `transmux` already emits `DemuxEvent::TrackAdded`/`TrackRemoved`/
    /// `TrackUpdated` at the demux layer — finer-grained, delta-shaped
    /// events aimed at a caller that wants to react to *what changed*. This
    /// layer deliberately does not mirror that shape: `IngestSession`
    /// implementors translate whatever demux-layer deltas they see into one
    /// full snapshot per change, for the same reason [`SessionEvent::NewProgram`]
    /// carries a full `tracks: Vec<TrackSpec>` rather than a "here is track
    /// N" event per track — see [the module docs](self#the-program-dimension-b5-sessionevent-newprogram-at-any-time).
    ///
    /// # Scope: ingress→`Trunk` plumbing only
    ///
    /// This variant and [`IngestDriver`]'s handling of it stop at storing
    /// the new set on the program's `Trunk` (see
    /// [`crate::TrunkWriter::set_tracks`]). Deciding whether/when to admit a
    /// newly-appeared track into an egress manifest (LL-HLS/DASH rendering)
    /// is a separate, deliberate decision belonging to its own issue — not
    /// something a track-set snapshot arriving at the `Trunk` should trigger
    /// implicitly.
    TracksChanged {
        /// Which program's track set changed — must match a program already
        /// announced via [`SessionEvent::NewProgram`].
        program: ProgramId,
        /// The complete replacement track set — see this variant's own doc
        /// for why this is a full snapshot rather than a delta.
        tracks: Vec<TrackSpec>,
    },
}

/// The sans-IO ingress drive contract: a specialisation of [`Stage`] whose
/// output is [`SessionEvent`] — see
/// [the module docs](self#ingestsession-is-a-stage-matching-bytestages-precedent).
/// `Stage::In<'a>` is **not** pinned here (round 3 relaxed it from `&'a
/// [u8]`) — see
/// [Pull sources need a typed request/response identity](self#pull-sources-need-a-typed-requestresponse-identity-round-3)
/// for why: a byte-stream source still states `type In<'a> = &'a [u8]`, but a
/// pull source (HLS/DASH/Smooth) states its own request/response identity
/// instead.
///
/// # Why this is explicitly implemented, unlike [`crate::ByteStage`]
///
/// `ByteStage` gets a blanket `impl<T> ByteStage for T where T: Stage<…>`
/// because it adds **nothing** to `Stage` — it is a pure alias, so a blanket
/// impl costs nothing and saves every implementor a line. `IngestSession`
/// adds [`poll_transmit`](Self::poll_transmit) (which has a default) and
/// [`Request`](Self::Request) (which, being an associated type, cannot have
/// one). A blanket impl here would make `poll_transmit`'s default
/// **impossible to override** (the blanket would already be the one impl for
/// every type, and a second manual impl would collide), quietly breaking the
/// handshake mechanism it exists for — and could not exist at all once
/// `Request` has no default value to blanket-supply. So implementors write at
/// least one extra line — `type Request = Bytes;` for every byte-stream
/// source, plus a real `poll_transmit` body for the two or three that send
/// back. This asymmetry with `ByteStage` is deliberate and is the reason it
/// is documented rather than "fixed".
pub trait IngestSession: for<'a> Stage<Out = SessionEvent> + Send {
    /// What [`poll_transmit`](Self::poll_transmit) hands back — opaque to the
    /// plane, deliberately: see
    /// [the module docs](self#pull-sources-need-a-typed-requestresponse-identity-round-3)
    /// for why this is not a fixed enum. A byte-stream source (RTSP/RTMP/SRT/
    /// TS-*) sets this to [`bytes::Bytes`]; a pull source sets it to its own
    /// protocol's action type (e.g. `hls_runtime::client::Action`).
    ///
    /// No default: unlike `poll_transmit`, there is no value every
    /// implementor could reasonably start from, so every `IngestSession`
    /// names its own type explicitly, one line, even the sessions that never
    /// override `poll_transmit`'s body.
    type Request: Send;

    /// The next outbound request this session wants performed — bytes
    /// written to an already-open connection, or (for a pull source) a
    /// fetch/wait action for the driver to carry out.
    ///
    /// Two uses of the byte-stream case, one mechanism: the **handshake**
    /// requests that establish the session (an RTSP `DESCRIBE`, then `SETUP`,
    /// then `PLAY` — see
    /// [Establishment is ordinary driving](self#establishment-is-ordinary-driving--dial-performs-no-io)),
    /// and in-session traffic afterwards (RTCP receiver reports, an RTSP
    /// keepalive `OPTIONS`, an SRT ACK). This is exactly what
    /// `rtsp_runtime::client::ClientSession`'s request builders return, only
    /// pulled rather than returned. A pull source's own [`Request`](Self::Request)
    /// plays the identical role in its own protocol — see e.g.
    /// `hls_runtime::client::Action`, returned unchanged through this
    /// method by an HLS-pull `IngestSession`.
    ///
    /// A driver drains this in a loop after every
    /// [`Stage::feed`]/[`Stage::on_deadline`] (and once immediately after
    /// [`Dialer::dial`], to send the first handshake request), exactly like
    /// [`Stage::poll`]. A session with nothing to send never overrides it.
    fn poll_transmit(&mut self) -> Option<Self::Request> {
        None
    }
}

/// Outbound connect: RTSP, raw RTP/UDP, TS-over-UDP, an SRT caller, an
/// HLS/DASH/Smooth pull client.
pub trait Dialer: Send {
    /// The session a dial produces.
    type Session: IngestSession;
    /// Why constructing the session failed.
    type Error;

    /// **Construct** a session — performing no I/O and completing no
    /// handshake.
    ///
    /// The returned session is *not yet established*: it starts in
    /// [`HealthState::Establishing`], and the handshake completes through the
    /// ordinary pump ([`IngestSession::poll_transmit`] out,
    /// [`Stage::feed`] in) until it emits [`SessionEvent::Established`] — see
    /// [Establishment is ordinary driving](self#establishment-is-ordinary-driving--dial-performs-no-io).
    /// An implementation should queue its first handshake request for
    /// `poll_transmit` here.
    ///
    /// The `Err` path is for purely local construction failures — a URL that
    /// will not parse, contradictory config — **not** for connect failures,
    /// which this method never attempts and therefore cannot observe. A peer
    /// that refuses or never answers surfaces later, as
    /// [`HealthState::Failed`] or [`HealthState::HandshakeTimedOut`].
    fn dial(&mut self) -> Result<Self::Session, Self::Error>;
}

/// Identifies one session a [`ListenDriver`] currently has admitted, for
/// every call that needs to name which one ([`ListenDriver::feed`],
/// [`ListenDriver::health`], ...).
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct SessionId(pub u64);

/// Inbound accept: RTMP push, an SRT listener, a future WHIP. See
/// [`max_sessions` is enforced by the driver](self#listener-and-max_sessions-enforced-by-the-driver-not-by-convention).
pub trait Listener: Send {
    /// The session one accepted connection produces.
    type Session: IngestSession;
    /// Why an accept attempt failed.
    type Error;

    /// The hard bound on concurrently admitted sessions — see the module
    /// docs. A fixed accessor, not a mutable knob: this step has no use case
    /// for changing it mid-flight, and a fixed value is what lets
    /// [`ListenDriver`] reason about it as a real bound rather than a
    /// point-in-time snapshot that might already be stale.
    fn max_sessions(&self) -> usize;

    /// Try to accept the next inbound connection. `Ok(None)` means nothing
    /// is waiting right now (a non-blocking poll, matching [`Stage::poll`]'s
    /// shape) — not an error and not end-of-input; a [`Listener`] has no
    /// "end" the way a single connection does.
    fn poll_accept(&mut self) -> Result<Option<Self::Session>, Self::Error>;
}

/// A session's phase, distinguishing "still handshaking" from "live", and a
/// clean end from a real failure — see
/// [Supervision: EOF is not an error](self#supervision-eof-is-not-an-error-the-healthstate-fix).
///
/// [`Establishing`](Self::Establishing) and [`Live`](Self::Live) are the two
/// running states; the other three are terminal (in a [`ListenDriver`],
/// reaching any of them reaps the session and frees its `max_sessions` slot).
///
/// `#[non_exhaustive]`: a later step may add `Reconnecting` once a driver owns
/// a full redial loop rather than just the bounded initial dial
/// [`DialSupervisor`] covers in this step.
#[derive(Debug)]
#[non_exhaustive]
pub enum HealthState<E> {
    /// Constructed by [`Dialer::dial`] (or accepted by a [`Listener`]) but
    /// the handshake has not finished: the session has not yet emitted
    /// [`SessionEvent::Established`]. Bounded by
    /// [`HandshakePolicy::establish_by`] — see
    /// [the handshake is bounded](self#the-handshake-is-bounded-by-a-caller-supplied-deadline).
    Establishing,
    /// Established and actively driving; no end or error observed yet.
    Live,
    /// An [`IngestSession`]'s [`Stage::finish`] returned `Ok(())` with no
    /// prior error — the source ended on its own; this is not a failure.
    Ended,
    /// An [`IngestSession`]'s [`Stage::feed`] or [`Stage::finish`] returned
    /// `Err`. Carries the concrete error rather than a formatted string, so a
    /// caller that cares can match on it.
    Failed(E),
    /// [`HandshakePolicy::establish_by`] passed while the session was still
    /// [`Establishing`](Self::Establishing) — the peer opened a connection and
    /// never completed the handshake.
    ///
    /// Deliberately **not** folded into [`Failed`](Self::Failed): that variant
    /// carries the *session's* own error type, and a handshake that simply
    /// never progressed produced no session error to carry — the session did
    /// nothing wrong, it was starved of input. Inventing an `E` to put here
    /// would mean either fabricating one or forcing every implementor's error
    /// type to grow a timeout variant it cannot itself raise.
    HandshakeTimedOut {
        /// The deadline that passed.
        deadline: Timestamp,
    },
}

impl<E: PartialEq> PartialEq for HealthState<E> {
    fn eq(&self, other: &Self) -> bool {
        match (self, other) {
            (HealthState::Establishing, HealthState::Establishing) => true,
            (HealthState::Live, HealthState::Live) => true,
            (HealthState::Ended, HealthState::Ended) => true,
            (HealthState::Failed(a), HealthState::Failed(b)) => a == b,
            (
                HealthState::HandshakeTimedOut { deadline: a },
                HealthState::HandshakeTimedOut { deadline: b },
            ) => a == b,
            _ => false,
        }
    }
}

impl<E> HealthState<E> {
    /// `true` while this session is still being driven —
    /// [`Establishing`](Self::Establishing) or [`Live`](Self::Live). `false`
    /// once it has reached a terminal state.
    pub fn is_running(&self) -> bool {
        matches!(self, HealthState::Establishing | HealthState::Live)
    }
}

/// Bounds how long a session may stay in [`HealthState::Establishing`] — see
/// [the handshake is bounded](self#the-handshake-is-bounded-by-a-caller-supplied-deadline)
/// for why this is a wall-clock deadline rather than an attempt or
/// pump-iteration cap.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct HandshakePolicy {
    /// Absolute [`Timestamp`], on the same driver-chosen epoch as
    /// [`Stage::feed`]'s `now`, by which [`SessionEvent::Established`] must
    /// have arrived.
    pub establish_by: Timestamp,
}

impl HandshakePolicy {
    /// Require the handshake to complete by the absolute timestamp
    /// `establish_by`.
    pub fn establish_by(establish_by: Timestamp) -> Self {
        HandshakePolicy { establish_by }
    }
}

/// Drives one connected [`IngestSession`], dispatching every
/// [`SessionEvent`] it yields into a fresh per-[`ProgramId`] [`Trunk`], and
/// tracking [`HealthState`] — the "pump that owns the feed/poll/deadline
/// loop" for a single dialed-out connection (see [`run_dial`]). [`ListenDriver`]
/// embeds one of these per admitted session, so this is also where an
/// accepted connection's per-program `Trunk` bookkeeping actually lives.
pub struct IngestDriver<S: IngestSession> {
    session: S,
    trunk_config: TrunkConfig,
    handshake: HandshakePolicy,
    max_programs: NonZeroUsize,
    programs: HashMap<ProgramId, Arc<Trunk>>,
    writers: HashMap<ProgramId, TrunkWriter>,
    /// Count of `NewProgram` events refused because `max_programs` was
    /// already reached — see
    /// [the module docs](self#max_programs-the-fifth-unbounded-allocation-vector-and-why-it-needs-its-own-bound-rather-than-reusing-max_sessions-or-a-trunk-capacity).
    /// A counter, not a stored list, so this field is itself `O(1)` no
    /// matter how large a flood of refused programs is.
    refused_programs: u64,
    health: HealthState<S::Error>,
}

impl<S: IngestSession> IngestDriver<S> {
    /// Wrap a freshly-constructed (**not yet established**) `session`, ready
    /// to be pumped: it starts in [`HealthState::Establishing`] and reaches
    /// [`HealthState::Live`] when it emits [`SessionEvent::Established`],
    /// bounded by `handshake`. Every program it later announces gets a fresh
    /// [`Trunk`] built from `trunk_config`, up to `max_programs` distinct
    /// [`ProgramId`]s — see
    /// [the module docs](self#max_programs-the-fifth-unbounded-allocation-vector-and-why-it-needs-its-own-bound-rather-than-reusing-max_sessions-or-a-trunk-capacity)
    /// for what happens past that bound.
    pub fn new(
        session: S,
        trunk_config: TrunkConfig,
        handshake: HandshakePolicy,
        max_programs: NonZeroUsize,
    ) -> Self {
        IngestDriver {
            session,
            trunk_config,
            handshake,
            max_programs,
            programs: HashMap::new(),
            writers: HashMap::new(),
            refused_programs: 0,
            health: HealthState::Establishing,
        }
    }

    /// The bound this driver enforces on distinct admitted programs.
    pub fn max_programs(&self) -> NonZeroUsize {
        self.max_programs
    }

    /// Currently-admitted distinct program count. Never exceeds
    /// [`Self::max_programs`], however many `NewProgram` events this session
    /// reports.
    pub fn program_count(&self) -> usize {
        self.programs.len()
    }

    /// How many `NewProgram` events this driver has refused because
    /// [`Self::max_programs`] was already reached — see
    /// [the module docs](self#max_programs-the-fifth-unbounded-allocation-vector-and-why-it-needs-its-own-bound-rather-than-reusing-max_sessions-or-a-trunk-capacity).
    /// Monotonically increasing; never resets.
    pub fn refused_program_count(&self) -> u64 {
        self.refused_programs
    }

    /// Feed more input read from this session's connection — a handshake
    /// response while [`HealthState::Establishing`], media once
    /// [`HealthState::Live`]; the same call either way. Generic over
    /// `Stage::In` (round 3: no longer pinned to `&[u8]`) so a pull
    /// source can feed its own `(id, bytes)` response shape through the same
    /// method a byte-stream source feeds raw bytes through — see
    /// [the module docs](self#pull-sources-need-a-typed-requestresponse-identity-round-3).
    /// A no-op once the session has reached a terminal state — it is never
    /// fed again.
    pub fn feed(&mut self, input: S::In<'_>, now: Timestamp) {
        if !self.health.is_running() {
            return;
        }
        match self.session.feed(input, now) {
            Ok(()) => {
                self.drain();
                self.check_handshake_deadline(now);
            }
            Err(e) => self.health = HealthState::Failed(e),
        }
    }

    /// Let the session act on the passage of time (an in-flight handshake
    /// retransmit, rate-scheduled re-emission, a keepalive interval) — see
    /// [`Stage::on_deadline`]. Also where a blown
    /// [`HandshakePolicy::establish_by`] is observed for a peer that has gone
    /// silent mid-handshake and so is producing no `feed` calls at all. A
    /// no-op once terminated, matching [`Self::feed`].
    pub fn on_deadline(&mut self, now: Timestamp) {
        if !self.health.is_running() {
            return;
        }
        self.session.on_deadline(now);
        self.drain();
        self.check_handshake_deadline(now);
    }

    /// Signal clean end-of-input. `Ok(())` from the session drives
    /// [`HealthState::Ended`] (not a failure); `Err` drives
    /// [`HealthState::Failed`] — this is the method the mutation-checked
    /// EOF-vs-failure test in this module drives directly. A no-op once
    /// already terminated.
    pub fn finish(&mut self) {
        if !self.health.is_running() {
            return;
        }
        match self.session.finish() {
            Ok(()) => {
                self.drain();
                self.health = HealthState::Ended;
            }
            Err(e) => self.health = HealthState::Failed(e),
        }
    }

    /// Drain the next outbound request the session wants performed —
    /// handshake requests included, and (for a pull source) a fetch/wait
    /// action. See [`IngestSession::poll_transmit`]/[`IngestSession::Request`].
    pub fn poll_transmit(&mut self) -> Option<S::Request> {
        self.session.poll_transmit()
    }

    /// The next point in time this driver has work to do: the earlier of the
    /// session's own [`Stage::next_deadline`] and — while still
    /// [`HealthState::Establishing`] — [`HandshakePolicy::establish_by`], so a
    /// caller driving off this value alone still learns about a stalled
    /// handshake at the right moment rather than never.
    pub fn next_deadline(&self) -> Option<Timestamp> {
        let session = self.session.next_deadline();
        let handshake =
            matches!(self.health, HealthState::Establishing).then_some(self.handshake.establish_by);
        match (session, handshake) {
            (Some(a), Some(b)) => Some(a.min(b)),
            (a, b) => a.or(b),
        }
    }

    /// This session's current health.
    pub fn health(&self) -> &HealthState<S::Error> {
        &self.health
    }

    /// Consume this driver, yielding its final [`HealthState`] **by value** —
    /// so a caller that is tearing the route down can move the concrete
    /// `S::Error` out of [`HealthState::Failed`] and return it.
    ///
    /// [`Self::health`] only lends a `&HealthState`, which is right for
    /// polling but cannot hand back the error: a session error type is not
    /// required to be [`Clone`] (`multimux::MultimuxError` is not), so a
    /// borrowing accessor forces a caller to degrade the typed error into a
    /// formatted string — exactly the loss this module's docs call out as
    /// the bug `HealthState<E>` exists to fix. Without this, an
    /// [`IngestSession`] whose `feed` returns `Err` is *unreportable* by its
    /// own driver loop: `feed` records the error in `health` and returns
    /// `()`, so a loop that only ever calls `feed` sees no failure at all and
    /// spins forever. (That is not hypothetical — it is exactly how
    /// `multimux::source::smooth_pull`'s PlayReady-detection error escaped
    /// its drive loop until this method existed.)
    ///
    /// Consuming (rather than a `&mut` "take the error out") is deliberate:
    /// every terminal state is final, so there is no valid use for a driver
    /// whose failure has been moved out from under it. Any [`Trunk`] this
    /// driver minted stays alive independently — they are [`Arc`]s a caller
    /// will already have cloned out via [`Self::trunk`].
    pub fn into_health(self) -> HealthState<S::Error> {
        self.health
    }

    /// Terminate a still-`Establishing` session whose deadline has passed.
    ///
    /// Called *after* the session has been fed/drained, never before, so a
    /// handshake response that arrives exactly at the deadline and completes
    /// the handshake still establishes rather than being rejected by a
    /// millisecond.
    fn check_handshake_deadline(&mut self, now: Timestamp) {
        if matches!(self.health, HealthState::Establishing) && now >= self.handshake.establish_by {
            self.health = HealthState::HandshakeTimedOut {
                deadline: self.handshake.establish_by,
            };
        }
    }

    /// The [`Trunk`] for `program`, if it has been announced yet.
    pub fn trunk(&self, program: ProgramId) -> Option<&Arc<Trunk>> {
        self.programs.get(&program)
    }

    /// Every program this session has announced so far.
    pub fn programs(&self) -> impl Iterator<Item = ProgramId> + '_ {
        self.programs.keys().copied()
    }

    /// Read-only access to the underlying session — for state a driver loop
    /// needs to observe beyond what [`SessionEvent`]/[`HealthState`] already
    /// expose. Concretely: a pull source (HLS/DASH/Smooth) knows it has
    /// reached true end-of-stream (the origin's playlist/manifest said so
    /// *and* every outstanding fetch it named is accounted for) entirely from
    /// its own protocol bookkeeping — unlike a byte-stream transport, whose
    /// "ended" signal is external (the HTTP body closed, the socket peer
    /// disconnected) and therefore already known to the driver loop without
    /// reaching in here at all. `SessionEvent` deliberately has no `Ended`
    /// variant to carry that (see its own doc's "no field/variant without a
    /// correct producer" discipline), so a pull source's own inherent
    /// accessor (e.g. `HlsIngestSession::ended`) is what a driver loop reads
    /// to decide when to call [`Self::finish`].
    pub fn session(&self) -> &S {
        &self.session
    }

    /// Drain every ready [`SessionEvent`], dispatching each into its
    /// program's `Trunk`. A `Sample` for a program never announced via
    /// `NewProgram` is dropped (documented `IngestSession` contract
    /// violation, not a panic — see [`SessionEvent::Sample`]'s docs); a
    /// `NewProgram` past `max_programs` is refused the exact same way — see
    /// [the module docs](self#max_programs-the-fifth-unbounded-allocation-vector-and-why-it-needs-its-own-bound-rather-than-reusing-max_sessions-or-a-trunk-capacity).
    fn drain(&mut self) {
        while let Some(event) = self.session.poll() {
            match event {
                SessionEvent::Established => {
                    // Only ever a promotion out of Establishing: a duplicate
                    // Established from a misbehaving session must not resurrect
                    // an already-terminal driver, and must not be treated as a
                    // fresh start for a live one.
                    if matches!(self.health, HealthState::Establishing) {
                        self.health = HealthState::Live;
                    }
                }
                SessionEvent::NewProgram { program, tracks } => {
                    // A repeat announcement of an already-admitted program
                    // does not grow `self.programs`, so it is never refused
                    // here regardless of how full the bound is — only a
                    // genuinely new `ProgramId` can hit the cap.
                    if !self.programs.contains_key(&program)
                        && self.programs.len() >= self.max_programs.get()
                    {
                        // Refused: no `Trunk::new` call happens at all (not
                        // merely an unstored one), and no `writers` entry is
                        // created, so this program's later `Sample`s fall
                        // through the existing "unannounced program" drop
                        // path below rather than needing a second one.
                        self.refused_programs += 1;
                        continue;
                    }
                    // A REPEAT announcement updates the existing `Trunk` in
                    // place; it must never mint a replacement. Re-minting
                    // would swap a fresh, empty `Trunk` into `self.programs`
                    // while every already-issued cursor kept reading the
                    // orphaned one that no longer receives writes — existing
                    // subscribers would see a permanently stalled stream, and
                    // whatever the old `Trunk` still held (samples, segments,
                    // parts, and so the DVR window) would be silently dropped.
                    //
                    // Treating the repeat as a track-set update is exactly
                    // what `TracksChanged` does, so this defers to the same
                    // path rather than duplicating it: a re-announcement is a
                    // restatement of the program's tracks, not a new program.
                    if let Some(writer) = self.writers.get(&program) {
                        writer.set_tracks(tracks);
                        continue;
                    }
                    let trunk = Trunk::new(self.trunk_config);
                    let writer = trunk
                        .writer()
                        .expect("a freshly constructed Trunk always has an unclaimed writer");
                    // Seed the freshly-minted Trunk's track set from this
                    // event's `tracks` (issue #781) — previously discarded
                    // entirely (the `..` this match arm used to bind with),
                    // leaving every Trunk's track set permanently empty.
                    writer.set_tracks(tracks);
                    self.programs.insert(program, trunk);
                    self.writers.insert(program, writer);
                }
                SessionEvent::Sample {
                    program,
                    track_id,
                    retention,
                    sample,
                } => {
                    if let Some(writer) = self.writers.get(&program) {
                        writer.publish(track_id, retention, sample);
                    }
                }
                SessionEvent::TracksChanged { program, tracks } => {
                    // Same "unannounced program is a dropped contract
                    // violation, not a panic" contract as `Sample` above —
                    // reuses the exact same `writers` lookup, not a second
                    // drop path.
                    if let Some(writer) = self.writers.get(&program) {
                        writer.set_tracks(tracks);
                    }
                }
            }
        }
    }
}

/// Construct a session via [`Dialer::dial`] and wrap it for driving — see
/// [`IngestDriver`]. Performs **no I/O and completes no handshake**: the
/// returned driver starts in [`HealthState::Establishing`], and the caller
/// pumps it ([`IngestDriver::poll_transmit`] out, [`IngestDriver::feed`] in)
/// until it reports [`HealthState::Live`]. [`DialSupervisor`] adds bounded
/// retry on top for a `Dialer` whose local construction fails outright.
pub fn run_dial<D: Dialer>(
    dialer: &mut D,
    trunk_config: TrunkConfig,
    handshake: HandshakePolicy,
    max_programs: NonZeroUsize,
) -> Result<IngestDriver<D::Session>, D::Error> {
    let session = dialer.dial()?;
    Ok(IngestDriver::new(
        session,
        trunk_config,
        handshake,
        max_programs,
    ))
}

/// Bounded retry policy for [`DialSupervisor`] — how many times
/// [`Dialer::dial`] is retried before giving up, never how long to wait
/// between attempts (that stays with the caller; see
/// [Reconnect](self#reconnect-caller-chosen-backoff-never-a-hardcoded-sleep)).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct ReconnectPolicy {
    /// Maximum number of consecutive [`Dialer::dial`] attempts before
    /// [`DialSupervisor::try_dial`] gives up.
    pub max_attempts: u32,
}

impl ReconnectPolicy {
    /// Build a policy bounded to `max_attempts` consecutive dial failures.
    ///
    /// Panics if `max_attempts == 0` — a policy that never even tries once
    /// is a construction mistake, not a real bound.
    pub fn new(max_attempts: u32) -> Self {
        assert!(max_attempts > 0, "ReconnectPolicy max_attempts must be > 0");
        ReconnectPolicy { max_attempts }
    }
}

/// The result of one [`DialSupervisor::try_dial`] call.
///
/// Not `Debug`: [`DialAttempt::Connected`] carries an [`IngestDriver`], which
/// carries a `Trunk`/`TrunkWriter` — neither implements `Debug` (a `Trunk`
/// holds live synchronization primitives, not inspectable state), so this
/// type cannot either without breaking that up.
#[non_exhaustive]
pub enum DialAttempt<S: IngestSession, E> {
    /// A session was constructed — a driveable [`IngestDriver`], starting in
    /// [`HealthState::Establishing`]. Named `Connected` for the caller's
    /// mental model of "the dial step succeeded"; nothing is connected yet in
    /// the I/O sense (see [`Dialer::dial`]).
    Connected(IngestDriver<S>),
    /// This attempt failed, but retries remain: wait (your own chosen
    /// backoff) and call [`DialSupervisor::try_dial`] again.
    Retry(E),
    /// This attempt failed and it was the last one allowed by
    /// [`ReconnectPolicy::max_attempts`] — carries the error that caused
    /// this final attempt to fail.
    GaveUp(E),
    /// [`DialAttempt::GaveUp`] already fired on an earlier call: no new
    /// dial was attempted this time, and none ever will be again from this
    /// [`DialSupervisor`] — see
    /// [Reconnect](self#reconnect-caller-chosen-backoff-never-a-hardcoded-sleep)
    /// for why this is what keeps a permanently-failing dialer from
    /// spinning.
    Exhausted,
}

/// Bounds [`Dialer::dial`] retry — see [`ReconnectPolicy`] and
/// [Reconnect](self#reconnect-caller-chosen-backoff-never-a-hardcoded-sleep).
pub struct DialSupervisor<D: Dialer> {
    dialer: D,
    policy: ReconnectPolicy,
    attempts: u32,
    exhausted: bool,
}

impl<D: Dialer> DialSupervisor<D> {
    /// Build a supervisor over `dialer`, bounded by `policy`.
    pub fn new(dialer: D, policy: ReconnectPolicy) -> Self {
        DialSupervisor {
            dialer,
            policy,
            attempts: 0,
            exhausted: false,
        }
    }

    /// Consecutive failed attempts so far. Reset to `0` on a successful
    /// dial; never exceeds [`ReconnectPolicy::max_attempts`], regardless of
    /// how many times [`Self::try_dial`] is called afterward — see
    /// [`DialAttempt::Exhausted`].
    pub fn attempts(&self) -> u32 {
        self.attempts
    }

    /// `true` once [`ReconnectPolicy::max_attempts`] has been exhausted —
    /// every subsequent [`Self::try_dial`] call returns
    /// [`DialAttempt::Exhausted`] without touching [`Dialer::dial`] again.
    pub fn is_exhausted(&self) -> bool {
        self.exhausted
    }

    /// Try once more to dial, wrapping success into a driveable
    /// [`IngestDriver`] (starting in [`HealthState::Establishing`], bounded by
    /// `handshake`; every program it later announces gets a `Trunk` built from
    /// `trunk_config`). Never sleeps — see the module docs.
    pub fn try_dial(
        &mut self,
        trunk_config: TrunkConfig,
        handshake: HandshakePolicy,
        max_programs: NonZeroUsize,
    ) -> DialAttempt<D::Session, D::Error> {
        if self.exhausted {
            return DialAttempt::Exhausted;
        }
        self.attempts += 1;
        match self.dialer.dial() {
            Ok(session) => {
                self.attempts = 0;
                DialAttempt::Connected(IngestDriver::new(
                    session,
                    trunk_config,
                    handshake,
                    max_programs,
                ))
            }
            Err(e) => {
                if self.attempts >= self.policy.max_attempts {
                    self.exhausted = true;
                    DialAttempt::GaveUp(e)
                } else {
                    DialAttempt::Retry(e)
                }
            }
        }
    }
}

/// The outcome of one [`ListenDriver::poll_accept`] call.
#[derive(Debug)]
#[non_exhaustive]
pub enum AcceptOutcome<E> {
    /// A new connection was accepted and admitted under `max_sessions`; use
    /// this id with [`ListenDriver::feed`]/[`ListenDriver::health`]/etc.
    Admitted(SessionId),
    /// Nothing was waiting to be accepted right now — not an error.
    Idle,
    /// A connection was accepted, but `max_sessions` was already reached: it
    /// was dropped immediately, without ever being fed a byte — see
    /// [`max_sessions` is enforced by the driver](self#listener-and-max_sessions-enforced-by-the-driver-not-by-convention).
    Refused,
    /// [`Listener::poll_accept`] itself reported a failure (distinct from a
    /// refusal: the transport-level accept failed, not the admission bound).
    Error(E),
}

/// Drives a [`Listener`], admitting up to its [`Listener::max_sessions`]
/// concurrently and dispatching every admitted session's [`SessionEvent`]s
/// into per-[`ProgramId`] [`Trunk`]s exactly like [`IngestDriver`] (one is
/// embedded per admitted session). See
/// [`max_sessions` is enforced by the driver](self#listener-and-max_sessions-enforced-by-the-driver-not-by-convention).
pub struct ListenDriver<L: Listener> {
    listener: L,
    trunk_config: TrunkConfig,
    handshake: HandshakePolicy,
    max_programs: NonZeroUsize,
    sessions: HashMap<SessionId, IngestDriver<L::Session>>,
    next_id: u64,
}

impl<L: Listener> ListenDriver<L> {
    /// Build a driver over `listener`. Every admitted session starts in
    /// [`HealthState::Establishing`] bounded by `handshake` — which is what
    /// stops a flood of half-open inbound connections from squatting the
    /// `max_sessions` bound indefinitely — and every program any of them
    /// announces gets a `Trunk` built from `trunk_config`, up to
    /// `max_programs` distinct programs per session — see
    /// [the module docs](self#max_programs-the-fifth-unbounded-allocation-vector-and-why-it-needs-its-own-bound-rather-than-reusing-max_sessions-or-a-trunk-capacity).
    pub fn new(
        listener: L,
        trunk_config: TrunkConfig,
        handshake: HandshakePolicy,
        max_programs: NonZeroUsize,
    ) -> Self {
        ListenDriver {
            listener,
            trunk_config,
            handshake,
            max_programs,
            sessions: HashMap::new(),
            next_id: 0,
        }
    }

    /// The per-session program bound this driver enforces — see
    /// [`IngestDriver::max_programs`].
    pub fn max_programs(&self) -> NonZeroUsize {
        self.max_programs
    }

    /// How many `NewProgram` events session `id` has refused because
    /// `max_programs` was already reached — see
    /// [`IngestDriver::refused_program_count`]. `None` for an unknown or
    /// already-reaped `id`.
    pub fn refused_program_count(&self, id: SessionId) -> Option<u64> {
        self.sessions
            .get(&id)
            .map(IngestDriver::refused_program_count)
    }

    /// Currently-admitted (not yet terminated) session count. Never exceeds
    /// [`Listener::max_sessions`].
    pub fn session_count(&self) -> usize {
        self.sessions.len()
    }

    /// The bound this driver enforces, from the underlying [`Listener`].
    pub fn max_sessions(&self) -> usize {
        self.listener.max_sessions()
    }

    /// Try to accept one more connection, enforcing `max_sessions` (see the
    /// module docs). Calling this in a tight loop with nothing waiting, or
    /// with the bound already reached, is `O(1)` per call and never grows
    /// [`Self::session_count`] past [`Listener::max_sessions`] — flooding it
    /// is exactly the scenario this method exists to make safe.
    pub fn poll_accept(&mut self) -> AcceptOutcome<L::Error> {
        match self.listener.poll_accept() {
            Ok(None) => AcceptOutcome::Idle,
            Ok(Some(session)) => {
                if self.sessions.len() >= self.listener.max_sessions() {
                    // Dropped right here: `session` is never fed, never
                    // polled, never stored.
                    drop(session);
                    AcceptOutcome::Refused
                } else {
                    let id = SessionId(self.next_id);
                    self.next_id += 1;
                    self.sessions.insert(
                        id,
                        IngestDriver::new(
                            session,
                            self.trunk_config,
                            self.handshake,
                            self.max_programs,
                        ),
                    );
                    AcceptOutcome::Admitted(id)
                }
            }
            Err(e) => AcceptOutcome::Error(e),
        }
    }

    /// Feed bytes read for session `id`. Returns `Some(health)` exactly when
    /// this call caused the session to terminate (`Ended`, `Failed`, or
    /// `HandshakeTimedOut`) — at which point it is removed from this driver
    /// (its slot is free for a future [`Self::poll_accept`]); returns `None`
    /// for an unknown `id` or a session still running (`Establishing` or
    /// `Live`) after this call.
    ///
    /// Pinned to `&[u8]` (unlike [`IngestDriver::feed`], generic over
    /// `Stage::In` since round 3): every [`Listener`] implementor
    /// today is a push accept over a byte-stream transport (RTMP/SRT
    /// listener), never a pull source (a pull source dials out via
    /// [`Dialer`], it does not accept — see `multimux::source::srt`'s "Why
    /// listener mode is not a `Listener` yet"), so there is no caller needing
    /// anything else here yet.
    pub fn feed(
        &mut self,
        id: SessionId,
        input: &[u8],
        now: Timestamp,
    ) -> Option<HealthState<<L::Session as Stage>::Error>>
    where
        L::Session: for<'a> Stage<In<'a> = &'a [u8]>,
    {
        self.drive(id, |d| d.feed(input, now))
    }

    /// Let session `id` act on the passage of time — see
    /// [`IngestDriver::on_deadline`]. Same removal-on-termination contract
    /// as [`Self::feed`].
    pub fn on_deadline(
        &mut self,
        id: SessionId,
        now: Timestamp,
    ) -> Option<HealthState<<L::Session as Stage>::Error>> {
        self.drive(id, |d| d.on_deadline(now))
    }

    /// Signal clean end-of-input for session `id` — see
    /// [`IngestDriver::finish`]. Same removal-on-termination contract as
    /// [`Self::feed`].
    pub fn finish(&mut self, id: SessionId) -> Option<HealthState<<L::Session as Stage>::Error>> {
        self.drive(id, IngestDriver::finish)
    }

    /// This session's current health, if it is still admitted (a terminated
    /// session is removed by the call that terminated it — see
    /// [`Self::feed`] — so query the return value of that call for the
    /// terminal state).
    pub fn health(&self, id: SessionId) -> Option<&HealthState<<L::Session as Stage>::Error>> {
        self.sessions.get(&id).map(IngestDriver::health)
    }

    /// The `Trunk` for `program` under session `id`, if announced yet.
    pub fn trunk(&self, id: SessionId, program: ProgramId) -> Option<&Arc<Trunk>> {
        self.sessions.get(&id).and_then(|d| d.trunk(program))
    }

    /// Read-only access to session `id`'s underlying [`IngestDriver`], if
    /// still admitted.
    ///
    /// # Why this (and [`Self::driver_mut`]/[`Self::reap_if_terminal`]) exist
    ///
    /// [`Self::feed`] is pinned to `&[u8]` because it bundles three steps —
    /// feed, then check `is_running()`, then remove if not — into one call,
    /// which only works when the caller has nothing it needs to observe
    /// *between* "the session just went terminal" and "the session is gone".
    /// RTMP (issue #805 task 4) is exactly a caller that does: its
    /// `Listener::Session` is fed already-parsed-and-replied-to
    /// `rtmp_runtime::server::ServerEvent`s (`Stage::In<'a> = &'a
    /// [ServerEvent]`, not `&'a [u8]` — see `multimux::source::rtmp`'s module
    /// doc for why `RtmpConnection` makes that the honest shape), so
    /// [`Self::feed`]'s bound does not apply; and every driver-backed `run_*`
    /// entry point (`crate::source::report_driver_progress`,
    /// `crate::source::segment::drive_program_segmenters` in the `multimux`
    /// crate) needs to publish this session's newly-announced programs and
    /// flush its segmenter *before* a just-terminated session is reaped,
    /// exactly like the single-`IngestDriver` drive loops
    /// (`multimux::source::rtsp::run_rtsp` et al.) already do against an
    /// `IngestDriver` they own outright.
    ///
    /// These three methods let a driving loop reassemble that same
    /// feed → observe → reap sequence for a session admitted by a
    /// [`ListenDriver`], for any `Stage::In` shape: `driver_mut(id)` to feed
    /// (via [`IngestDriver::feed`], generic since round 3) or finish, `driver(id)`
    /// to read back `programs()`/`trunk()`/`health()`/[`IngestDriver::session`]
    /// afterward, then `reap_if_terminal(id)` to perform the exact removal
    /// [`Self::feed`] would have, once the caller is done observing.
    pub fn driver(&self, id: SessionId) -> Option<&IngestDriver<L::Session>> {
        self.sessions.get(&id)
    }

    /// Mutable access to session `id`'s underlying [`IngestDriver`], if still
    /// admitted — see [`Self::driver`] for why this exists. Does **not**
    /// reap a session that becomes terminal as a result of a call made
    /// through this reference; call [`Self::reap_if_terminal`] afterward.
    pub fn driver_mut(&mut self, id: SessionId) -> Option<&mut IngestDriver<L::Session>> {
        self.sessions.get_mut(&id)
    }

    /// If session `id` is admitted and has reached a terminal
    /// [`HealthState`] (`is_running() == false`), removes it and returns that
    /// final state — exactly the same removal step [`Self::feed`] performs
    /// internally, exposed standalone for a caller using [`Self::driver_mut`]
    /// instead of
    /// [`Self::feed`]/[`Self::on_deadline`]/[`Self::finish`] (see
    /// [`Self::driver`]'s doc). `None` for an unknown `id` or one still
    /// running — a no-op either way, so calling this speculatively every
    /// iteration is always safe.
    pub fn reap_if_terminal(
        &mut self,
        id: SessionId,
    ) -> Option<HealthState<<L::Session as Stage>::Error>> {
        let driver = self.sessions.get(&id)?;
        if driver.health().is_running() {
            None
        } else {
            self.sessions.remove(&id).map(|d| d.health)
        }
    }

    /// Runs `op` against session `id`'s driver, then — if that call left it in
    /// a terminal state ([`HealthState::is_running`] `== false`) — removes it
    /// and returns that final state. This is the one place a session leaves
    /// `self.sessions`, which is what keeps this driver's resident memory
    /// bounded to [`Listener::max_sessions`] rather than accumulating every
    /// session that has ever ended, failed, or timed out mid-handshake.
    fn drive(
        &mut self,
        id: SessionId,
        op: impl FnOnce(&mut IngestDriver<L::Session>),
    ) -> Option<HealthState<<L::Session as Stage>::Error>> {
        let driver = self.sessions.get_mut(&id)?;
        op(driver);
        if driver.health().is_running() {
            None
        } else {
            self.sessions.remove(&id).map(|d| d.health)
        }
    }
}

/// Build a [`ListenDriver`] over `listener` — the whole of `run_listen`.
pub fn run_listen<L: Listener>(
    listener: L,
    trunk_config: TrunkConfig,
    handshake: HandshakePolicy,
    max_programs: NonZeroUsize,
) -> ListenDriver<L> {
    ListenDriver::new(listener, trunk_config, handshake, max_programs)
}

#[cfg(test)]
mod tests {
    use super::*;
    use broadcast_common::Demand;
    use std::collections::VecDeque;
    use transmux::pipeline::{CodecConfig, DataCarriage};

    /// `NonZeroUsize` from a literal capacity — see `trunk`'s identical test
    /// helper.
    fn nz(n: usize) -> std::num::NonZeroUsize {
        std::num::NonZeroUsize::new(n).expect("test capacity must be non-zero")
    }

    /// Minimal config every test's `Trunk`s share — capacities are irrelevant
    /// to these tests beyond "large enough that nothing evicts mid-test".
    fn trunk_config() -> TrunkConfig {
        TrunkConfig::new(nz(64), nz(16), nz(8), nz(8), nz(8))
    }

    /// A handshake deadline far enough out that it never fires — for the
    /// tests that are not about the handshake bound. The two that *are* about
    /// it set their own tight deadline explicitly.
    fn handshake() -> HandshakePolicy {
        HandshakePolicy::establish_by(Timestamp::from_nanos(u64::MAX))
    }

    /// An ambient `max_programs` bound for tests that are not about the
    /// program cap itself — large enough that no test relying on this
    /// helper ever hits it. The tests that *are* about the cap pass their
    /// own small `nz(N)` explicitly.
    fn max_programs() -> std::num::NonZeroUsize {
        nz(1024)
    }

    fn sample(byte: u8) -> Sample {
        Sample::new(Bytes::from(vec![byte; 4]), Some(0), Some(0), Some(1), true)
    }

    fn opaque_track(track_id: u32) -> TrackSpec {
        TrackSpec::new(
            track_id,
            90_000,
            CodecConfig::Data {
                stream_type: 0x06,
                descriptors: Vec::new(),
                carriage: DataCarriage::Pes,
            },
        )
    }

    /// A fake, `#[cfg(test)]`-only error type for scripted sessions/dialers —
    /// carries a reason string purely for assertion messages.
    #[derive(Debug, Clone, PartialEq, Eq)]
    struct FakeError(&'static str);

    /// What one `feed()` call on a [`ScriptedSession`] does.
    enum FeedOutcome {
        /// Succeed, queuing these events for `poll()` to hand back.
        Events(Vec<SessionEvent>),
        /// Fail outright with this error.
        Err(FakeError),
    }

    /// A fully scripted [`IngestSession`]: each `feed()` call consumes the
    /// next entry of `script`, either queuing its events or failing; `finish`
    /// hands back `finish_outcome` (defaults to a clean `Ok(())`).
    ///
    /// Starts with [`SessionEvent::Established`] already queued — modelling a
    /// source whose handshake is a purely local operation with nothing to
    /// negotiate (binding a UDP socket), which is a real case, not a shortcut.
    /// The genuinely multi-round-trip case has its own session type below
    /// ([`HandshakeSession`]).
    struct ScriptedSession {
        script: VecDeque<FeedOutcome>,
        pending: VecDeque<SessionEvent>,
        finish_outcome: Result<(), FakeError>,
    }

    impl ScriptedSession {
        fn new(script: Vec<FeedOutcome>) -> Self {
            ScriptedSession {
                script: script.into(),
                pending: VecDeque::from(vec![SessionEvent::Established]),
                finish_outcome: Ok(()),
            }
        }

        fn failing_finish(mut self, err: FakeError) -> Self {
            self.finish_outcome = Err(err);
            self
        }
    }

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

        fn feed(&mut self, _input: &[u8], _now: Timestamp) -> Result<(), FakeError> {
            match self.script.pop_front() {
                Some(FeedOutcome::Events(evs)) => {
                    self.pending.extend(evs);
                    Ok(())
                }
                Some(FeedOutcome::Err(e)) => Err(e),
                None => Ok(()),
            }
        }

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

        fn finish(&mut self) -> Result<(), FakeError> {
            self.finish_outcome.clone()
        }

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

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

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

    /// Nothing to send: takes `poll_transmit`'s default.
    impl IngestSession for ScriptedSession {
        type Request = Bytes;
    }

    /// A fake [`Dialer`] yielding one pre-built session then erroring on
    /// every call after (or always erroring, for the reconnect test).
    struct ScriptedDialer {
        sessions: VecDeque<ScriptedSession>,
        fail_with: FakeError,
    }

    impl Dialer for ScriptedDialer {
        type Session = ScriptedSession;
        type Error = FakeError;

        fn dial(&mut self) -> Result<ScriptedSession, FakeError> {
            self.sessions
                .pop_front()
                .ok_or_else(|| self.fail_with.clone())
        }
    }

    // --- run_dial: happy path, samples land in the Trunk ------------------

    #[test]
    fn run_dial_drives_fake_session_end_to_end_samples_land_in_trunk() {
        let session = ScriptedSession::new(vec![
            FeedOutcome::Events(vec![SessionEvent::NewProgram {
                program: ProgramId(1),
                tracks: vec![opaque_track(7)],
            }]),
            FeedOutcome::Events(vec![SessionEvent::Sample {
                program: ProgramId(1),
                track_id: 7,
                retention: RetentionClass::Timed,
                sample: sample(0xAB),
            }]),
        ]);
        let mut dialer = ScriptedDialer {
            sessions: VecDeque::from(vec![session]),
            fail_with: FakeError("unused"),
        };

        let mut driver = run_dial(&mut dialer, trunk_config(), handshake(), max_programs())
            .expect("fake dial succeeds");
        let trunk_before = driver.trunk(ProgramId(1)).cloned();
        assert!(
            trunk_before.is_none(),
            "no Trunk before NewProgram is announced"
        );

        driver.feed(b"pat", Timestamp::ZERO);
        let trunk = driver
            .trunk(ProgramId(1))
            .cloned()
            .expect("NewProgram announced a Trunk for program 1");
        let mut cursor = trunk.subscribe();

        driver.feed(b"pes", Timestamp::from_nanos(1));

        let item = cursor.poll().expect("the published sample is on the ring");
        match item {
            crate::SampleCursorItem::Timed { track_id, sample } => {
                assert_eq!(track_id, 7);
                assert_eq!(sample.data.as_ref(), &[0xAB; 4]);
            }
            other => panic!("expected Timed, got {other:?}"),
        }
    }

    // --- Track-set plumbing (issue #781): NewProgram seeds, TracksChanged --
    // --- replaces, unannounced TracksChanged drops -------------------------

    /// The discarded-track-list fix, made to fail against the pre-fix code:
    /// before `drain()`'s `NewProgram` arm called `writer.set_tracks(tracks)`,
    /// `tracks` was bound with `..` and never touched a `Trunk` at all, so
    /// every `Trunk::tracks()` stayed permanently empty.
    ///
    /// MUTATION VERIFIED: reverting `drain()`'s `NewProgram` arm to bind
    /// `SessionEvent::NewProgram { program, .. }` (dropping `tracks`, as it
    /// was before this change) and removing the `writer.set_tracks(tracks)`
    /// call makes the `assert_eq!` below fail — `track_ids` reads back `[]`
    /// instead of `[3, 9]`. Recompiled and re-run to confirm the failure,
    /// then reverted.
    #[test]
    fn new_program_seeds_the_trunk_with_exactly_the_announced_tracks() {
        let session =
            ScriptedSession::new(vec![FeedOutcome::Events(vec![SessionEvent::NewProgram {
                program: ProgramId(1),
                tracks: vec![opaque_track(3), opaque_track(9)],
            }])]);
        let mut dialer = ScriptedDialer {
            sessions: VecDeque::from(vec![session]),
            fail_with: FakeError("unused"),
        };
        let mut driver = run_dial(&mut dialer, trunk_config(), handshake(), max_programs())
            .expect("fake dial succeeds");

        driver.feed(b"pat", Timestamp::ZERO);

        let trunk = driver
            .trunk(ProgramId(1))
            .cloned()
            .expect("NewProgram announced a Trunk for program 1");
        let track_ids: Vec<u32> = trunk.tracks().iter().map(|t| t.track_id).collect();
        assert_eq!(
            track_ids,
            vec![3, 9],
            "the Trunk must expose exactly the tracks NewProgram carried"
        );
    }

    /// `TracksChanged` replaces the previously-seeded set wholesale and
    /// bumps `track_generation` — NewProgram's own seeding call counts as
    /// the first `set_tracks`, so generation reads `1` right after
    /// admission and `2` after the `TracksChanged`.
    ///
    /// MUTATION VERIFIED: changing `drain()`'s `TracksChanged` arm from
    /// `writer.set_tracks(tracks)` to `writer.publish_event(..)`-style no-op
    /// (concretely: commenting out the `writer.set_tracks(tracks);` call,
    /// leaving the event silently absorbed) makes both assertions below
    /// fail — `track_ids` still reads back `[1]` instead of `[1, 2]`, and
    /// `track_generation()` stays at `1` instead of advancing to `2`.
    /// Recompiled and re-run to confirm the failure, then reverted.
    #[test]
    fn tracks_changed_replaces_the_set_and_bumps_generation() {
        let session = ScriptedSession::new(vec![
            FeedOutcome::Events(vec![SessionEvent::NewProgram {
                program: ProgramId(1),
                tracks: vec![opaque_track(1)],
            }]),
            FeedOutcome::Events(vec![SessionEvent::TracksChanged {
                program: ProgramId(1),
                tracks: vec![opaque_track(1), opaque_track(2)],
            }]),
        ]);
        let mut dialer = ScriptedDialer {
            sessions: VecDeque::from(vec![session]),
            fail_with: FakeError("unused"),
        };
        let mut driver = run_dial(&mut dialer, trunk_config(), handshake(), max_programs())
            .expect("fake dial succeeds");

        driver.feed(b"pat", Timestamp::from_nanos(0));
        let trunk = driver.trunk(ProgramId(1)).cloned().unwrap();
        assert_eq!(
            trunk.track_generation(),
            1,
            "NewProgram's seed counts as the first set_tracks call"
        );

        driver.feed(b"pmt-bump", Timestamp::from_nanos(1));
        let track_ids: Vec<u32> = trunk.tracks().iter().map(|t| t.track_id).collect();
        assert_eq!(
            track_ids,
            vec![1, 2],
            "TracksChanged must replace the set with the new complete snapshot"
        );
        assert_eq!(
            trunk.track_generation(),
            2,
            "TracksChanged must bump the generation exactly once"
        );
    }

    /// A `TracksChanged` for a program never announced via `NewProgram` is a
    /// contract violation, handled exactly like an unannounced `Sample`:
    /// dropped, not panicking, and never minting a `Trunk` on its own.
    ///
    /// MUTATION VERIFIED: changing `drain()`'s `TracksChanged` arm from
    /// `if let Some(writer) = self.writers.get(&program) { .. }` to
    /// unconditionally minting a fresh `Trunk`/`writer` for `program`
    /// (mirroring what `NewProgram` does) makes the `assert!` below fail —
    /// `driver.trunk(ProgramId(1))` comes back `Some(..)` instead of `None`.
    /// Recompiled and re-run to confirm the failure, then reverted.
    #[test]
    fn tracks_changed_for_an_unannounced_program_is_dropped_not_panicking_and_mints_nothing() {
        let session = ScriptedSession::new(vec![FeedOutcome::Events(vec![
            SessionEvent::TracksChanged {
                program: ProgramId(1),
                tracks: vec![opaque_track(1)],
            },
        ])]);
        let mut driver = IngestDriver::new(session, trunk_config(), handshake(), max_programs());

        // Must not panic.
        driver.feed(b"stray", Timestamp::ZERO);

        assert!(
            driver.trunk(ProgramId(1)).is_none(),
            "TracksChanged must never mint a Trunk on its own"
        );
        assert_eq!(
            driver.program_count(),
            0,
            "an unannounced program must not be admitted by TracksChanged"
        );
    }

    /// `track_generation` is stable across everything that is *not* a
    /// `NewProgram` seed or a `TracksChanged` — ordinary samples and no-op
    /// feed calls must never bump it, so a consumer polling the generation
    /// as a cheap "did the track set change" check sees no false positives.
    ///
    /// MUTATION VERIFIED: adding a `writer.set_tracks(Vec::new())` call to
    /// `drain()`'s `Sample` arm (simulating "generation accidentally bumped
    /// by unrelated activity") makes the final `assert_eq!` below fail —
    /// `track_generation()` reads back `2` instead of `1` after the sample
    /// is published. Recompiled and re-run to confirm the failure, then
    /// reverted.
    #[test]
    fn track_generation_is_stable_when_nothing_changes() {
        let session = ScriptedSession::new(vec![
            FeedOutcome::Events(vec![SessionEvent::NewProgram {
                program: ProgramId(1),
                tracks: vec![opaque_track(1)],
            }]),
            FeedOutcome::Events(vec![SessionEvent::Sample {
                program: ProgramId(1),
                track_id: 1,
                retention: RetentionClass::Timed,
                sample: sample(0xAB),
            }]),
            FeedOutcome::Events(vec![]),
        ]);
        let mut driver = IngestDriver::new(session, trunk_config(), handshake(), max_programs());

        driver.feed(b"1", Timestamp::from_nanos(0));
        let trunk = driver.trunk(ProgramId(1)).cloned().unwrap();
        assert_eq!(trunk.track_generation(), 1);

        driver.feed(b"2", Timestamp::from_nanos(1));
        driver.feed(b"3", Timestamp::from_nanos(2));

        assert_eq!(
            trunk.track_generation(),
            1,
            "samples and no-op feeds must never bump track_generation"
        );
    }

    /// A REPEAT `NewProgram` for an already-admitted program must update the
    /// existing `Trunk` in place, never mint a replacement.
    ///
    /// This asserts continuity from the *subscriber's* side, which is the
    /// property that actually matters and the one a track-set assertion
    /// alone would miss: `Trunk` is a cloneable handle over shared state, so
    /// re-minting swaps a fresh empty `Trunk` into `programs` while every
    /// already-issued cursor keeps reading the orphaned one that no longer
    /// receives writes. The stream does not error — it silently stops, which
    /// is far harder to diagnose in production than a crash, and whatever
    /// the old `Trunk` still buffered (and so the DVR window) goes with it.
    ///
    /// MUTATION VERIFIED: removing the `if let Some(writer) =
    /// self.writers.get(&program) { .. continue }` early-return from
    /// `drain()`'s `NewProgram` arm (restoring the unconditional
    /// `Trunk::new`) fails this test on the track-set assertion first —
    /// `left: [7], right: [7, 8]`, the re-announcement's tracks having
    /// landed on a replacement `Trunk` the subscriber cannot see.
    ///
    /// Both assertions were confirmed to bite independently: re-running the
    /// mutation with the track-set assertion suppressed then fails on
    /// `cursor.poll()` returning `None` ("a cursor subscribed before the
    /// re-announcement must still receive samples"), which is the direct
    /// proof of subscriber stranding rather than an inference from the
    /// track set. Recompiled and re-run for each, then reverted.
    #[test]
    fn repeat_new_program_updates_in_place_and_does_not_strand_subscribers() {
        let session = ScriptedSession::new(vec![
            FeedOutcome::Events(vec![SessionEvent::NewProgram {
                program: ProgramId(1),
                tracks: vec![opaque_track(7)],
            }]),
            // The same program announced again, with a grown track set --
            // what a session that re-states its program on a PMT change
            // emits, rather than using `TracksChanged`.
            FeedOutcome::Events(vec![SessionEvent::NewProgram {
                program: ProgramId(1),
                tracks: vec![opaque_track(7), opaque_track(8)],
            }]),
            FeedOutcome::Events(vec![SessionEvent::Sample {
                program: ProgramId(1),
                track_id: 7,
                retention: RetentionClass::Timed,
                sample: sample(0xCD),
            }]),
        ]);
        let mut dialer = ScriptedDialer {
            sessions: VecDeque::from(vec![session]),
            fail_with: FakeError("unused"),
        };
        let mut driver = run_dial(&mut dialer, trunk_config(), handshake(), max_programs())
            .expect("fake dial succeeds");

        driver.feed(b"pat", Timestamp::ZERO);
        let trunk = driver
            .trunk(ProgramId(1))
            .cloned()
            .expect("NewProgram announced a Trunk for program 1");
        let mut cursor = trunk.subscribe();

        driver.feed(b"pat-again", Timestamp::from_nanos(1));

        // The re-announcement is an update, not a new program.
        assert_eq!(
            driver.program_count(),
            1,
            "a repeat announcement must not add a program"
        );
        let track_ids: Vec<u32> = trunk.tracks().iter().map(|t| t.track_id).collect();
        assert_eq!(
            track_ids,
            vec![7, 8],
            "the re-announcement's track set must land on the SAME Trunk the \
             subscriber already holds"
        );

        driver.feed(b"pes", Timestamp::from_nanos(2));

        let item = cursor
            .poll()
            .expect("a cursor subscribed before the re-announcement must still receive samples");
        match item {
            crate::SampleCursorItem::Timed { track_id, sample } => {
                assert_eq!(track_id, 7);
                assert_eq!(sample.data.as_ref(), &[0xCD; 4]);
            }
            other => panic!("expected Timed, got {other:?}"),
        }
    }

    // --- EOF vs failure: the test that makes HealthState::Failed real -----

    #[test]
    fn clean_finish_yields_ended_not_failed() {
        let session = ScriptedSession::new(vec![]);
        let mut driver = IngestDriver::new(session, trunk_config(), handshake(), max_programs());
        assert!(
            matches!(driver.health(), HealthState::Establishing),
            "a freshly dialled session has not established yet"
        );

        driver.finish();

        // MUTATION-CHECKED: flip this to `Failed` in the impl (or make
        // `finish()`'s `Ok` arm also set `Failed`) and this assertion is the
        // one that catches it.
        assert!(
            matches!(driver.health(), HealthState::Ended),
            "a session that finished cleanly must be Ended, not Failed: {:?}",
            driver.health()
        );
    }

    #[test]
    fn erroring_feed_yields_failed_not_ended() {
        let session = ScriptedSession::new(vec![FeedOutcome::Err(FakeError("bad continuity"))]);
        let mut driver = IngestDriver::new(session, trunk_config(), handshake(), max_programs());

        driver.feed(b"garbage", Timestamp::ZERO);

        match driver.health() {
            HealthState::Failed(FakeError(reason)) => assert_eq!(*reason, "bad continuity"),
            other => panic!("expected Failed(\"bad continuity\"), got {other:?}"),
        }
    }

    #[test]
    fn erroring_finish_yields_failed_not_ended() {
        let session = ScriptedSession::new(vec![]).failing_finish(FakeError("truncated tail"));
        let mut driver = IngestDriver::new(session, trunk_config(), handshake(), max_programs());

        driver.finish();

        match driver.health() {
            HealthState::Failed(FakeError(reason)) => assert_eq!(*reason, "truncated tail"),
            other => panic!("expected Failed(\"truncated tail\"), got {other:?}"),
        }
    }

    #[test]
    fn terminated_driver_ignores_further_feed_and_finish() {
        let session = ScriptedSession::new(vec![FeedOutcome::Err(FakeError("boom"))]);
        let mut driver = IngestDriver::new(session, trunk_config(), handshake(), max_programs());
        driver.feed(b"x", Timestamp::ZERO);
        assert!(matches!(driver.health(), HealthState::Failed(_)));

        // Once Failed, feed/finish must be no-ops: no panic, and health does
        // not flip back to Ended via a stray finish() call.
        driver.finish();
        assert!(matches!(driver.health(), HealthState::Failed(_)));
    }

    // --- Multi-program (B5): two programs -> two Trunks; late program -----

    #[test]
    fn one_connection_two_programs_yields_two_trunks_including_one_announced_late() {
        let session = ScriptedSession::new(vec![
            FeedOutcome::Events(vec![
                SessionEvent::NewProgram {
                    program: ProgramId(1),
                    tracks: vec![opaque_track(1)],
                },
                SessionEvent::Sample {
                    program: ProgramId(1),
                    track_id: 1,
                    retention: RetentionClass::Timed,
                    sample: sample(0x01),
                },
            ]),
            // Nothing new this round: proves NewProgram isn't required on
            // every feed call.
            FeedOutcome::Events(vec![]),
            // Program 2 appears only now, after program 1's samples were
            // already flowing — the exact "announced after ingest started"
            // case B5 requires.
            FeedOutcome::Events(vec![
                SessionEvent::NewProgram {
                    program: ProgramId(2),
                    tracks: vec![opaque_track(9)],
                },
                SessionEvent::Sample {
                    program: ProgramId(2),
                    track_id: 9,
                    retention: RetentionClass::Timed,
                    sample: sample(0x02),
                },
            ]),
        ]);
        let mut dialer = ScriptedDialer {
            sessions: VecDeque::from(vec![session]),
            fail_with: FakeError("unused"),
        };
        let mut driver =
            run_dial(&mut dialer, trunk_config(), handshake(), max_programs()).unwrap();

        driver.feed(b"1", Timestamp::from_nanos(0));
        assert!(driver.trunk(ProgramId(1)).is_some());
        assert!(
            driver.trunk(ProgramId(2)).is_none(),
            "program 2 must not exist before it is announced"
        );

        driver.feed(b"2", Timestamp::from_nanos(1));
        assert!(
            driver.trunk(ProgramId(2)).is_none(),
            "a no-op feed must not fabricate a program"
        );

        driver.feed(b"3", Timestamp::from_nanos(2));
        let trunk1 = driver.trunk(ProgramId(1)).cloned().unwrap();
        let trunk2 = driver
            .trunk(ProgramId(2))
            .cloned()
            .expect("program 2 announced mid-session must get its own Trunk");
        assert!(
            !Arc::ptr_eq(&trunk1, &trunk2),
            "each program must get a genuinely distinct Trunk"
        );

        let mut programs: Vec<_> = driver.programs().collect();
        programs.sort();
        assert_eq!(programs, vec![ProgramId(1), ProgramId(2)]);

        // Both Trunks actually carry their own program's sample, subscribed
        // fresh now (after the fact) — proving the two rings are genuinely
        // independent, not aliases of the same one.
        assert_eq!(trunk1.timed_len(), 1);
        assert_eq!(trunk2.timed_len(), 1);
    }

    // --- max_programs: the fifth unbounded-allocation vector, bounded ------

    #[test]
    fn programs_up_to_max_get_a_trunk_each_the_next_one_is_refused_and_reported() {
        let cap = 2;
        let session = ScriptedSession::new(vec![FeedOutcome::Events(vec![
            SessionEvent::NewProgram {
                program: ProgramId(1),
                tracks: vec![opaque_track(1)],
            },
            SessionEvent::NewProgram {
                program: ProgramId(2),
                tracks: vec![opaque_track(2)],
            },
            // The (cap+1)th distinct program in the same drain() call.
            SessionEvent::NewProgram {
                program: ProgramId(3),
                tracks: vec![opaque_track(3)],
            },
        ])]);
        let mut driver = IngestDriver::new(session, trunk_config(), handshake(), nz(cap));

        driver.feed(b"pat", Timestamp::ZERO);

        assert!(driver.trunk(ProgramId(1)).is_some(), "program 1 admitted");
        assert!(driver.trunk(ProgramId(2)).is_some(), "program 2 admitted");
        assert!(
            driver.trunk(ProgramId(3)).is_none(),
            "the (cap+1)th program must be refused a Trunk"
        );
        assert_eq!(
            driver.program_count(),
            cap,
            "admitted program count must sit exactly at the cap, not above it"
        );
        // MUTATION-CHECKED: dropping the `refused_programs += 1` (or the
        // whole cap check) in `drain()`'s `NewProgram` arm makes this fail —
        // reported, not a silent drop.
        assert_eq!(
            driver.refused_program_count(),
            1,
            "the refusal must be reported via a queryable counter, never silent"
        );
    }

    #[test]
    fn repeat_announcement_of_an_already_admitted_program_is_never_refused() {
        // A program re-announcing itself (e.g. a PMT version bump reiterating
        // the same program_number) must not be treated as a new admission and
        // so must never count against the cap or be refused.
        let cap = 1;
        let session = ScriptedSession::new(vec![FeedOutcome::Events(vec![
            SessionEvent::NewProgram {
                program: ProgramId(1),
                tracks: vec![opaque_track(1)],
            },
            SessionEvent::NewProgram {
                program: ProgramId(1),
                tracks: vec![opaque_track(1)],
            },
        ])]);
        let mut driver = IngestDriver::new(session, trunk_config(), handshake(), nz(cap));

        driver.feed(b"pat", Timestamp::ZERO);

        assert_eq!(driver.program_count(), 1);
        assert_eq!(
            driver.refused_program_count(),
            0,
            "re-announcing an already-admitted program must not be refused"
        );
    }

    #[test]
    fn refusal_does_not_disturb_already_admitted_programs_their_samples_keep_flowing() {
        let cap = 1;
        let session = ScriptedSession::new(vec![
            FeedOutcome::Events(vec![SessionEvent::NewProgram {
                program: ProgramId(1),
                tracks: vec![opaque_track(1)],
            }]),
            // In the SAME drain() call: program 2 is refused, and a sample
            // for the already-admitted program 1 is published — proving the
            // refusal of one program does not interrupt delivery for another
            // already flowing, which is the whole justification for
            // "refuse the extra program" over "fail the session".
            FeedOutcome::Events(vec![
                SessionEvent::NewProgram {
                    program: ProgramId(2),
                    tracks: vec![opaque_track(2)],
                },
                SessionEvent::Sample {
                    program: ProgramId(1),
                    track_id: 1,
                    retention: RetentionClass::Timed,
                    sample: sample(0x01),
                },
            ]),
        ]);
        let mut driver = IngestDriver::new(session, trunk_config(), handshake(), nz(cap));

        driver.feed(b"1", Timestamp::from_nanos(0));
        let trunk1 = driver.trunk(ProgramId(1)).cloned().unwrap();
        let mut cursor = trunk1.subscribe();

        driver.feed(b"2", Timestamp::from_nanos(1));

        assert!(
            driver.trunk(ProgramId(2)).is_none(),
            "program 2 must be refused, not given a Trunk"
        );
        assert_eq!(driver.refused_program_count(), 1);
        // MUTATION-CHECKED: if refusing program 2 were implemented by
        // failing the whole session (e.g. setting `self.health =
        // HealthState::Failed(..)`) instead of just skipping the one
        // `NewProgram`, this poll would come back empty because `feed`
        // would have stopped draining before the Sample event — this is
        // the assertion that would catch that.
        match cursor
            .poll()
            .expect("program 1's sample must still land despite program 2 being refused")
        {
            crate::SampleCursorItem::Timed { track_id, sample } => {
                assert_eq!(track_id, 1);
                assert_eq!(sample.data.as_ref(), &[0x01; 4]);
            }
            other => panic!("expected Timed, got {other:?}"),
        }
        assert!(
            matches!(driver.health(), HealthState::Live),
            "refusing an extra program must not fail the session: {:?}",
            driver.health()
        );
    }

    #[test]
    fn newprogram_flood_is_bounded_admits_exactly_max_programs_and_allocates_no_more_trunks() {
        let cap = 3;
        let mut events = Vec::with_capacity(10_000);
        for i in 0..10_000u32 {
            events.push(SessionEvent::NewProgram {
                program: ProgramId(i),
                tracks: vec![opaque_track(i)],
            });
        }
        let session = ScriptedSession::new(vec![FeedOutcome::Events(events)]);
        let mut driver = IngestDriver::new(session, trunk_config(), handshake(), nz(cap));

        driver.feed(b"flood", Timestamp::ZERO);

        // The count, not merely the outcome: exactly `cap` Trunks exist no
        // matter how many thousands of distinct ProgramIds were announced —
        // this is the assertion a `Vec<ProgramId>`-of-refusals regression
        // (unbounded in the same way the bug itself was) would still pass,
        // which is why it is checked here rather than only "some program
        // beyond the cap has no Trunk".
        assert_eq!(
            driver.program_count(),
            cap,
            "a 10,000-program flood must admit exactly max_programs Trunks, never more"
        );
        assert_eq!(
            driver.refused_program_count(),
            10_000 - cap as u64,
            "every program past the cap must be counted as refused"
        );
    }

    // --- run_listen: max_sessions is a hard bound --------------------------

    /// A [`Listener`] that always has a fresh session ready to accept —
    /// models an unbounded flood of inbound connections.
    struct FloodingListener {
        max_sessions: usize,
    }

    impl Listener for FloodingListener {
        type Session = ScriptedSession;
        type Error = FakeError;

        fn max_sessions(&self) -> usize {
            self.max_sessions
        }

        fn poll_accept(&mut self) -> Result<Option<ScriptedSession>, FakeError> {
            Ok(Some(ScriptedSession::new(vec![])))
        }
    }

    #[test]
    fn run_listen_admits_up_to_max_sessions_then_refuses_and_stays_bounded() {
        let max_sessions = 3;
        let mut driver = run_listen(
            FloodingListener { max_sessions },
            trunk_config(),
            handshake(),
            max_programs(),
        );

        for _ in 0..max_sessions {
            assert!(matches!(driver.poll_accept(), AcceptOutcome::Admitted(_)));
        }
        assert_eq!(driver.session_count(), max_sessions);

        // Flood far beyond the bound: every one of these must be refused,
        // and — the memory-growth assertion — session_count must never
        // exceed max_sessions, checked on every single iteration, not just
        // at the end.
        for _ in 0..10_000 {
            assert!(matches!(driver.poll_accept(), AcceptOutcome::Refused));
            assert!(
                driver.session_count() <= max_sessions,
                "session_count grew past max_sessions under flood"
            );
        }
        assert_eq!(driver.session_count(), max_sessions);
    }

    #[test]
    fn ended_session_is_reaped_freeing_a_slot() {
        let mut driver = run_listen(
            FloodingListener { max_sessions: 1 },
            trunk_config(),
            handshake(),
            max_programs(),
        );
        let AcceptOutcome::Admitted(id) = driver.poll_accept() else {
            panic!("expected admission");
        };
        assert_eq!(driver.session_count(), 1);
        assert!(matches!(driver.poll_accept(), AcceptOutcome::Refused));

        let health = driver.finish(id).expect("finish terminates the session");
        assert!(matches!(health, HealthState::Ended));
        assert_eq!(
            driver.session_count(),
            0,
            "a terminated session must be reaped, freeing its slot"
        );
        assert!(
            driver.health(id).is_none(),
            "a reaped session is no longer queryable by id"
        );

        // The freed slot admits a new connection.
        assert!(matches!(driver.poll_accept(), AcceptOutcome::Admitted(_)));
    }

    /// `driver`/`driver_mut`/`reap_if_terminal` (issue #805 task 4) must let a
    /// caller reassemble exactly what `Self::feed`/`Self::finish` do in one
    /// call, but with the observe step exposed *between* the state change and
    /// the reap — the whole reason these exist (see `Self::driver`'s doc: a
    /// `Listener::Session` whose `Stage::In` isn't `&[u8]`, e.g.
    /// `multimux::source::rtmp`'s `RtmpIngestSession`, cannot use
    /// `Self::feed` at all).
    ///
    /// MUTATION-CHECKED: dropping the `driver_mut(id).finish()` call (so the
    /// session is never actually finished) would leave `driver(id)`'s health
    /// at `Live`, failing the `Ended` assertion below; dropping the
    /// `reap_if_terminal` call would leave `session_count()` at 1 and
    /// `driver(id)` still `Some(_)`, failing the assertions after it.
    #[test]
    fn driver_mut_and_reap_if_terminal_mirror_feed_semantics() {
        let mut driver = run_listen(
            FloodingListener { max_sessions: 1 },
            trunk_config(),
            handshake(),
            max_programs(),
        );
        let AcceptOutcome::Admitted(id) = driver.poll_accept() else {
            panic!("expected admission");
        };
        assert!(
            matches!(
                driver.driver(id).map(IngestDriver::health),
                Some(HealthState::Establishing)
            ),
            "a freshly admitted session must be Establishing, observable via driver()"
        );

        // `driver_mut` feeds through the exact same `IngestDriver::feed`
        // `Self::feed` calls internally, but does NOT reap on termination.
        driver
            .driver_mut(id)
            .expect("session just admitted")
            .feed(b"reply", Timestamp::from_nanos(1));
        assert!(
            matches!(
                driver.driver(id).map(IngestDriver::health),
                Some(HealthState::Live)
            ),
            "driver_mut's feed must reach the session exactly like Self::feed would"
        );
        assert_eq!(driver.session_count(), 1, "not reaped: still Live");

        // Finishing must be observable via `driver()` BEFORE `reap_if_terminal`
        // removes it -- the exact ordering a driving loop that must
        // publish/flush before reaping depends on.
        driver.driver_mut(id).expect("still admitted").finish();
        assert!(
            matches!(
                driver.driver(id).map(IngestDriver::health),
                Some(HealthState::Ended)
            ),
            "finish() through driver_mut must be observable via driver() before reaping"
        );
        assert_eq!(driver.session_count(), 1, "not yet reaped");

        let health = driver
            .reap_if_terminal(id)
            .expect("a terminal session must be reaped");
        assert!(matches!(health, HealthState::Ended));
        assert_eq!(
            driver.session_count(),
            0,
            "reap_if_terminal must free the slot"
        );
        assert!(driver.driver(id).is_none());

        // A second call on an already-reaped id is a no-op, not a panic.
        assert!(driver.reap_if_terminal(id).is_none());
    }

    // --- Reconnect: bounded, caller-configurable, never spins --------------

    #[test]
    fn permanently_failing_dial_is_bounded_and_does_not_spin_or_grow() {
        let dialer = ScriptedDialer {
            sessions: VecDeque::new(),
            fail_with: FakeError("connection refused"),
        };
        let mut supervisor = DialSupervisor::new(dialer, ReconnectPolicy::new(3));

        assert!(matches!(
            supervisor.try_dial(trunk_config(), handshake(), max_programs()),
            DialAttempt::Retry(_)
        ));
        assert_eq!(supervisor.attempts(), 1);
        assert!(matches!(
            supervisor.try_dial(trunk_config(), handshake(), max_programs()),
            DialAttempt::Retry(_)
        ));
        assert_eq!(supervisor.attempts(), 2);
        assert!(matches!(
            supervisor.try_dial(trunk_config(), handshake(), max_programs()),
            DialAttempt::GaveUp(_)
        ));
        assert_eq!(supervisor.attempts(), 3);
        assert!(supervisor.is_exhausted());

        // Flood: however many more times this is called, it must never dial
        // again (no growth in `attempts`) and must always report Exhausted,
        // not spin back into Retry/GaveUp.
        for _ in 0..10_000 {
            assert!(matches!(
                supervisor.try_dial(trunk_config(), handshake(), max_programs()),
                DialAttempt::Exhausted
            ));
            assert_eq!(
                supervisor.attempts(),
                3,
                "attempts must not grow past max_attempts under flood"
            );
        }
    }

    #[test]
    fn dial_supervisor_succeeds_within_the_bound_and_resets_attempts() {
        let good_session = ScriptedSession::new(vec![]);
        let dialer = ScriptedDialer {
            sessions: VecDeque::from(vec![good_session]),
            fail_with: FakeError("refused"),
        };
        let mut supervisor = DialSupervisor::new(dialer, ReconnectPolicy::new(2));

        // First attempt succeeds immediately (the scripted dialer's one
        // queued session comes out on the very first `dial()` call).
        match supervisor.try_dial(trunk_config(), handshake(), max_programs()) {
            DialAttempt::Connected(_) => {}
            DialAttempt::Retry(_) => panic!("expected Connected, got Retry"),
            DialAttempt::GaveUp(_) => panic!("expected Connected, got GaveUp"),
            DialAttempt::Exhausted => panic!("expected Connected, got Exhausted"),
        }
        assert_eq!(supervisor.attempts(), 0);
        assert!(!supervisor.is_exhausted());
    }

    // --- Establishment: a real multi-round-trip handshake, no I/O ----------

    /// A genuinely multi-round-trip [`IngestSession`], shaped exactly like
    /// `rtsp_runtime::client::ClientSession`'s DESCRIBE → SETUP → PLAY
    /// sequence: it emits one request at a time via `poll_transmit`, consumes
    /// the peer's reply via `feed`, and only announces
    /// [`SessionEvent::Established`] after the third exchange completes.
    ///
    /// **It performs no I/O of any kind** — it has no socket, no runtime, and
    /// no `async fn`; it only moves bytes between its own two queues. The test
    /// below owns the "wire" itself, which is what makes the no-I/O claim an
    /// observable property rather than a promise.
    struct HandshakeSession {
        /// How many peer replies have been consumed so far.
        step: usize,
        outbound: VecDeque<Bytes>,
        pending: VecDeque<SessionEvent>,
    }

    /// The three requests `HandshakeSession` sends, in order — named after
    /// the RTSP sequence they stand in for.
    const HANDSHAKE_REQUESTS: [&[u8]; 3] = [b"DESCRIBE", b"SETUP", b"PLAY"];

    impl HandshakeSession {
        /// Queues the *first* request only. Note this is all `dial()` does —
        /// no connection, no negotiation.
        fn new() -> Self {
            HandshakeSession {
                step: 0,
                outbound: VecDeque::from(vec![Bytes::from_static(HANDSHAKE_REQUESTS[0])]),
                pending: VecDeque::new(),
            }
        }
    }

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

        fn feed(&mut self, input: &[u8], _now: Timestamp) -> Result<(), FakeError> {
            if self.step >= HANDSHAKE_REQUESTS.len() {
                // Post-handshake media.
                self.pending.push_back(SessionEvent::Sample {
                    program: ProgramId(1),
                    track_id: 7,
                    retention: RetentionClass::Timed,
                    sample: sample(0xAB),
                });
                return Ok(());
            }
            // Each reply must be the 200 for the request we actually sent —
            // a real state machine correlates, so this fake does too.
            let expected = format!(
                "200 {}",
                String::from_utf8_lossy(HANDSHAKE_REQUESTS[self.step])
            );
            if input != expected.as_bytes() {
                return Err(FakeError("handshake reply out of sequence"));
            }
            self.step += 1;
            match HANDSHAKE_REQUESTS.get(self.step) {
                // More handshake to do: queue the next request.
                Some(next) => self.outbound.push_back(Bytes::from_static(next)),
                // Final reply consumed: now established, and the track set is
                // known (it came from the DESCRIBE-equivalent).
                None => {
                    self.pending.push_back(SessionEvent::Established);
                    self.pending.push_back(SessionEvent::NewProgram {
                        program: ProgramId(1),
                        tracks: vec![opaque_track(7)],
                    });
                }
            }
            Ok(())
        }

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

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

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

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

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

    /// The handshake's outbound side: this is the *only* way a request leaves
    /// the session — it has no socket to write to.
    impl IngestSession for HandshakeSession {
        type Request = Bytes;

        fn poll_transmit(&mut self) -> Option<Bytes> {
            self.outbound.pop_front()
        }
    }

    /// A [`Dialer`] over [`HandshakeSession`] — `dial()` constructs and
    /// returns immediately, connecting nothing.
    struct HandshakeDialer;

    impl Dialer for HandshakeDialer {
        type Session = HandshakeSession;
        type Error = FakeError;

        fn dial(&mut self) -> Result<HandshakeSession, FakeError> {
            Ok(HandshakeSession::new())
        }
    }

    #[test]
    fn multi_round_trip_handshake_completes_through_feed_and_poll_transmit_only() {
        let mut dialer = HandshakeDialer;
        let mut driver = run_dial(&mut dialer, trunk_config(), handshake(), max_programs())
            .expect("dial constructs a session");

        // `dial()` did no I/O and did not establish anything.
        assert!(
            matches!(driver.health(), HealthState::Establishing),
            "dial() must not establish the session: {:?}",
            driver.health()
        );

        // The whole "network" is this Vec — every byte the session sends is
        // recorded here by the test, and every reply is handed back through
        // `feed`. Nothing else can move bytes, so a session that tried to do
        // its own I/O simply would not progress.
        let mut wire: Vec<Bytes> = Vec::new();
        let mut now = 0u64;

        // Pump: drain poll_transmit, answer each request, feed the reply.
        // Three round trips, driven entirely by this loop.
        for _ in 0..HANDSHAKE_REQUESTS.len() {
            let req = driver
                .poll_transmit()
                .expect("the session has a handshake request to send");
            assert!(
                driver.poll_transmit().is_none(),
                "one request in flight at a time"
            );
            wire.push(req.clone());

            let reply = format!("200 {}", String::from_utf8_lossy(&req));
            now += 1;
            driver.feed(reply.as_bytes(), Timestamp::from_nanos(now));
        }

        // The exact request sequence went out, in order, through
        // poll_transmit — nowhere else.
        let sent: Vec<&[u8]> = wire.iter().map(|b| b.as_ref()).collect();
        assert_eq!(sent, HANDSHAKE_REQUESTS, "handshake request sequence");

        // MUTATION-CHECKED: the promotion out of Establishing lives in
        // `drain()`'s `Established` arm.
        assert!(
            matches!(driver.health(), HealthState::Live),
            "after the final handshake reply the session must be Live: {:?}",
            driver.health()
        );

        // And it is genuinely usable: the program announced with Established
        // has a Trunk, and post-handshake media lands in it.
        let trunk = driver
            .trunk(ProgramId(1))
            .cloned()
            .expect("the handshake announced program 1");
        let mut cursor = trunk.subscribe();
        driver.feed(b"media", Timestamp::from_nanos(now + 1));
        match cursor.poll().expect("post-handshake sample on the ring") {
            crate::SampleCursorItem::Timed { track_id, .. } => assert_eq!(track_id, 7),
            other => panic!("expected Timed, got {other:?}"),
        }
    }

    /// A session that sends its first request and then never establishes,
    /// whatever it is fed — the stalled/half-open peer.
    struct StallingSession {
        outbound: VecDeque<Bytes>,
    }

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

        fn feed(&mut self, _input: &[u8], _now: Timestamp) -> Result<(), FakeError> {
            Ok(()) // never errors, never establishes — just silence
        }

        fn poll(&mut self) -> Option<SessionEvent> {
            None
        }

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

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

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

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

    impl IngestSession for StallingSession {
        type Request = Bytes;

        fn poll_transmit(&mut self) -> Option<Bytes> {
            self.outbound.pop_front()
        }
    }

    struct StallingListener {
        max_sessions: usize,
    }

    impl Listener for StallingListener {
        type Session = StallingSession;
        type Error = FakeError;

        fn max_sessions(&self) -> usize {
            self.max_sessions
        }

        fn poll_accept(&mut self) -> Result<Option<StallingSession>, FakeError> {
            // Queues its first request, exactly like a real session would —
            // so this models "we sent our opening request and the peer went
            // silent", not "nothing ever happened".
            Ok(Some(StallingSession {
                outbound: VecDeque::from(vec![Bytes::from_static(HANDSHAKE_REQUESTS[0])]),
            }))
        }
    }

    #[test]
    fn never_completing_handshake_is_bounded_and_reported_not_leaked() {
        const DEADLINE: Timestamp = Timestamp::from_nanos(1_000);
        let mut driver = run_listen(
            StallingListener { max_sessions: 1 },
            trunk_config(),
            HandshakePolicy::establish_by(DEADLINE),
            max_programs(),
        );

        let AcceptOutcome::Admitted(id) = driver.poll_accept() else {
            panic!("expected admission");
        };
        assert!(matches!(driver.health(id), Some(HealthState::Establishing)));
        // The one slot is taken, so nothing else gets in while this peer
        // stalls — which is exactly why the bound below must exist.
        assert!(matches!(driver.poll_accept(), AcceptOutcome::Refused));

        // Before the deadline, feeding it more silence must NOT terminate it:
        // a slow-but-progressing handshake is legitimate.
        assert!(
            driver
                .feed(id, b"...", Timestamp::from_nanos(DEADLINE.as_nanos() - 1))
                .is_none(),
            "must not time out before the deadline"
        );
        assert!(matches!(driver.health(id), Some(HealthState::Establishing)));
        assert_eq!(driver.session_count(), 1);

        // At the deadline, with the handshake still incomplete, it terminates
        // — reported, with the deadline that was blown.
        // MUTATION-CHECKED: `check_handshake_deadline`.
        let health = driver
            .on_deadline(id, DEADLINE)
            .expect("the blown deadline must terminate the session");
        assert_eq!(
            health,
            HealthState::HandshakeTimedOut { deadline: DEADLINE },
            "a never-completing handshake must be reported as HandshakeTimedOut"
        );

        // And it is REAPED, not leaked: the slot is free again, so a flood of
        // half-open connections cannot squat max_sessions forever.
        assert_eq!(
            driver.session_count(),
            0,
            "a timed-out session must be reaped, not left pinning its slot"
        );
        assert!(driver.health(id).is_none());
        assert!(matches!(driver.poll_accept(), AcceptOutcome::Admitted(_)));
    }

    #[test]
    fn handshake_completing_exactly_at_the_deadline_still_establishes() {
        const DEADLINE: Timestamp = Timestamp::from_nanos(500);
        // A locally-established session (Established already queued), fed at
        // exactly the deadline: the deadline check runs *after* the feed is
        // drained, so this must be Live, not HandshakeTimedOut.
        let session = ScriptedSession::new(vec![]);
        let mut driver = IngestDriver::new(
            session,
            trunk_config(),
            HandshakePolicy::establish_by(DEADLINE),
            max_programs(),
        );
        driver.feed(b"reply", DEADLINE);
        assert!(
            matches!(driver.health(), HealthState::Live),
            "a handshake completing exactly at the deadline must establish, \
             not be rejected by a nanosecond: {:?}",
            driver.health()
        );
    }

    #[test]
    fn next_deadline_surfaces_the_handshake_bound_while_establishing() {
        const DEADLINE: Timestamp = Timestamp::from_nanos(9_000);
        let mut dialer = HandshakeDialer;
        let mut driver = run_dial(
            &mut dialer,
            trunk_config(),
            HandshakePolicy::establish_by(DEADLINE),
            max_programs(),
        )
        .unwrap();

        // The session itself has no deadline of its own, so a caller driving
        // purely off next_deadline() would never fire the timeout check
        // unless the driver contributes the handshake bound here.
        assert_eq!(
            driver.next_deadline(),
            Some(DEADLINE),
            "while Establishing, next_deadline must surface the handshake bound"
        );

        // Once established it drops out again (the session's own None wins).
        for _ in 0..HANDSHAKE_REQUESTS.len() {
            let req = driver.poll_transmit().expect("handshake request");
            let reply = format!("200 {}", String::from_utf8_lossy(&req));
            driver.feed(reply.as_bytes(), Timestamp::ZERO);
        }
        assert!(matches!(driver.health(), HealthState::Live));
        assert_eq!(
            driver.next_deadline(),
            None,
            "the handshake bound must not linger after establishment"
        );
    }
}