g2g_core/

fanout.rs

//! Fan-out primitives for the dynamic graph layer (DESIGN.md §4.8.4).
//!
//! M9 (1→N slice): a multi-output sink abstraction plus the two routing
//! primitives that cover branch enable/disable and A/B switching:
//!
//! - [`Gate`] — 1→1. Forwards or drops each `DataFrame` by an atomic flag.
//!   It is a plain [`AsyncElement`], so it drops into the existing
//!   `run_source_transform_sink` runner unchanged.
//! - [`Router`] — 1→N. Sends each `DataFrame` to exactly one output port
//!   chosen by an atomic discriminator, and broadcasts `CapsChanged` to
//!   every port. It implements [`MultiOutputElement`], driven by the
//!   `run_source_fanout` runner.
//!
//! Both expose a cloneable control handle ([`GateHandle`], [`RouterHandle`]),
//! mirroring `SwapHandle` (`slot.rs`), so application code or another task
//! flips routing mid-stream without stalling the pipeline.
//!
//! The Merger (fan-in) and `BranchSlot` are a later slice. EOS broadcast on
//! the `Router` is the runner's responsibility, matching the existing
//! "runner forwards Eos" transform contract.

use alloc::boxed::Box;
use alloc::sync::Arc;
use alloc::vec::Vec;
use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering};

use crate::caps::Caps;
use crate::element::{
    AsyncElement, BoxFuture, ConfigureOutcome, ElementBound, OutputSink, PushOutcome, Reconfigure,
};
use crate::error::G2gError;
use crate::format_element::CapsConstraint;
use crate::frame::PipelinePacket;
use crate::memory::DomainSet;
use crate::property::{ElementMetadata, PropError, PropValue, PropertySpec};
use crate::runtime::SenderSink;
use crate::runtime::{PadKind, PadRequest};
use portable_atomic::AtomicU64;

/// Sentinel for "no bitrate pending" in a [`ReverseChannel`] (a real target is
/// a `u32`, so `u64::MAX` can never collide).
const NO_BITRATE: u64 = u64::MAX;

/// A per-input reverse-signal handle a fan-in session shares with the runner, so
/// a signal that originates on one track (a WebRTC PLI or BWE estimate arriving
/// for a given m-line) reaches the *matching* upstream source rather than being
/// broadcast or lost. It is the multi-input analog of the single-track
/// [`AsyncElement::take_reconfigure`](crate::AsyncElement::take_reconfigure) /
/// [`take_bitrate`](crate::AsyncElement::take_bitrate) the linear runner polls.
///
/// Cloneable and `Send` (`Arc`-backed atomics): the session task (which owns the
/// `Rtc` / network handle) writes with [`request_keyframe`](Self::request_keyframe)
/// / [`set_bitrate`](Self::set_bitrate); the runner's per-source adapter reads
/// the highest-priority pending signal with [`take`](Self::take) after each push
/// and surfaces it to that source as a [`PushOutcome`], exactly as a linked sink
/// would. Reconfigure outranks bitrate, matching the linear reverse channel.
#[derive(Debug, Clone)]
pub struct ReverseChannel {
    keyframe: Arc<AtomicBool>,
    bitrate: Arc<AtomicU64>,
}

impl Default for ReverseChannel {
    fn default() -> Self {
        Self {
            keyframe: Arc::new(AtomicBool::new(false)),
            bitrate: Arc::new(AtomicU64::new(NO_BITRATE)),
        }
    }
}

impl ReverseChannel {
    /// A fresh channel with nothing pending.
    pub fn new() -> Self {
        Self::default()
    }

    /// Session side: request an upstream keyframe (a remote PLI for this track).
    pub fn request_keyframe(&self) {
        self.keyframe.store(true, Ordering::Relaxed);
    }

    /// Session side: relay a target send bitrate (bits/second) for this track.
    pub fn set_bitrate(&self, bps: u32) {
        self.bitrate.store(bps as u64, Ordering::Relaxed);
    }

    /// Runner side: consume the highest-priority pending signal as a
    /// [`PushOutcome`], or `None` if nothing is pending. Reconfigure (keyframe)
    /// takes priority over a bitrate update, as on the linear reverse channel.
    pub fn take(&self) -> Option<PushOutcome> {
        if self.keyframe.swap(false, Ordering::Relaxed) {
            return Some(PushOutcome::Reconfigure(Reconfigure::ForceKeyframe));
        }
        match self.bitrate.swap(NO_BITRATE, Ordering::Relaxed) {
            NO_BITRATE => None,
            bps => Some(PushOutcome::Bitrate(bps as u32)),
        }
    }
}

/// Downstream output addressing one of N ports. The fan-out analog of
/// [`OutputSink`]: `push_to` selects the destination port. Dyn-safe via the
/// poll form (no heap), mirroring [`OutputSink::poll_push`]; `push_to` wraps
/// it in the concrete [`PushToFuture`] so `&mut dyn MultiOutputSink` callers
/// await it unchanged.
pub trait MultiOutputSink {
    /// Drive one packet toward `port`. Same slot contract as
    /// [`OutputSink::poll_push`].
    fn poll_push_to(
        &mut self,
        cx: &mut core::task::Context<'_>,
        port: usize,
        packet: &mut Option<PipelinePacket>,
    ) -> core::task::Poll<Result<PushOutcome, G2gError>>;

    /// Discard any phase a cancelled earlier push to `port` left behind.
    fn begin_push_to(&mut self, _port: usize) {}

    fn port_count(&self) -> usize;

    /// Add an output port carrying `caps` and return its index, or `None` if this
    /// sink cannot grow (M1014). A duplex session calls it when the peer adds a
    /// track and none of its declared pads is free; the dynamic duplex runner
    /// answers by building that port's link and finding it a sink. Never blocks,
    /// so it is callable from a session's own poll loop: a sink that cannot take
    /// the port right now answers `None` and the caller keeps whatever it does for
    /// a track it cannot place.
    ///
    /// Default `None`: the fixed-arity multi-sinks the static runners build refuse
    /// to grow, so a session written against this sees exactly today's behavior.
    fn add_port(&mut self, _caps: &Caps) -> Option<usize> {
        None
    }
}

/// `push_to` for concrete (sized) sinks; split out for the same
/// dyn-vs-provided-method ambiguity [`OutputSinkExt`] resolves.
///
/// [`OutputSinkExt`]: crate::element::OutputSinkExt
pub trait MultiOutputSinkExt: MultiOutputSink + Sized {
    fn push_to(&mut self, port: usize, packet: PipelinePacket) -> PushToFuture<'_, Self> {
        self.begin_push_to(port);
        PushToFuture {
            sink: self,
            port,
            packet: Some(packet),
        }
    }
}

impl<S: MultiOutputSink> MultiOutputSinkExt for S {}

impl<'e> dyn MultiOutputSink + 'e {
    /// [`MultiOutputSink::push_to`] for trait objects (the provided method
    /// needs `Self: Sized`).
    pub fn push_to(
        &mut self,
        port: usize,
        packet: PipelinePacket,
    ) -> PushToFuture<'_, dyn MultiOutputSink + 'e> {
        self.begin_push_to(port);
        PushToFuture {
            sink: self,
            port,
            packet: Some(packet),
        }
    }
}

/// Concrete future behind [`MultiOutputSink::push_to`]: the packet slot
/// [`MultiOutputSink::poll_push_to`] drains. No heap.
#[allow(missing_debug_implementations)]
pub struct PushToFuture<'a, S: MultiOutputSink + ?Sized> {
    sink: &'a mut S,
    port: usize,
    packet: Option<PipelinePacket>,
}

impl<S: MultiOutputSink + ?Sized> core::future::Future for PushToFuture<'_, S> {
    type Output = Result<PushOutcome, G2gError>;

    fn poll(
        self: core::pin::Pin<&mut Self>,
        cx: &mut core::task::Context<'_>,
    ) -> core::task::Poll<Self::Output> {
        let this = self.get_mut();
        this.sink.poll_push_to(cx, this.port, &mut this.packet)
    }
}

/// [`MultiOutputSink`] backed by one [`SenderSink`] per output link. Built
/// by the fan-out runner from the branch links; `push_to` forwards to the
/// addressed branch.
#[derive(Debug)]
pub struct MultiSenderSink {
    ports: Vec<SenderSink>,
}

impl MultiSenderSink {
    pub fn new(ports: Vec<SenderSink>) -> Self {
        Self { ports }
    }

    /// Bank every port's push-wait on the producing element's probe (M947), so a
    /// fan-out element's `proc` percentiles exclude the backpressure of whichever
    /// branch is slowest.
    // Only the std fan-out / demux runners build a probed multi-sink, so without
    // std this would be dead code (which the workspace denies).
    #[cfg(feature = "std")]
    pub(crate) fn set_push_wait_probe(&mut self, probe: crate::runtime::Probe) {
        for port in self.ports.iter_mut() {
            port.set_push_wait_probe(probe.clone());
        }
    }

    /// Append a port, for the growable multi-sink the dynamic duplex runner wraps
    /// this in (M1014).
    // Only that runner grows a multi-sink, so without std this would be dead code
    // (which the workspace denies).
    #[cfg(feature = "std")]
    pub(crate) fn push_port(&mut self, port: SenderSink) {
        self.ports.push(port);
    }
}

impl MultiOutputSink for MultiSenderSink {
    fn begin_push_to(&mut self, port: usize) {
        if let Some(sink) = self.ports.get_mut(port) {
            sink.begin_push();
        }
    }

    fn poll_push_to(
        &mut self,
        cx: &mut core::task::Context<'_>,
        port: usize,
        packet: &mut Option<PipelinePacket>,
    ) -> core::task::Poll<Result<PushOutcome, G2gError>> {
        // Port range is an internal invariant: `Router` clamps its selection
        // and broadcasts only over `0..port_count`, so an out-of-range port
        // is a framework bug, not a runtime error.
        let sink = self
            .ports
            .get_mut(port)
            .expect("push_to: port out of range");
        sink.poll_push(cx, packet)
    }

    fn port_count(&self) -> usize {
        self.ports.len()
    }
}

/// A terminal multi-output *source*: 0 inputs to N outputs, driven by
/// [`run_fanout_session`](crate::runtime::run_fanout_session). Where
/// [`MultiOutputElement`] demultiplexes an upstream input stream, this generates
/// its outputs itself from an external source (e.g. a WHEP session that receives
/// video + audio over one PeerConnection and emits each on its own pad). It is
/// the fan-out mirror of a [`MultiInputElement`] used as a terminal session sink.
pub trait MultiOutputSource: ElementBound {
    type RunFuture<'a>: core::future::Future<Output = Result<u64, G2gError>> + 'a
    where
        Self: 'a;

    /// Number of output pads (one per produced track).
    fn output_count(&self) -> usize;

    /// The caps this source produces on `output`. The runner fixates each and
    /// configures the matching downstream sink before [`Self::run`]. Geometry the
    /// source only learns later (e.g. H.264 dimensions from the in-band SPS) is
    /// reported as `Any`, exactly as a single-output `SourceLoop` does.
    fn output_caps(&self, output: usize) -> Result<Caps, G2gError>;

    /// Run until EOS / disconnect, pushing frames to outputs via
    /// `out.push_to(port, ..)`. The implementation MUST push a
    /// [`PipelinePacket::Eos`] to every output before returning `Ok`, so no
    /// downstream branch is stranded. Returns the count of `DataFrame`s pushed.
    fn run<'a>(&'a mut self, out: &'a mut dyn MultiOutputSink) -> Self::RunFuture<'a>;

    /// Settable properties, so a `gst-launch` line can configure the source
    /// (M727). Default none.
    fn properties(&self) -> &'static [PropertySpec] {
        &[]
    }

    fn set_property(&mut self, _name: &str, _value: PropValue) -> Result<(), PropError> {
        Err(PropError::Unknown)
    }

    fn get_property(&self, _name: &str) -> Option<PropValue> {
        None
    }

    /// Receive this instance's log name and a per-instance log category
    /// override, mirroring
    /// [`AsyncElement::set_instance_name`](crate::AsyncElement::set_instance_name)
    /// / [`set_log_category`](crate::AsyncElement::set_log_category). Default:
    /// ignore. A session source that logs about itself stores them in a
    /// [`LogName`](crate::log::LogName).
    fn set_instance_name(&mut self, _name: alloc::string::String) {}

    /// See [`set_instance_name`](Self::set_instance_name).
    fn set_log_category(&mut self, _category: alloc::string::String) {}
}

/// Dyn-safe mirror of [`MultiOutputSource`] (boxed-future `run`), so a terminal
/// fan-out source can be a graph node payload (M727), like
/// [`DynSourceLoop`](crate::runtime::DynSourceLoop) for single-output sources.
pub trait DynMultiOutputSource: ElementBound {
    fn output_count(&self) -> usize;
    fn output_caps(&self, output: usize) -> Result<Caps, G2gError>;
    fn run<'a>(
        &'a mut self,
        out: &'a mut dyn MultiOutputSink,
    ) -> BoxFuture<'a, Result<u64, G2gError>>;
    fn properties(&self) -> &'static [PropertySpec] {
        &[]
    }
    fn set_property(&mut self, _name: &str, _value: PropValue) -> Result<(), PropError> {
        Err(PropError::Unknown)
    }
    fn get_property(&self, _name: &str) -> Option<PropValue> {
        None
    }
    /// Dyn-safe mirror of [`MultiOutputSource::set_instance_name`].
    fn set_instance_name(&mut self, _name: alloc::string::String) {}
    /// Dyn-safe mirror of [`MultiOutputSource::set_log_category`].
    fn set_log_category(&mut self, _category: alloc::string::String) {}
}

impl<T: MultiOutputSource> DynMultiOutputSource for T {
    fn output_count(&self) -> usize {
        MultiOutputSource::output_count(self)
    }
    fn output_caps(&self, output: usize) -> Result<Caps, G2gError> {
        MultiOutputSource::output_caps(self, output)
    }
    fn run<'a>(
        &'a mut self,
        out: &'a mut dyn MultiOutputSink,
    ) -> BoxFuture<'a, Result<u64, G2gError>> {
        Box::pin(MultiOutputSource::run(self, out))
    }
    fn properties(&self) -> &'static [PropertySpec] {
        MultiOutputSource::properties(self)
    }
    fn set_property(&mut self, name: &str, value: PropValue) -> Result<(), PropError> {
        MultiOutputSource::set_property(self, name, value)
    }
    fn get_property(&self, name: &str) -> Option<PropValue> {
        MultiOutputSource::get_property(self, name)
    }
    fn set_instance_name(&mut self, name: alloc::string::String) {
        MultiOutputSource::set_instance_name(self, name)
    }
    fn set_log_category(&mut self, category: alloc::string::String) {
        MultiOutputSource::set_log_category(self, category)
    }
}

/// Forwarding impl so a borrowed `&mut dyn DynMultiOutputSource` can be boxed
/// as a graph payload without taking ownership.
impl<'b> DynMultiOutputSource for &'b mut (dyn DynMultiOutputSource + 'b) {
    fn output_count(&self) -> usize {
        (**self).output_count()
    }
    fn output_caps(&self, output: usize) -> Result<Caps, G2gError> {
        (**self).output_caps(output)
    }
    fn run<'a>(
        &'a mut self,
        out: &'a mut dyn MultiOutputSink,
    ) -> BoxFuture<'a, Result<u64, G2gError>> {
        (**self).run(out)
    }
    fn set_instance_name(&mut self, name: alloc::string::String) {
        (**self).set_instance_name(name)
    }
    fn set_log_category(&mut self, category: alloc::string::String) {
        (**self).set_log_category(category)
    }
}

/// Inbound side of a [`MultiDuplexSession`]: the runner hands the session a
/// stream of `(input_index, packet)` drawn from its N send-side sources, the
/// receive-end analog of the [`MultiOutputSink`] it pushes received tracks into.
/// `recv` yields `None` once every send source has ended (all senders dropped),
/// so the session can stop publishing while still draining the peer.
pub trait DuplexInbound {
    fn recv(&mut self) -> BoxFuture<'_, Option<(usize, PipelinePacket)>>;

    /// The [`ReverseChannel`] of send input `input`, for a pad the session only
    /// learned about mid-run (M1014). The runner asks
    /// [`MultiDuplexSession::reverse_channel`] for the pads that exist when the
    /// run starts, so a track attached later has no route for its PLI / BWE until
    /// the session reads it back here, keyed by the index its packets arrive on.
    /// Default `None`: the fixed-arity runner hands every channel over up front.
    fn reverse_channel(&self, _input: usize) -> Option<ReverseChannel> {
        None
    }
}

/// A terminal **duplex** session: N send-side inputs **and** M recv-side outputs
/// over one connection, with no external upstream or downstream beyond itself.
/// The union of [`MultiInputElement`] used as a terminal sink
/// ([`run_fanin_session`](crate::runtime::run_fanin_session)) and
/// [`MultiOutputSource`] ([`run_fanout_session`](crate::runtime::run_fanout_session)):
/// a `WebRtcBin`-style sendrecv PeerConnection both publishes local tracks and
/// emits the peer's tracks. Driven by
/// [`run_duplex_session`](crate::runtime::run_duplex_session).
///
/// One `run` loop owns the connection and is the sole holder of `&mut self`, so
/// (unlike the egress session, which spawns a detached task to dodge aliasing)
/// the send and recv halves share state directly: `run` selects over the inbound
/// packets (`inbound.recv()`) and the network, feeding the former into the
/// connection and pushing the latter to `out`.
pub trait MultiDuplexSession: ElementBound {
    type RunFuture<'a>: core::future::Future<Output = Result<u64, G2gError>> + 'a
    where
        Self: 'a;

    /// Number of send-side input pads (local tracks published to the peer).
    fn input_count(&self) -> usize;

    /// Number of recv-side output pads (peer tracks emitted locally).
    fn output_count(&self) -> usize;

    /// Phase 1 for one send-side input pad: narrow that input's proposed caps.
    fn intercept_caps(&self, input: usize, upstream_caps: &Caps) -> Result<Caps, G2gError>;

    /// Phase 2 for one send-side input pad: fixate and configure it (the session
    /// reads the track kind, e.g. H.264 video vs Opus audio, from these caps).
    fn configure_input(
        &mut self,
        input: usize,
        absolute_caps: &Caps,
    ) -> Result<ConfigureOutcome, G2gError>;

    /// The caps this session produces on one recv-side output pad. Geometry only
    /// learned later (e.g. H.264 dimensions from the in-band SPS) is reported as a
    /// `Range` placeholder, exactly as [`MultiOutputSource`] does.
    fn output_caps(&self, output: usize) -> Result<Caps, G2gError>;

    /// A [`ReverseChannel`] for send-side input pad `input`, if this session routes
    /// reverse signals (WebRTC PLI / BWE arriving on that track's m-line) back to
    /// the individual upstream source feeding it. The duplex runner clones it
    /// before running and polls it after each push from that source, surfacing any
    /// pending signal as a [`PushOutcome`], exactly as
    /// [`MultiInputElement::reverse_channel`] does for a fan-in session. Default
    /// `None`: no per-input reverse channel.
    fn reverse_channel(&self, _input: usize) -> Option<ReverseChannel> {
        None
    }

    /// Declare one send input pad's negotiation-time constraint, mirroring
    /// [`MultiInputElement::caps_constraint_as_input`].
    fn caps_constraint_as_input(&self, input: usize) -> CapsConstraint<'_>
    where
        Self: Sized,
    {
        CapsConstraint::LegacySink(alloc::boxed::Box::new(move |c: &Caps| {
            <Self as MultiDuplexSession>::intercept_caps(self, input, c)
        }))
    }

    /// Drive the session until the connection ends: drain `inbound` (the send-side
    /// packets, tagged with their input pad) into the connection and push received
    /// frames to `out`. Must push a [`PipelinePacket::Eos`] to every output before
    /// returning `Ok`, so no downstream branch is stranded. Returns the count of
    /// received `DataFrame`s pushed to outputs.
    fn run<'a>(
        &'a mut self,
        inbound: &'a mut dyn DuplexInbound,
        out: &'a mut dyn MultiOutputSink,
    ) -> Self::RunFuture<'a>;
}

/// Multi-output element trait variant: identical negotiation to
/// [`AsyncElement`], but `process` emits into a [`MultiOutputSink`] rather
/// than a single downstream. [`Router`] is the first implementor; user code
/// can write others (e.g. a content-based demux).
pub trait MultiOutputElement: ElementBound {
    type ProcessFuture<'a>: core::future::Future<Output = Result<(), G2gError>> + 'a
    where
        Self: 'a;

    fn intercept_caps(&self, upstream_caps: &Caps) -> Result<Caps, G2gError>;

    fn configure_pipeline(&mut self, absolute_caps: &Caps) -> Result<ConfigureOutcome, G2gError>;

    fn process<'a>(
        &'a mut self,
        packet: PipelinePacket,
        out: &'a mut dyn MultiOutputSink,
    ) -> Self::ProcessFuture<'a>;

    /// M18 step 1: declare the fan-out's input-side negotiation
    /// constraint. Default wraps `intercept_caps(...)` as a
    /// `LegacySink` (the fan-out narrows what it accepts from the
    /// upstream source; downstream branches all receive the narrowed
    /// caps via broadcast). Migrated fan-outs override to return
    /// native variants (typically `AcceptsAny` for pass-through fan-
    /// outs like `Router` whose output broadcasts the input verbatim,
    /// or `Accepts(set)` for fan-outs that filter format on the input
    /// side).
    ///
    /// Phase C FO-2 (per-branch downstream re-solve) will sit on top
    /// of this: once the runner has the fan-out's negotiated input
    /// caps, it broadcasts to each branch and runs Phase B's
    /// `re_solve_downstream_sink` per branch sink.
    fn caps_constraint_as_input(&self) -> CapsConstraint<'_>
    where
        Self: Sized,
    {
        CapsConstraint::LegacySink(alloc::boxed::Box::new(move |c: &Caps| {
            <Self as MultiOutputElement>::intercept_caps(self, c)
        }))
    }

    /// The caps output port `port` carries, if this fan-out is a **demux** whose
    /// ports each carry a distinct elementary stream, rather than a broadcast tee
    /// (M380). When every port returns `Some`, the graph solver negotiates each
    /// branch against its port's caps (decoupled from the byte-stream input), so a
    /// real decoder downstream of the port configures against its codec at startup
    /// instead of having to retype from the input caps at runtime. Default `None`
    /// for every port: a broadcast fan-out (e.g. [`Router`]) whose branches all
    /// carry the input caps and negotiate as a tee. A demux (e.g. `MkvDemuxN`)
    /// overrides it, returning `Some` for every port it exposes. The caps may be a
    /// placeholder (geometry as a `Range`), refined at runtime via the port's
    /// `CapsChanged`, exactly as a single-output demuxer's output caps are.
    fn port_output_caps(&self, _port: usize) -> Option<Caps> {
        None
    }

    /// The memory domains this fan-out accepts on its single input pad,
    /// mirroring [`AsyncElement::input_domains`](crate::AsyncElement::input_domains).
    /// Default [`DomainSet::ALL`] (no requirement). A demux that parses host
    /// bytes narrows it to `System`, and the allocation cascade turns that into a
    /// download demand on a GPU producer feeding it.
    fn input_domains(&self) -> DomainSet {
        DomainSet::ALL
    }

    /// Receive this instance's log name and a per-instance log category
    /// override, mirroring
    /// [`AsyncElement::set_instance_name`](crate::AsyncElement::set_instance_name)
    /// / [`set_log_category`](crate::AsyncElement::set_log_category). Default:
    /// ignore. A demux that logs about itself stores them in a
    /// [`LogName`](crate::log::LogName).
    fn set_instance_name(&mut self, _name: alloc::string::String) {}

    /// See [`set_instance_name`](Self::set_instance_name).
    fn set_log_category(&mut self, _category: alloc::string::String) {}

    /// Runtime properties this demux exposes (M104), mirroring
    /// [`AsyncElement::properties`](crate::AsyncElement::properties). Default:
    /// none. A demux overrides this (with `set_property` / `get_property`) to be
    /// settable by name from a `gst-launch` line, the same as a transform.
    fn properties(&self) -> &'static [PropertySpec] {
        &[]
    }

    /// Set a property by name (M104). Default: every name is unknown.
    fn set_property(&mut self, _name: &str, _value: PropValue) -> Result<(), PropError> {
        Err(PropError::Unknown)
    }

    /// Read a property back by name (M104). Default: `None`.
    fn get_property(&self, _name: &str) -> Option<PropValue> {
        None
    }
}

/// Multi-input element trait variant: an N-input, 1-output element (a
/// muxer). The mirror of [`MultiOutputElement`]. Negotiation is **per
/// input** — each input pad narrows and fixates its own caps — and the
/// element exposes a single merged `output_caps`. The fan-in runner
/// (`run_muxer_sink`) aggregates EOS itself, so `process` is only ever
/// handed `DataFrame`/`CapsChanged`, tagged with the originating `input`.
pub trait MultiInputElement: ElementBound {
    type ProcessFuture<'a>: core::future::Future<Output = Result<(), G2gError>> + 'a
    where
        Self: 'a;

    fn input_count(&self) -> usize;

    /// Whether the runner should deliver this element's inputs in global
    /// presentation-timestamp order. Default `false` (arrival-order round-robin,
    /// the historical behavior). When `true`, the runner merges the per-input
    /// streams by `DataFrame` PTS, releasing the globally-earliest only once every
    /// still-open input has one queued, so `process(pad, DataFrame(..))` arrives in
    /// non-decreasing PTS across all pads.
    ///
    /// An element wanting time-aligned input without hand-rolling an
    /// [`InputAggregator`](crate::InputAggregator) (a muxer, a multi-camera grid, a
    /// PTS-synchronized compositor) opts in by returning `true`. Per-input `Eos`
    /// and `CapsChanged` are still delivered as they occur; the merge holds only
    /// `DataFrame`s. Inputs are assumed monotonic in PTS (the merge invariant).
    fn input_pts_ordered(&self) -> bool {
        false
    }

    /// Deadline tick period in nanoseconds, or `None` (the default) for no ticks.
    ///
    /// When set, and the runner was given a clock to sleep on, the fan-in arm
    /// delivers [`PipelinePacket::Tick`] as `process(0, Tick, ..)` every period
    /// even while the inputs are silent, so an element whose output cadence is its
    /// own (a compositor holding the last frame of a stalled pad, zero-order-hold)
    /// can emit without a packet arriving. The tick may fire spuriously: it says
    /// the period elapsed, not that output is due, so the element decides.
    fn tick_interval_ns(&self) -> Option<u64> {
        None
    }

    /// Phase 1 for one input pad: narrow that input's proposed caps.
    fn intercept_caps(&self, input: usize, upstream_caps: &Caps) -> Result<Caps, G2gError>;

    /// Phase 2 for one input pad: fixate and configure that input.
    fn configure_pipeline(
        &mut self,
        input: usize,
        absolute_caps: &Caps,
    ) -> Result<ConfigureOutcome, G2gError>;

    /// The merged-output caps, valid once every input has been configured.
    fn output_caps(&self) -> Result<Caps, G2gError>;

    /// A [`ReverseChannel`] for input pad `input`, if this session routes reverse
    /// signals (WebRTC PLI / BWE) back to individual upstream sources. The runner
    /// clones it before running and polls it after each push from that source,
    /// surfacing any pending signal as a [`PushOutcome`]. Default `None`: no
    /// per-input reverse channel (a plain muxer imposes none).
    fn reverse_channel(&self, _input: usize) -> Option<ReverseChannel> {
        None
    }

    /// Whether this element is a terminal fan-in (a session sink that consumes
    /// its inputs and produces no merged output, e.g. a WebRTC publisher). A
    /// terminal element may end a graph with nothing downstream
    /// (`Graph::add_fanin_sink`); a merging muxer without a downstream stays a
    /// parse error, since its output would be silently dropped. Default `false`.
    fn is_terminal(&self) -> bool {
        false
    }

    /// If `Some(pad)`, the merged output's caps are the negotiated caps of input
    /// pad `pad` (an identity-passthrough mux: an overlay / watermark / alpha
    /// mixer that decorates a primary stream with a sidecar one). The solver then
    /// derives the output edge from that input edge instead of from
    /// [`caps_constraint_for_output`](Self::caps_constraint_for_output), so the
    /// element need not know the output caps up front. Default `None`: the output
    /// is independent (a container interleave, a fixed-size compositor), declared
    /// by `caps_constraint_for_output`.
    fn output_follows_input(&self) -> Option<usize> {
        None
    }

    /// Map a named input-pad request to this element's concrete input index
    /// (M481), so a `gst-launch` line can reference request pads by name
    /// (`... ! mux.audio_0`, `... ! o.text`) instead of relying on the order the
    /// branches are written. The transpose of the demuxer's output-pad selection
    /// (M476): `req.kind` is `Video` / `Audio` / `Text` / `Any` with an ordinal.
    ///
    /// The default maps a bare `mux.` (`PadKind::Any`) to `ordinal` (the
    /// positional behavior, unchanged), and declines a typed request (`None`) so a
    /// homogeneous muxer without a naming scheme reports "no such pad" rather than
    /// silently mis-routing. An element with named pads overrides this: an overlay
    /// maps `Video -> 0`, `Text -> 1`; a container mux maps `video_%u` / `audio_%u`.
    fn input_pad_index(&self, req: &PadRequest, ordinal: usize) -> Option<usize> {
        let _ = ordinal;
        match req.kind {
            // A generic request pad (`sink_%u`, or a bare `mux.` whose index the
            // parser set to the positional ordinal) maps straight to its index.
            PadKind::Any => Some(req.index),
            // A typed pad (`video_%u` / `audio_%u` / `text_%u`) needs an element
            // scheme; a muxer without one declines it (reported as "no such pad").
            _ => None,
        }
    }

    /// Whether this element takes another input at runtime (M975), asked before a
    /// source added through
    /// [`DynamicFaninHandle::add_input`](crate::runtime::DynamicFaninHandle::add_input)
    /// is attached to `pad`, with the caps that source would arrive with. The
    /// runner has already checked `pad` against the declared
    /// [`input_count`](Self::input_count) and the pad's own
    /// [`caps_constraint_as_input`](Self::caps_constraint_as_input); this is the
    /// element's veto for what neither expresses: a session with no spare pad of
    /// that media kind, a container that cannot carry a second video track.
    /// Refusing fails that one add ([`G2gError::InputRefused`] reaches the
    /// caller), and the run continues on the inputs it already has.
    ///
    /// Default `true`: pad capacity is the only limit, the fixed-arity behavior.
    /// The pad is not live until [`configure_pipeline`](Self::configure_pipeline)
    /// runs for it, so do not commit pad state here.
    fn accepts_runtime_input(&self, _pad: usize, _caps: &Caps) -> bool {
        true
    }

    /// Combine one packet from `input` into the merged output.
    ///
    /// M22: a per-input `Eos` is delivered here when that input ends, so a
    /// stateful muxer (a batcher) can flush per-input state. Implementations
    /// must NOT forward `Eos` downstream: the runner aggregates input ends
    /// and emits the single merged `Eos` itself.
    fn process<'a>(
        &'a mut self,
        input: usize,
        packet: PipelinePacket,
        out: &'a mut dyn OutputSink,
    ) -> Self::ProcessFuture<'a>;

    /// M18 step 1: declare this input pad's negotiation-time
    /// constraint. Default wraps `intercept_caps(input, ...)` as a
    /// `LegacySink` (per-pad legacy bridge). Migrated muxers override
    /// to return native variants (typically `AcceptsAny` for
    /// per-frame-tagged interleave muxers, or `Accepts(set)` for
    /// per-input format-restricted muxers).
    ///
    /// The runner calls this per-input during startup negotiation
    /// (replacing the inline `LegacySink` construction in
    /// `run_muxer_sink`) and during per-input mid-stream re-solve
    /// once Phase C MX-1 lands.
    fn caps_constraint_as_input(&self, input: usize) -> CapsConstraint<'_>
    where
        Self: Sized,
    {
        CapsConstraint::LegacySink(alloc::boxed::Box::new(move |c: &Caps| {
            <Self as MultiInputElement>::intercept_caps(self, input, c)
        }))
    }

    /// M18 step 1: declare the merged output's negotiation-time
    /// constraint, evaluated against the muxer's current configured
    /// inputs. Default eagerly calls `output_caps()` and wraps as
    /// `LegacySource`. Migrated muxers with static or input-derived
    /// output may override with `Produces(set)` or `DerivedOutput(fn)`.
    ///
    /// The runner uses this in place of `output_caps()?.fixate()` so
    /// the downstream sink sees a uniformly-shaped constraint and the
    /// Phase B-style re-solve (workaround #3 §4) extends naturally to
    /// the muxer-output boundary.
    fn caps_constraint_for_output(&self) -> Result<CapsConstraint<'_>, G2gError>
    where
        Self: Sized,
    {
        Ok(CapsConstraint::LegacySource(self.output_caps()?))
    }

    /// The allocation this muxer wants on one input pad, given that pad's
    /// negotiated caps. The DAG runner stores it on the input edge during the
    /// reverse-topo allocation cascade, so the demand crosses the muxer boundary
    /// and re-cascades up that branch independently (a CUDA-resident interleave
    /// muxer asking each video pad for device buffers, say). Default `None`: a
    /// plain container muxer imposes no per-pad allocation, so the branch keeps
    /// its own. Mirrors [`AsyncElement::propose_allocation`](crate::AsyncElement::propose_allocation),
    /// but per input pad rather than on the single input.
    fn propose_allocation_for_input(
        &self,
        _input: usize,
        _caps: &Caps,
    ) -> Option<crate::query::AllocationParams> {
        None
    }

    /// The allocation this muxer's merged output needs, given the output's
    /// negotiated caps. Default `None`: a container muxer's byte output has no
    /// memory-domain tie to its inputs, so it imposes nothing downstream. A muxer
    /// whose output pool is derived from its inputs overrides it (a device-resident
    /// interleave writing into a surface sized by its video pads).
    ///
    /// The DAG runner re-queries this whenever an allocation change re-cascades
    /// into any input pad, so the answer must fold in the current per-pad state
    /// *and* whatever the element last absorbed through
    /// [`configure_allocation_for_output`](Self::configure_allocation_for_output).
    /// The runner walks to a fixed point of that pair, so an override that never
    /// stops changing fails the run with `AllocationConflict` rather than looping.
    fn propose_allocation_for_output(
        &self,
        _caps: &Caps,
    ) -> Option<crate::query::AllocationParams> {
        None
    }

    /// Absorb the allocation now in force on the merged output: this muxer's own
    /// re-derived proposal, or a downstream consumer's demand that re-cascaded
    /// into the output boundary. Default: ignore.
    ///
    /// A muxer whose output pool constrains its inputs overrides it and folds the
    /// params into its state, so the following
    /// [`propose_allocation_for_input`](Self::propose_allocation_for_input) answers
    /// carry the constraint and the runner re-cascades it up the pads whose demand
    /// actually moved.
    fn configure_allocation_for_output(&mut self, _params: &crate::query::AllocationParams) {}

    /// The memory domains this fan-in accepts on **every** input pad, mirroring
    /// [`AsyncElement::input_domains`](crate::AsyncElement::input_domains).
    /// Default [`DomainSet::ALL`] (no requirement). A muxer that reads host
    /// memory narrows it to `System`, and the allocation cascade turns that into
    /// a download demand on each GPU producer feeding a pad. Per-pad domains are
    /// not expressible: a fan-in whose pads differ declares the union it can take
    /// on any pad and rejects the rest at
    /// [`configure_pipeline`](Self::configure_pipeline).
    fn input_domains(&self) -> DomainSet {
        DomainSet::ALL
    }

    /// Receive this instance's log name and a per-instance log category
    /// override, mirroring
    /// [`AsyncElement::set_instance_name`](crate::AsyncElement::set_instance_name)
    /// / [`set_log_category`](crate::AsyncElement::set_log_category). Default:
    /// ignore. A muxer that logs about itself stores them in a
    /// [`LogName`](crate::log::LogName).
    fn set_instance_name(&mut self, _name: alloc::string::String) {}

    /// See [`set_instance_name`](Self::set_instance_name).
    fn set_log_category(&mut self, _category: alloc::string::String) {}

    /// Runtime properties this muxer exposes (M104), mirroring
    /// [`AsyncElement::properties`](crate::AsyncElement::properties). Default:
    /// none. A muxer overrides this (with `set_property` / `get_property`) to be
    /// settable by name from a `gst-launch` line, the same as a transform.
    fn properties(&self) -> &'static [PropertySpec] {
        &[]
    }

    /// Set a property by name (M104). Default: every name is unknown.
    fn set_property(&mut self, _name: &str, _value: PropValue) -> Result<(), PropError> {
        Err(PropError::Unknown)
    }

    /// Read a property back by name (M104). Default: `None`.
    fn get_property(&self, _name: &str) -> Option<PropValue> {
        None
    }

    /// Static introspection metadata for this muxer (M178), the `gst-inspect`
    /// "Factory Details" (long-name / classification / description / author),
    /// mirroring [`AsyncElement::metadata`](crate::AsyncElement::metadata).
    /// Default: empty. A muxer overrides it with a `const ElementMetadata`.
    fn metadata(&self) -> ElementMetadata {
        ElementMetadata::default()
    }
}

/// 1→1 enable/disable element. Forwards `CapsChanged` unconditionally and
/// `DataFrame` only while open; `Eos` is forwarded by the runner, never by
/// the element (the transform contract). Drops dropped frames silently —
/// observability of gate drops is a tracing concern for a later milestone.
///
/// # Example
///
/// ```no_run
/// use g2g_core::fanout::Gate;
///
/// let gate = Gate::new(true);
/// let handle = gate.handle();
/// handle.set_open(false);
/// ```
#[derive(Debug)]
pub struct Gate {
    open: Arc<AtomicBool>,
}

impl Gate {
    pub fn new(open: bool) -> Self {
        Self {
            open: Arc::new(AtomicBool::new(open)),
        }
    }

    /// A cloneable handle that flips this gate from another task while the
    /// runner drives it.
    pub fn handle(&self) -> GateHandle {
        GateHandle {
            open: self.open.clone(),
        }
    }
}

/// Detached control handle for a [`Gate`].
#[derive(Debug, Clone)]
pub struct GateHandle {
    open: Arc<AtomicBool>,
}

impl GateHandle {
    pub fn set_open(&self, open: bool) {
        self.open.store(open, Ordering::SeqCst);
    }

    pub fn is_open(&self) -> bool {
        self.open.load(Ordering::SeqCst)
    }
}

impl AsyncElement for Gate {
    type ProcessFuture<'a>
        = BoxFuture<'a, Result<(), G2gError>>
    where
        Self: 'a;

    fn intercept_caps(&self, upstream_caps: &Caps) -> Result<Caps, G2gError> {
        Ok(upstream_caps.clone())
    }

    fn configure_pipeline(&mut self, _absolute_caps: &Caps) -> Result<ConfigureOutcome, G2gError> {
        Ok(ConfigureOutcome::Accepted)
    }

    fn process<'a>(
        &'a mut self,
        packet: PipelinePacket,
        out: &'a mut dyn OutputSink,
    ) -> Self::ProcessFuture<'a> {
        let open = self.open.load(Ordering::SeqCst);
        Box::pin(async move {
            match packet {
                PipelinePacket::DataFrame(f) => {
                    if open {
                        out.push(PipelinePacket::DataFrame(f)).await?;
                    }
                }
                PipelinePacket::CapsChanged(c) => {
                    out.push(PipelinePacket::CapsChanged(c)).await?;
                }
                // Flush is control: forward regardless of open state.
                PipelinePacket::Flush => {
                    out.push(PipelinePacket::Flush).await?;
                }
                // Segment is control: forward regardless of open state.
                PipelinePacket::Segment(s) => {
                    out.push(PipelinePacket::Segment(s)).await?;
                }
                // Runner forwards Eos after process() returns. A fan-in arm's
                // deadline tick never reaches a 1->1 element.
                PipelinePacket::Eos | PipelinePacket::Tick => {}
            }
            Ok(())
        })
    }
}

/// 1→N router. Each `DataFrame` goes to the single port named by an atomic
/// discriminator; `CapsChanged` is broadcast to every port so all branches
/// stay configured. `Eos` is broadcast by the runner.
///
/// # Example
///
/// ```no_run
/// use g2g_core::fanout::Router;
///
/// let router = Router::new(3);
/// let handle = router.handle();
/// handle.select(2);
/// ```
#[derive(Debug)]
pub struct Router {
    selected: Arc<AtomicUsize>,
    ports: usize,
}

impl Router {
    pub fn new(ports: usize) -> Self {
        assert!(ports > 0, "Router needs at least one output port");
        Self {
            selected: Arc::new(AtomicUsize::new(0)),
            ports,
        }
    }

    /// Number of output ports. The fan-out runner allocates one branch link
    /// per port.
    pub fn port_count(&self) -> usize {
        self.ports
    }

    /// A cloneable handle that re-targets this router from another task.
    pub fn handle(&self) -> RouterHandle {
        RouterHandle {
            selected: self.selected.clone(),
            ports: self.ports,
        }
    }
}

/// Detached control handle for a [`Router`].
#[derive(Debug, Clone)]
pub struct RouterHandle {
    selected: Arc<AtomicUsize>,
    ports: usize,
}

impl RouterHandle {
    /// Select the output port subsequent `DataFrame`s route to. Panics if
    /// `port >= port_count`.
    pub fn select(&self, port: usize) {
        assert!(port < self.ports, "select: port out of range");
        self.selected.store(port, Ordering::SeqCst);
    }

    pub fn selected(&self) -> usize {
        self.selected.load(Ordering::SeqCst)
    }
}

impl MultiOutputElement for Router {
    type ProcessFuture<'a>
        = BoxFuture<'a, Result<(), G2gError>>
    where
        Self: 'a;

    fn intercept_caps(&self, upstream_caps: &Caps) -> Result<Caps, G2gError> {
        Ok(upstream_caps.clone())
    }

    /// M18 step 1: pass-through wildcard. `Router` broadcasts the
    /// upstream caps verbatim to every active branch and has no
    /// per-branch format restriction. `AcceptsAny` is the native
    /// shape; skips the dynamic intercept callback on the solver
    /// path.
    fn caps_constraint_as_input(&self) -> CapsConstraint<'_> {
        CapsConstraint::AcceptsAny
    }

    fn configure_pipeline(&mut self, _absolute_caps: &Caps) -> Result<ConfigureOutcome, G2gError> {
        Ok(ConfigureOutcome::Accepted)
    }

    fn process<'a>(
        &'a mut self,
        packet: PipelinePacket,
        out: &'a mut dyn MultiOutputSink,
    ) -> Self::ProcessFuture<'a> {
        // Clamp defensively so a stale handle write can never index past the
        // port list (the runner allocated exactly `ports` branches).
        let selected = self.selected.load(Ordering::SeqCst).min(self.ports - 1);
        let ports = self.ports;
        Box::pin(async move {
            match packet {
                PipelinePacket::DataFrame(f) => {
                    out.push_to(selected, PipelinePacket::DataFrame(f)).await?;
                }
                PipelinePacket::CapsChanged(c) => {
                    for port in 0..ports {
                        out.push_to(port, PipelinePacket::CapsChanged(c.clone()))
                            .await?;
                    }
                }
                // Flush is broadcast to every branch, like CapsChanged.
                PipelinePacket::Flush => {
                    for port in 0..ports {
                        out.push_to(port, PipelinePacket::Flush).await?;
                    }
                }
                // Segment is broadcast to every branch, like CapsChanged.
                PipelinePacket::Segment(s) => {
                    for port in 0..ports {
                        out.push_to(port, PipelinePacket::Segment(s)).await?;
                    }
                }
                // Runner broadcasts Eos to all ports after process() returns. A
                // fan-in arm's deadline tick never reaches a 1->N element.
                PipelinePacket::Eos | PipelinePacket::Tick => {}
            }
            Ok(())
        })
    }
}

/// N→1 fan-in selector: the control-driven mirror of [`Router`]. An atomic
/// discriminator names the single active input; the fan-in runner forwards
/// that input's frames and drains/discards the rest. The merged stream ends
/// only once every input has reached EOS (see `run_fanin_sink`). `Merger`
/// holds just the selector; the forwarding lives in the runner.
///
/// # Example
///
/// ```no_run
/// use g2g_core::fanout::Merger;
///
/// let merger = Merger::new(2);
/// let handle = merger.handle();
/// handle.select(1);
/// ```
#[derive(Debug)]
pub struct Merger {
    selected: Arc<AtomicUsize>,
    inputs: usize,
}

impl Merger {
    pub fn new(inputs: usize) -> Self {
        assert!(inputs > 0, "Merger needs at least one input");
        Self {
            selected: Arc::new(AtomicUsize::new(0)),
            inputs,
        }
    }

    /// Number of input ports. The fan-in runner allocates one branch link
    /// per input.
    pub fn input_count(&self) -> usize {
        self.inputs
    }

    /// A cloneable handle that re-selects the active input from another task.
    pub fn handle(&self) -> MergerHandle {
        MergerHandle {
            selected: self.selected.clone(),
            inputs: self.inputs,
        }
    }
}

/// Detached control handle for a [`Merger`].
#[derive(Debug, Clone)]
pub struct MergerHandle {
    selected: Arc<AtomicUsize>,
    inputs: usize,
}

impl MergerHandle {
    /// Select which input feeds the merged output. Panics if
    /// `input >= input_count`.
    pub fn select(&self, input: usize) {
        assert!(input < self.inputs, "select: input out of range");
        self.selected.store(input, Ordering::SeqCst);
    }

    pub fn selected(&self) -> usize {
        self.selected.load(Ordering::SeqCst)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::caps::{Dim, Rate, RawVideoFormat};
    use crate::frame::{Frame, FrameTiming};
    use crate::memory::{MemoryDomain, SystemSlice};
    use core::future::Future;
    use core::pin::Pin;

    fn caps() -> Caps {
        Caps::RawVideo {
            format: RawVideoFormat::Rgba8,
            width: Dim::Fixed(16),
            height: Dim::Fixed(16),
            framerate: Rate::Fixed(30 << 16),
            interlace: crate::Interlace::Any,
        }
    }

    fn data(seq: u64) -> PipelinePacket {
        PipelinePacket::DataFrame(Frame {
            domain: MemoryDomain::System(SystemSlice::from_boxed(Box::new([0u8; 4]))),
            timing: FrameTiming::default(),
            sequence: seq,
            meta: Default::default(),
        })
    }

    /// Records the kind of every packet pushed, per port, without channels.
    #[derive(Default)]
    struct RecordingMultiSink {
        ports: usize,
        data_seqs: Vec<Vec<u64>>,
        caps_changes: Vec<usize>,
    }

    impl RecordingMultiSink {
        fn new(ports: usize) -> Self {
            Self {
                ports,
                data_seqs: alloc::vec![Vec::new(); ports],
                caps_changes: alloc::vec![0; ports],
            }
        }
    }

    impl MultiOutputSink for RecordingMultiSink {
        fn poll_push_to(
            &mut self,
            _cx: &mut core::task::Context<'_>,
            port: usize,
            packet: &mut Option<PipelinePacket>,
        ) -> core::task::Poll<Result<PushOutcome, G2gError>> {
            match packet.take().expect("poll_push_to without a packet") {
                PipelinePacket::DataFrame(f) => self.data_seqs[port].push(f.sequence),
                PipelinePacket::CapsChanged(_) => self.caps_changes[port] += 1,
                PipelinePacket::Eos
                | PipelinePacket::Flush
                | PipelinePacket::Segment(_)
                | PipelinePacket::Tick => {}
            }
            core::task::Poll::Ready(Ok(PushOutcome::Accepted))
        }

        fn port_count(&self) -> usize {
            self.ports
        }
    }

    /// Records every packet a single-output element forwards.
    #[derive(Default)]
    struct RecordingSink {
        data_seqs: Vec<u64>,
        caps_changes: usize,
    }

    impl OutputSink for RecordingSink {
        fn poll_push(
            &mut self,
            _cx: &mut core::task::Context<'_>,
            packet: &mut Option<PipelinePacket>,
        ) -> core::task::Poll<Result<PushOutcome, G2gError>> {
            match packet.take().expect("poll_push without a packet") {
                PipelinePacket::DataFrame(f) => self.data_seqs.push(f.sequence),
                PipelinePacket::CapsChanged(_) => self.caps_changes += 1,
                PipelinePacket::Eos
                | PipelinePacket::Flush
                | PipelinePacket::Segment(_)
                | PipelinePacket::Tick => {}
            }
            core::task::Poll::Ready(Ok(PushOutcome::Accepted))
        }
    }

    /// Single-poll block_on; all futures here resolve immediately.
    fn block_on<F: Future>(mut fut: F) -> F::Output {
        use core::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
        static VT: RawWakerVTable = RawWakerVTable::new(
            |_| RawWaker::new(core::ptr::null(), &VT),
            |_| {},
            |_| {},
            |_| {},
        );
        // SAFETY: VT's hooks never dereference the data pointer.
        let waker = unsafe { Waker::from_raw(RawWaker::new(core::ptr::null(), &VT)) };
        let mut cx = Context::from_waker(&waker);
        // SAFETY: `fut` is pinned to the stack for the duration of this call.
        let mut pinned = unsafe { Pin::new_unchecked(&mut fut) };
        match pinned.as_mut().poll(&mut cx) {
            Poll::Ready(v) => v,
            Poll::Pending => panic!("fanout::tests::block_on saw Pending"),
        }
    }

    #[test]
    fn router_input_constraint_is_wildcard() {
        // M18 step 1: Router broadcasts upstream caps verbatim, no
        // per-branch format restriction. AcceptsAny is the native
        // shape; skips the dynamic intercept callback on the solver
        // path.
        let r = Router::new(3);
        let c = r.caps_constraint_as_input();
        assert!(
            matches!(c, CapsConstraint::AcceptsAny),
            "Router input should be AcceptsAny, got {c:?}"
        );
    }

    #[test]
    fn router_sends_each_frame_to_selected_port() {
        let mut router = Router::new(2);
        let handle = router.handle();
        let mut out = RecordingMultiSink::new(2);

        block_on(router.process(data(0), &mut out)).unwrap(); // port 0
        handle.select(1);
        block_on(router.process(data(1), &mut out)).unwrap(); // port 1
        block_on(router.process(data(2), &mut out)).unwrap(); // port 1 (sticky)
        handle.select(0);
        block_on(router.process(data(3), &mut out)).unwrap(); // port 0

        assert_eq!(out.data_seqs[0], alloc::vec![0, 3]);
        assert_eq!(out.data_seqs[1], alloc::vec![1, 2]);
    }

    #[test]
    fn router_broadcasts_caps_changed_to_all_ports() {
        let mut router = Router::new(3);
        let mut out = RecordingMultiSink::new(3);

        block_on(router.process(PipelinePacket::CapsChanged(caps()), &mut out)).unwrap();

        assert_eq!(out.caps_changes, alloc::vec![1, 1, 1]);
    }

    #[test]
    fn gate_open_forwards_data_closed_drops_it() {
        let gate = Gate::new(true);
        let handle = gate.handle();
        let mut gate = gate;
        let mut out = RecordingSink::default();

        block_on(gate.process(data(0), &mut out)).unwrap(); // open -> pass
        handle.set_open(false);
        block_on(gate.process(data(1), &mut out)).unwrap(); // closed -> drop
        handle.set_open(true);
        block_on(gate.process(data(2), &mut out)).unwrap(); // open -> pass

        assert_eq!(
            out.data_seqs,
            alloc::vec![0, 2],
            "frame 1 dropped while closed"
        );
    }

    #[test]
    fn gate_forwards_caps_changed_regardless_of_open_state() {
        let mut gate = Gate::new(false);
        let mut out = RecordingSink::default();

        block_on(gate.process(PipelinePacket::CapsChanged(caps()), &mut out)).unwrap();

        assert_eq!(
            out.caps_changes, 1,
            "CapsChanged forwarded even while closed"
        );
    }

    #[test]
    fn merger_handle_selects_active_input() {
        let merger = Merger::new(3);
        let handle = merger.handle();
        assert_eq!(handle.selected(), 0, "defaults to input 0");
        handle.select(2);
        assert_eq!(handle.selected(), 2);
        assert_eq!(merger.input_count(), 3);
    }

    #[test]
    #[should_panic(expected = "input out of range")]
    fn merger_handle_rejects_out_of_range_input() {
        Merger::new(2).handle().select(2);
    }
}