g2g_core/runtime/

graph_runner.rs

//! DAG pipeline runner (DESIGN_TODO "DAG runner" D3).
//!
//! [`run_graph`] drives an arbitrary multimedia DAG built with [`Graph`]:
//! whole-graph CSP negotiation via [`solve_graph`] (D2), then one spawned arm
//! per node over per-edge channels, joined with [`join_all`]. It collapses the
//! linear + fan-out runner shapes into one entry point.
//!
//! Scope: source / transform / sink / tee (fan-out) + muxer (fan-in) + terminal
//! fan-in (M713). A tee
//! broadcasts each packet to all its branches via [`MemoryDomain::share`]
//! (M213): a zero-copy refcount bump for the GPU domains and the shared-CPU
//! `SystemView`, a deep copy only for owned-CPU `System` bytes. So a GPU-decoded
//! frame fans out to several consumers (inference + display) with no
//! device-to-host copy. A muxer node
//! runs a [`DynMultiInputElement`]: per-input forwarder arms tag each packet
//! with its pad and feed one muxer arm that combines them, emitting a single
//! `Eos` after every input ends (the `run_muxer_sink` shape). A terminal fan-in
//! node ([`Graph::add_fanin_sink`]) runs a `DynMultiInputElement` with NO
//! downstream (the `run_fanin_session` shape, e.g. a WebRTC session publishing
//! its inputs): one arm drains the input edges round-robin, delivers per-input
//! `Eos` flushes, ends once every input has ended, and relays each pad's
//! reverse signal (PLI / BWE) onto its own in-edge so it reaches the encoder
//! feeding that pad.
//!
//! D4 adds the mid-stream re-solve and the β allocation re-cascade over the
//! DAG. Each arm gets a per-edge downstream feasibility snapshot at startup
//! ([`graph_downstream_feasibility`]); on a mid-stream `CapsChanged` a transform
//! steers its forwarded output toward a downstream-acceptable shape (Caps-α),
//! and a sink re-solves its input against its declared constraint. A node-keyed
//! [`GraphCoordinator`] walks the sink's re-derived allocation proposal one hop
//! upstream per reply via [`ValidatedGraph::in_edges`], resolving through
//! structural tee nodes; a source terminates the walk. Tee branches re-solve
//! independently (each broadcast `CapsChanged` lands in its own arm); muxer
//! inputs re-configure per pad. A muxer's per-pad allocation demand
//! (`propose_allocation_for_input`) crosses the boundary both at startup
//! negotiation and mid-stream: a `CapsChanged` on one pad re-cascades the
//! re-derived proposal up that pad's branch alone (`RecascadeRoute::Pad`),
//! leaving the other inputs untouched. From there the walk continues *through*
//! the boundary (M839, [`MuxBeta`]): the merged-output pool is re-derived from
//! every pad, pushed to the consumer arms, and absorbed by the element, and the
//! other pads re-cascade only if that moved their demand.

use alloc::boxed::Box;
use alloc::string::ToString;
use alloc::vec::Vec;
use core::pin::Pin;

// The thread-per-arm `ThreadSpawner` uses `std::thread`; `std` is otherwise
// unused by this (no_std baseline) module.
#[cfg(all(feature = "std", feature = "multi-thread"))]
extern crate std;

use crate::aggregator::InputAggregator;
use crate::bus::{BusHandle, BusMessage};
use crate::caps::{Caps, CapsSet};
use crate::clock::{
    elect_clock, ClockCandidate, ClockPriority, ClockSync, DynAsyncClock, ElectedClock,
    PipelineClock,
};
use crate::controller::{ArmController, ControlTarget, CONTROL_CATEGORY};
use crate::element::{
    AsyncElement, BoxFuture, ConfigureOutcome, DynAsyncElement, ElementBound, OutputSink,
    OutputSinkExt, PushOutcome, Reconfigure,
};
use crate::error::G2gError;
use crate::fanout::{
    DynMultiOutputSource, MultiInputElement, MultiOutputElement, MultiOutputSink,
    MultiOutputSinkExt, MultiOutputSource, MultiSenderSink, ReverseChannel,
};
use crate::format_element::{CapsConstraint, CapsPreferences};
use crate::frame::{Frame, PipelinePacket};
use crate::graph::{FanOutPolicy, Graph, NodeId, NodeKind, ValidatedGraph};
use crate::memory::{DomainSet, MemoryDomainKind};
use crate::meta::MetaRequests;
use crate::property::{PropError, PropValue, PropertySpec};
use crate::query::{with_meta_demand, AllocationParams, LatencyReport};
use crate::runtime::channel::{
    advertise_orientation, bounded, link, link_with_transit, LinkReceiver, LinkSender, Receiver,
    ReconfigureAnswered, RecvFuture, Sender, SenderSink,
};
use crate::runtime::coordinator::{
    log_caps_forward, log_caps_rejected, realloc_local_dyn, report_nego_failure, ArmDirective,
};
use crate::runtime::fanin::{DynMultiInputElement, DynSourceLoop};
use crate::runtime::instrument::{snapshot_all, ElementProbe, Probe};
use crate::runtime::join::{join_all, select2, Either};
use crate::runtime::progress::PipelineProgress;
use crate::runtime::runner::{
    re_solve_downstream_dyn_sink, LinkCapacity, NullSink, RunStats, SourceLoop,
};
use crate::runtime::solver::{
    graph_downstream_feasibility, resolve_forward_output, solve_graph_preferred, solve_linear,
    ForwardResolve, NegotiationFailure, NodeConstraint,
};
use crate::runtime::state::{Flow, StateController};
use crate::runtime::FlightRecorder;
use crate::runtime::Observer;
use crate::segment::Segment;

/// Element payload for a [`Graph`] driven by [`run_graph`]. Sources,
/// transforms/sinks, and muxers implement different traits (a source has no
/// input pad, a muxer has many), so the payload is an enum the runner matches
/// on per node kind. A tee carries no element (`Graph::add_tee` takes none).
///
/// The `'a` lifetime is the lifetime of the boxed elements. Owned `'static`
/// elements (the common case, [`GraphNode`]) use `source` / `element` / `muxer`;
/// the convenience wrappers build a *borrowing* graph over their `&mut` element
/// references with `source_ref` / `element_ref` / `muxer_ref`, so they can call
/// `run_graph` without taking ownership and the caller keeps its elements.
pub enum GraphNodeRef<'a> {
    Source(Box<dyn DynSourceLoop + 'a>),
    Element(Box<dyn DynAsyncElement + 'a>),
    Muxer(Box<dyn DynMultiInputElement + 'a>),
    /// A terminal fan-out source (M727): 0 inputs, N outputs it generates
    /// itself (a WebRTC session receiving several tracks). `Graph::add_fanout_src`.
    FanoutSource(Box<dyn DynMultiOutputSource + 'a>),
    /// A content-routing demultiplexer: 1 input, N outputs. Structurally a tee
    /// (its node kind is `Tee(n)`), so it negotiates as a tee at startup, but it
    /// carries a [`MultiOutputElement`] that routes each packet to a chosen
    /// output instead of broadcasting, and emits per-output `CapsChanged` so each
    /// branch retypes from the byte-stream input (M210). `Graph::add_demux`.
    Demux(Box<dyn DynMultiOutputElement + 'a>),
}

/// The owning, `'static` graph payload: what most callers build directly.
pub type GraphNode = GraphNodeRef<'static>;

impl<'a> GraphNodeRef<'a> {
    /// Box an owned source (`add_source`).
    pub fn source<S: SourceLoop + 'static>(source: S) -> Self {
        GraphNodeRef::Source(Box::new(source))
    }

    /// Box an owned transform or sink (`add_transform` / `add_sink`).
    pub fn element<E: AsyncElement + 'static>(element: E) -> Self {
        GraphNodeRef::Element(Box::new(element))
    }

    /// Box an owned fan-in muxer (`add_muxer`).
    pub fn muxer<M: MultiInputElement + 'static>(muxer: M) -> Self {
        GraphNodeRef::Muxer(Box::new(muxer))
    }

    /// Box an owned terminal fan-out source (`add_fanout_src`).
    pub fn fanout_source<S: MultiOutputSource + 'static>(source: S) -> Self {
        GraphNodeRef::FanoutSource(Box::new(source))
    }

    /// Box a borrowed source, for a borrowing graph (the convenience wrappers).
    pub fn source_ref(source: &'a mut (dyn DynSourceLoop + 'a)) -> Self {
        GraphNodeRef::Source(Box::new(source))
    }

    /// Box a borrowed transform or sink.
    pub fn element_ref(element: &'a mut (dyn DynAsyncElement + 'a)) -> Self {
        GraphNodeRef::Element(Box::new(element))
    }

    /// Box a borrowed fan-in muxer.
    pub fn muxer_ref(muxer: &'a mut (dyn DynMultiInputElement + 'a)) -> Self {
        GraphNodeRef::Muxer(Box::new(muxer))
    }

    /// Box a borrowed terminal fan-out source.
    pub fn fanout_source_ref(source: &'a mut (dyn DynMultiOutputSource + 'a)) -> Self {
        GraphNodeRef::FanoutSource(Box::new(source))
    }

    /// Box an owned fan-out demultiplexer (`add_demux`).
    pub fn demux<D: MultiOutputElement + 'static>(demux: D) -> Self {
        GraphNodeRef::Demux(Box::new(demux))
    }

    /// Box a borrowed fan-out demultiplexer.
    pub fn demux_ref(demux: &'a mut (dyn DynMultiOutputElement + 'a)) -> Self {
        GraphNodeRef::Demux(Box::new(demux))
    }

    /// The element's log category (M179), its short type name, e.g.
    /// `videotestsrc`. The runner uses it to derive instance names
    /// (`<category>N`); a DOT dump uses it as the node label before the run
    /// assigns the suffixed name. Fan-in / fan-out elements don't expose a
    /// category on their dyn trait (the runner doesn't name them either), so
    /// they report their structural role.
    pub fn log_category(&self) -> &'static str {
        match self {
            GraphNodeRef::Source(s) => s.log_category(),
            GraphNodeRef::Element(e) => e.log_category(),
            GraphNodeRef::Muxer(_) => "mux",
            GraphNodeRef::FanoutSource(_) => "session-src",
            GraphNodeRef::Demux(_) => "demux",
        }
    }

    /// The memory domain of the frames this node emits on its output pad(s)
    /// (M285): the source's / element's `output_memory`, surfaced per edge for
    /// the DOT dump so a GPU / zero-copy link is marked. Fan-in / fan-out
    /// elements are reported as `System` (their domain is the upstream's; the
    /// per-edge derivation does not propagate through them yet).
    pub fn output_memory(&self) -> crate::memory::MemoryDomainKind {
        match self {
            GraphNodeRef::Source(s) => s.output_memory(),
            GraphNodeRef::Element(e) => e.output_memory(),
            GraphNodeRef::Muxer(_) | GraphNodeRef::FanoutSource(_) | GraphNodeRef::Demux(_) => {
                crate::memory::MemoryDomainKind::System
            }
        }
    }

    /// The full set of memory domains this node can emit (M351), the
    /// producer-capability half of the two-sided allocation-domain negotiation.
    /// A source's / element's `output_domains`; fan-in / fan-out nodes report a
    /// System singleton (their domain follows the upstream, like
    /// [`output_memory`](Self::output_memory)).
    pub fn output_domains(&self) -> crate::memory::DomainSet {
        match self {
            GraphNodeRef::Source(s) => s.output_domains(),
            GraphNodeRef::Element(e) => e.output_domains(),
            GraphNodeRef::Muxer(_) | GraphNodeRef::FanoutSource(_) | GraphNodeRef::Demux(_) => {
                crate::memory::DomainSet::only(crate::memory::MemoryDomainKind::System)
            }
        }
    }

    /// The runtime properties this node's element declares, whatever its shape.
    /// The animated-property check (M882) reads it to validate a
    /// [`ControlProgram`](crate::ControlProgram) against the element it targets.
    pub fn properties(&self) -> &'static [PropertySpec] {
        match self {
            GraphNodeRef::Source(s) => s.properties(),
            GraphNodeRef::Element(e) => e.properties(),
            GraphNodeRef::Muxer(m) => m.properties(),
            GraphNodeRef::FanoutSource(s) => s.properties(),
            GraphNodeRef::Demux(d) => d.properties(),
        }
    }

    /// The memory domains this node accepts on its input pad(s) (M354), for the
    /// converter auto-plug and the allocation cascade. A muxer declares one set
    /// covering every input pad; a terminal fan-out source has no input, so it
    /// reports [`DomainSet::ALL`] (no requirement), as does a source.
    pub fn input_domains(&self) -> crate::memory::DomainSet {
        match self {
            GraphNodeRef::Element(e) => e.input_domains(),
            GraphNodeRef::Muxer(m) => m.input_domains(),
            GraphNodeRef::Demux(d) => d.input_domains(),
            GraphNodeRef::Source(_) | GraphNodeRef::FanoutSource(_) => {
                crate::memory::DomainSet::ALL
            }
        }
    }
}

impl core::fmt::Debug for GraphNodeRef<'_> {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        match self {
            GraphNodeRef::Source(_) => f.write_str("GraphNodeRef::Source(..)"),
            GraphNodeRef::Element(_) => f.write_str("GraphNodeRef::Element(..)"),
            GraphNodeRef::Muxer(_) => f.write_str("GraphNodeRef::Muxer(..)"),
            GraphNodeRef::FanoutSource(_) => f.write_str("GraphNodeRef::FanoutSource(..)"),
            GraphNodeRef::Demux(_) => f.write_str("GraphNodeRef::Demux(..)"),
        }
    }
}

/// Dyn-safe mirror of [`MultiOutputElement`] for a fan-out demux node in the DAG
/// runner, the transpose of [`DynMultiInputElement`]. Boxes `process`'s future
/// and forwards the `Self: Sized` constraint methods. Only the methods the
/// runner uses are mirrored.
pub trait DynMultiOutputElement: ElementBound {
    fn caps_constraint_as_input(&self) -> CapsConstraint<'_>;
    /// Dyn-safe mirror of [`MultiOutputElement::input_domains`]. Default
    /// [`DomainSet::ALL`].
    fn input_domains(&self) -> DomainSet {
        DomainSet::ALL
    }
    fn port_output_caps(&self, port: usize) -> Option<Caps>;
    fn configure_pipeline(&mut self, absolute_caps: &Caps) -> Result<ConfigureOutcome, G2gError>;
    fn process<'a>(
        &'a mut self,
        packet: PipelinePacket,
        out: &'a mut dyn MultiOutputSink,
    ) -> BoxFuture<'a, Result<(), G2gError>>;
    fn properties(&self) -> &'static [PropertySpec];
    fn set_property(&mut self, name: &str, value: PropValue) -> Result<(), PropError>;
    fn get_property(&self, name: &str) -> Option<PropValue>;

    /// Dyn-safe mirror of [`MultiOutputElement::set_instance_name`], so the
    /// runner can name an erased demux instance for logging.
    fn set_instance_name(&mut self, _name: alloc::string::String) {}

    /// Dyn-safe mirror of [`MultiOutputElement::set_log_category`].
    fn set_log_category(&mut self, _category: alloc::string::String) {}

    /// Consume this element into its graph-runner demux arm (M1009), the
    /// fan-out analog of
    /// [`DynAsyncElement::drive_transform_arm`](crate::element::DynAsyncElement::drive_transform_arm).
    /// The blanket impl monomorphizes the arm over the concrete element type,
    /// so the per-packet `process` future is unboxed. Implementations outside
    /// the blanket cannot build the runner's [`DemuxArmIo`]; implement
    /// [`MultiOutputElement`] instead.
    #[doc(hidden)]
    fn drive_demux_arm<'s>(self: Box<Self>, io: DemuxArmIo) -> BoxFuture<'s, Result<u64, G2gError>>
    where
        Self: 's;
}

impl<T: MultiOutputElement> DynMultiOutputElement for T {
    fn caps_constraint_as_input(&self) -> CapsConstraint<'_> {
        MultiOutputElement::caps_constraint_as_input(self)
    }

    fn input_domains(&self) -> DomainSet {
        MultiOutputElement::input_domains(self)
    }

    fn port_output_caps(&self, port: usize) -> Option<Caps> {
        MultiOutputElement::port_output_caps(self, port)
    }

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

    fn process<'a>(
        &'a mut self,
        packet: PipelinePacket,
        out: &'a mut dyn MultiOutputSink,
    ) -> BoxFuture<'a, Result<(), G2gError>> {
        Box::pin(MultiOutputElement::process(self, packet, out))
    }

    fn properties(&self) -> &'static [PropertySpec] {
        MultiOutputElement::properties(self)
    }

    fn set_property(&mut self, name: &str, value: PropValue) -> Result<(), PropError> {
        MultiOutputElement::set_property(self, name, value)
    }

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

    fn set_instance_name(&mut self, name: alloc::string::String) {
        MultiOutputElement::set_instance_name(self, name)
    }

    fn set_log_category(&mut self, category: alloc::string::String) {
        MultiOutputElement::set_log_category(self, category)
    }

    fn drive_demux_arm<'s>(self: Box<Self>, io: DemuxArmIo) -> BoxFuture<'s, Result<u64, G2gError>>
    where
        Self: 's,
    {
        Box::pin(demux_arm(*self, io))
    }
}

/// Private [`MultiOutputElement`] face over an erased demux, so the generic
/// (monomorphized) arm can drive a `&mut dyn DynMultiOutputElement` graph node
/// too. Its per-packet process future stays boxed (the element underneath is
/// erased). The `DynRef` shape, for the fan-out trait.
struct DemuxRef<'b>(&'b mut (dyn DynMultiOutputElement + 'b));

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

    /// Only reachable by a direct call: the arm drives `process`, and
    /// `caps_constraint_as_input` below forwards the erased element's own
    /// constraint rather than routing through here.
    fn intercept_caps(&self, upstream_caps: &Caps) -> Result<Caps, G2gError> {
        let upstream = CapsConstraint::LegacySource(upstream_caps.clone());
        let own = self.0.caps_constraint_as_input();
        solve_linear(&[&upstream, &own])
            .map_err(|_| G2gError::CapsMismatch)?
            .last()
            .cloned()
            .ok_or(G2gError::CapsMismatch)
    }

    fn caps_constraint_as_input(&self) -> CapsConstraint<'_> {
        self.0.caps_constraint_as_input()
    }

    fn configure_pipeline(&mut self, absolute_caps: &Caps) -> Result<ConfigureOutcome, G2gError> {
        self.0.configure_pipeline(absolute_caps)
    }

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

    fn input_domains(&self) -> DomainSet {
        self.0.input_domains()
    }

    fn port_output_caps(&self, port: usize) -> Option<Caps> {
        self.0.port_output_caps(port)
    }

    fn properties(&self) -> &'static [PropertySpec] {
        self.0.properties()
    }

    fn set_property(&mut self, name: &str, value: PropValue) -> Result<(), PropError> {
        self.0.set_property(name, value)
    }

    fn get_property(&self, name: &str) -> Option<PropValue> {
        self.0.get_property(name)
    }

    fn set_instance_name(&mut self, name: alloc::string::String) {
        self.0.set_instance_name(name)
    }

    fn set_log_category(&mut self, category: alloc::string::String) {
        self.0.set_log_category(category)
    }
}

/// Forwarding impl so a borrowed `&mut dyn DynMultiOutputElement` can be boxed
/// into a `Box<dyn DynMultiOutputElement + 'a>` graph node (the borrowing-graph
/// convenience wrappers). Disjoint from the `MultiOutputElement` blanket above.
impl<'b> DynMultiOutputElement for &'b mut (dyn DynMultiOutputElement + 'b) {
    fn caps_constraint_as_input(&self) -> CapsConstraint<'_> {
        (**self).caps_constraint_as_input()
    }

    fn input_domains(&self) -> DomainSet {
        (**self).input_domains()
    }

    fn port_output_caps(&self, port: usize) -> Option<Caps> {
        (**self).port_output_caps(port)
    }

    fn configure_pipeline(&mut self, absolute_caps: &Caps) -> Result<ConfigureOutcome, G2gError> {
        (**self).configure_pipeline(absolute_caps)
    }

    fn process<'a>(
        &'a mut self,
        packet: PipelinePacket,
        out: &'a mut dyn MultiOutputSink,
    ) -> BoxFuture<'a, Result<(), G2gError>> {
        (**self).process(packet, out)
    }

    fn properties(&self) -> &'static [PropertySpec] {
        (**self).properties()
    }

    fn set_property(&mut self, name: &str, value: PropValue) -> Result<(), PropError> {
        (**self).set_property(name, value)
    }

    fn get_property(&self, name: &str) -> Option<PropValue> {
        (**self).get_property(name)
    }

    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)
    }

    fn drive_demux_arm<'s>(self: Box<Self>, io: DemuxArmIo) -> BoxFuture<'s, Result<u64, G2gError>>
    where
        Self: 's,
    {
        Box::pin(demux_arm(DemuxRef(*self), io))
    }
}

/// A β allocation re-cascade report from an arm to the [`GraphCoordinator`].
/// A sink reports the proposal it re-derived on a mid-stream `CapsChanged`; an
/// interior transform reports the proposal it re-derived after applying an
/// upstream directive; a muxer reports the per-pad demands and the merged-output
/// pool it re-derived while walking a change through itself.
#[derive(Debug, Clone)]
struct Recascade {
    node: NodeId,
    route: RecascadeRoute,
    proposal: Option<AllocationParams>,
}

/// Where the coordinator sends a [`Recascade`]'s proposal.
#[derive(Debug, Clone, Copy)]
enum RecascadeRoute {
    /// One hop upstream through the graph topology, to every arm feeding the
    /// reporting node (the ordinary single-input transform / sink path).
    Upstream,
    /// One hop upstream to exactly this arm. A muxer has many inputs, so a change
    /// on one pad names the arm feeding *that* branch rather than re-cascading to
    /// all of them (which the node-keyed `upstream_arms` lookup would do).
    Pad(NodeId),
    /// Downstream to this consumer arm, as an
    /// [`ArmDirective::ProducerAllocation`]: the pool a muxer settled on for its
    /// merged output, told to the element that reads it (M839).
    Consumer(NodeId),
}

/// Producer end of the graph coordinator's control channel, cloned to each
/// transform and sink arm so it can report a [`Recascade`].
#[derive(Debug, Clone)]
pub(crate) struct GraphCoordHandle {
    tx: Sender<Recascade>,
}

impl GraphCoordHandle {
    async fn report(&self, event: Recascade) {
        let _ = self.tx.send(event).await;
    }
}

/// Node-keyed β coordinator for the DAG (the DAG analog of the linear
/// [`Coordinator`](crate::runtime::coordinator::Coordinator)). It owns one
/// [`ArmDirective`] sender per interruptible arm (transforms, sinks, muxers),
/// keyed by node id, plus `upstream_arms`: for each reporting node, the nearest
/// arm feeding each of its inputs (resolved through structural tee nodes; a
/// source terminates the walk). On each report it forwards an
/// `ArmDirective::Recascade` one hop upstream to those arms, which re-derive and
/// report again, so the cascade walks the DAG without a global lock. A report's
/// [`RecascadeRoute`] overrides that node-keyed walk: `Pad` names one branch of a
/// muxer, `Consumer` pushes the muxer's settled output pool downstream instead.
/// The walk is reactive and non-blocking (`try_send`), so it never wedges the
/// data plane.
#[derive(Debug)]
struct GraphCoordinator {
    rx: Receiver<Recascade>,
    arm_ctrl: Vec<Option<Sender<ArmDirective>>>,
    upstream_arms: Vec<Vec<NodeId>>,
}

impl GraphCoordinator {
    fn send_to(&self, node: NodeId, directive: ArmDirective) {
        if let Some(ctrl) = &self.arm_ctrl[node.0 as usize] {
            let _ = ctrl.try_send(directive);
        }
    }

    async fn run(self) -> u64 {
        let mut observed = 0u64;
        while let Some(event) = self.rx.recv().await {
            observed += 1;
            if let Some(p) = event.proposal {
                match event.route {
                    RecascadeRoute::Pad(t) => self.send_to(t, ArmDirective::Recascade(p)),
                    RecascadeRoute::Consumer(t) => {
                        self.send_to(t, ArmDirective::ProducerAllocation(p))
                    }
                    RecascadeRoute::Upstream => {
                        for &u in &self.upstream_arms[event.node.0 as usize] {
                            self.send_to(u, ArmDirective::Recascade(p));
                        }
                    }
                }
            }
        }
        observed
    }
}

/// The nearest upstream interruptible arm feeding `edge_id`: a transform is an
/// arm; a muxer is one too (M839: it continues the walk per input pad); a
/// structural tee is skipped to its own single input; a source terminates the β
/// walk (it is not interruptible).
fn nearest_upstream_arm<E>(vg: &ValidatedGraph<E>, edge_id: usize) -> Option<NodeId> {
    let src = vg.edge(edge_id).src.node;
    match vg.kind(src) {
        NodeKind::Transform | NodeKind::Muxer(_) => Some(src),
        NodeKind::Tee(_) => nearest_upstream_arm(vg, vg.in_edges(src)[0]),
        _ => None,
    }
}

/// The arms that read `edge_id`, so a muxer can tell its consumers the pool it
/// settled on for the merged output (M839). A transform / sink is such an arm; a
/// structural tee broadcasts, so every branch is one. The walk stops at another
/// multi-input boundary: the notice concerns one specific input pad, which this
/// downstream-facing route cannot name.
fn consumer_arms<E>(vg: &ValidatedGraph<E>, edge_id: usize, out: &mut Vec<NodeId>) {
    let dst = vg.edge(edge_id).dst.node;
    match vg.kind(dst) {
        NodeKind::Transform | NodeKind::Sink => {
            if !out.contains(&dst) {
                out.push(dst);
            }
        }
        NodeKind::Tee(_) => {
            for &oe in vg.out_edges(dst) {
                consumer_arms(vg, oe, out);
            }
        }
        _ => {}
    }
}

/// Build a muxer arm's β state from the solved graph (M839): one slot per input
/// edge (in `in_edges` order, so it aligns with the arm's `pad_rxs`) carrying that
/// pad's negotiated caps and the arm feeding its branch, plus the arms reading the
/// merged output.
fn mux_beta<E>(
    vg: &ValidatedGraph<E>,
    node: NodeId,
    in_e: &[usize],
    out_e: usize,
    solution: &[Caps],
    coord: GraphCoordHandle,
) -> MuxBeta {
    let pads = in_e
        .iter()
        .map(|&eid| MuxPad {
            pad: vg.edge(eid).dst.index as usize,
            upstream: nearest_upstream_arm(vg, eid),
            caps: solution[eid].clone(),
            alloc: None,
        })
        .collect();
    let mut consumers = Vec::new();
    consumer_arms(vg, out_e, &mut consumers);
    MuxBeta {
        node,
        coord,
        pads,
        consumers,
        out_alloc: None,
        rounds: 0,
    }
}

/// The fan-out policy of the nearest tee on the path from `node` up toward a
/// source, or `None` if `node` is on a single-producer chain. A node behind a
/// tee shares its upstream with sibling branches, so it can't reverse-
/// reconfigure on a rejected mid-stream change: under `FailLoud` it fails the
/// run (the `run_source_fanout` strict default), under `AllowBranchDrop` it
/// drops out. A node on a single-producer chain (`run_linear_chain` /
/// `run_source_transform_sink`) forwards a feasible re-solve and keeps flowing,
/// but a genuinely infeasible one fails loud too: no runtime producer
/// renegotiates its output caps, so there is nothing to reverse-reconfigure into.
fn behind_tee_policy<E>(vg: &ValidatedGraph<E>, node: NodeId) -> Option<FanOutPolicy> {
    let mut cur = node;
    loop {
        let ins = vg.in_edges(cur);
        if ins.is_empty() {
            return None;
        }
        let src = vg.edge(ins[0]).src.node;
        if matches!(vg.kind(src), NodeKind::Tee(_)) {
            return Some(vg.fanout_policy(src));
        }
        cur = src;
    }
}

/// How a branch arm reacts to a mid-stream `CapsChanged` it cannot negotiate,
/// derived from its position ([`behind_tee_policy`]).
#[derive(Clone, Copy, PartialEq, Eq)]
pub(crate) enum BranchMode {
    /// Single-producer chain: a feasible mid-stream re-solve is forwarded and
    /// the chain keeps flowing; a genuinely infeasible one fails the run loud,
    /// since no runtime producer renegotiates its output caps.
    Reconfigure,
    /// Behind a `FailLoud` tee: a rejected change fails the whole run loud.
    FailLoud,
    /// Behind an `AllowBranchDrop` tee: a rejected change drops this branch (its
    /// arm ends Ok) while the siblings keep flowing.
    Drop,
}

fn branch_mode<E>(vg: &ValidatedGraph<E>, node: NodeId) -> BranchMode {
    match behind_tee_policy(vg, node) {
        None => BranchMode::Reconfigure,
        Some(FanOutPolicy::FailLoud) => BranchMode::FailLoud,
        Some(FanOutPolicy::AllowBranchDrop) => BranchMode::Drop,
    }
}

/// Name the caps an arm could not re-solve at runtime, at error level on the
/// caps category (the same channel the negotiation solver narrates a conflict
/// on), so the run's `CapsMismatch` is diagnosable rather than opaque. The
/// common cause is a rate / format the chain has no converter for (e.g. a
/// 44.1 kHz decode reaching a `rate=48000` pin with no `audioresample` to bridge
/// it). A tee branch cannot reverse-reconfigure its shared upstream; a
/// single-producer chain has nothing to reverse-reconfigure into either, since
/// no runtime producer renegotiates its output caps. Either way the refinement
/// has no solution, so the run fails here rather than flowing stale caps.
fn report_runtime_caps_conflict(rejected: &Caps) {
    crate::g2g_error!(
        crate::log::Target::category(crate::log::CAPS_CATEGORY),
        "runtime caps {} cannot be re-solved against the downstream chain (no converter bridges the refinement, and no upstream element renegotiates its output caps)",
        rejected.to_gst_string()
    );
}

/// Resolve every node's animated-property program (M882) against the element it
/// targets, before negotiation and before any frame flows: an unknown property, a
/// non-numeric one, an empty curve, or a node whose arm has no per-frame hook
/// fails the run here rather than animating nothing. Returns the resolved
/// controllers indexed by node id, for the arm loop to take.
fn resolve_controllers(
    vg: &mut ValidatedGraph<GraphNodeRef<'_>>,
    topo: &[NodeId],
) -> Result<Vec<Option<ArmController>>, G2gError> {
    let mut resolved: Vec<Option<ArmController>> = (0..vg.node_count()).map(|_| None).collect();
    for &node in topo {
        let Some(program) = vg.take_node_control(node) else {
            continue;
        };
        if program.is_empty() {
            continue;
        }
        let name = vg.node_name(node).unwrap_or("<unnamed>");
        let kind = vg.kind(node);
        // Only an arm that hands packets to its element one at a time can sample
        // between frames. A source drives itself (it has no per-frame runner
        // hook), and a tee carries no element at all.
        if !matches!(
            kind,
            NodeKind::Transform | NodeKind::Sink | NodeKind::Muxer(_)
        ) {
            crate::g2g_error!(
                crate::log::Target::category(CONTROL_CATEGORY),
                "node {} ({name}) is a {kind:?} and cannot carry animated properties (transform, sink, and fan-in nodes can)",
                node.0
            );
            return Err(G2gError::ControlBinding);
        }
        let specs = vg.element(node).map(|e| e.properties()).unwrap_or(&[]);
        match program.resolve(specs) {
            Ok(controller) => resolved[node.0 as usize] = Some(controller),
            Err(fault) => {
                crate::g2g_error!(
                    crate::log::Target::category(CONTROL_CATEGORY),
                    "node {} ({name}): {fault}",
                    node.0
                );
                return Err(G2gError::ControlBinding);
            }
        }
    }
    Ok(resolved)
}

/// Sample a controlled element's animated properties at `pts_ns` and set them,
/// before the element sees the frame that carries that timestamp (M882). A
/// rejected value fails the run loud, naming the element and the property.
fn apply_control_at<T: ControlTarget + ?Sized>(
    control: Option<&ArmController>,
    target: &mut T,
    pts_ns: u64,
    probe: &Probe,
) -> Result<(), G2gError> {
    let Some(controller) = control else {
        return Ok(());
    };
    controller.apply(target, pts_ns).map_err(|fault| {
        let element = probe.as_deref().map(|p| p.name()).unwrap_or("element");
        crate::g2g_error!(
            crate::log::Target::category(CONTROL_CATEGORY),
            "{element} at pts {pts_ns}: {fault}"
        );
        G2gError::ControlBinding
    })
}

/// [`apply_control_at`] for a packet: a `DataFrame`'s PTS is the sampling time,
/// and a control packet (caps / segment / flush / tick) carries none, so it
/// samples nothing.
fn apply_control<T: ControlTarget + ?Sized>(
    control: Option<&ArmController>,
    target: &mut T,
    packet: &PipelinePacket,
    probe: &Probe,
) -> Result<(), G2gError> {
    match packet {
        PipelinePacket::DataFrame(frame) => {
            apply_control_at(control, target, frame.timing.pts_ns, probe)
        }
        _ => Ok(()),
    }
}

/// A reusable recipe for an owned graph: a builder closure that produces a fresh
/// [`Graph<GraphNode>`](Graph) each time it is [`instantiate`](Self::instantiate)d.
///
/// [`run_graph`] consumes the elements it runs (it `take()`s the boxed payloads
/// out of the graph), so a graph cannot be run twice. Seek-and-replay (re-run
/// from the start after a flushing seek), retry-on-error, and A/B benchmarking
/// all need a *fresh* set of elements per run, because real elements carry state
/// (a decoder's reference frames, a source's file offset) that cannot simply be
/// rewound. A template rebuilds them via the closure rather than cloning, which
/// is cleaner than making `Graph` itself reusable: that would force every element
/// to be `Clone` or re-initialisable in place, a contract the element traits
/// deliberately do not impose.
pub struct GraphTemplate {
    build: Box<dyn Fn() -> Graph<GraphNode> + Send + Sync>,
}

impl GraphTemplate {
    /// Wrap a graph-builder closure. The closure must construct the whole graph
    /// (nodes + links) from scratch on each call, so each instance gets its own
    /// elements.
    pub fn new(build: impl Fn() -> Graph<GraphNode> + Send + Sync + 'static) -> Self {
        Self {
            build: Box::new(build),
        }
    }

    /// Build a fresh runnable graph. Call once per [`run_graph`] invocation.
    pub fn instantiate(&self) -> Graph<GraphNode> {
        (self.build)()
    }
}

impl core::fmt::Debug for GraphTemplate {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("GraphTemplate").finish_non_exhaustive()
    }
}

/// Drive an arbitrary DAG to EOS. Negotiates the whole graph at once, then runs
/// one arm per node over per-edge channels. `link_capacity` accepts a
/// [`LatencyProfile`](crate::runtime::LatencyProfile) or a `usize` depth.
///
/// Reports the M12 stats (latency / clock / allocation) folded over the graph,
/// the same as the linear runners. The graph payload may own its elements
/// ([`GraphNode`]) or borrow them ([`GraphNodeRef<'a>`], what the convenience
/// wrappers build). A non-`System` frame in a tee or a negotiation conflict
/// fails loud.
///
/// When `clock` can sleep on a deadline
/// ([`PipelineClock::as_ticker`](crate::PipelineClock::as_ticker), which every
/// [`AsyncClock`](crate::AsyncClock) answers), each fan-in arm also gets a
/// **deadline tick** (M880): a fan-in element declaring a
/// [`tick_interval_ns`](crate::MultiInputElement::tick_interval_ns) receives
/// [`PipelinePacket::Tick`] on that period even while its inputs are silent, so a
/// compositor can hold its output rate over a stalled pad (zero-order-hold on
/// that pad's last frame) instead of freezing with it. A clock that only tells
/// time, or an element that declares no interval, runs untimed.
pub async fn run_graph<'a, Clk: PipelineClock>(
    graph: Graph<GraphNodeRef<'a>>,
    clock: &Clk,
    link_capacity: impl Into<LinkCapacity>,
) -> Result<RunStats, G2gError> {
    run_graph_inner(
        graph,
        clock,
        link_capacity,
        None,
        None,
        None,
        None,
        None,
        None,
        None,
    )
    .await
}

/// As [`run_graph`], but enforces a memory-domain [`CopyPolicy`](crate::copyplan::CopyPolicy)
/// as a graph-level contract (M617). After negotiation resolves every link's memory
/// domain, the copy plan (`crate::copyplan`) is checked against `policy` *before any
/// frame flows*: a pipeline that must stay zero-copy
/// ([`CopyPolicy::DenyAll`](crate::copyplan::CopyPolicy::DenyAll)) refuses to start
/// with [`G2gError::CopyBudget`] if an accidental host round-trip appears, rather than
/// paying it at runtime. This turns "is this pipeline zero-copy?" from a question
/// measured after the fact into a guarantee checked at construction. Use
/// [`copy_plan`] on the same graph for the offending-transfer detail.
pub async fn run_graph_with_copy_policy<'a, Clk: PipelineClock>(
    graph: Graph<GraphNodeRef<'a>>,
    clock: &Clk,
    link_capacity: impl Into<LinkCapacity>,
    policy: crate::copyplan::CopyPolicy,
) -> Result<RunStats, G2gError> {
    run_graph_inner(
        graph,
        clock,
        link_capacity,
        None,
        None,
        None,
        Some(policy),
        None,
        None,
        None,
    )
    .await
}

/// Splice memory-domain converters where a producer and consumer cannot agree on
/// a domain (M354), the structural complement to the M351/M352 in-band domain
/// negotiation: negotiation settles a *shared* domain when one exists, and this
/// inserts a converter when one does not. For each original edge `P -> C`, the
/// producer domain is traced through structural tee/demux nodes back to the real
/// producer ([`output_domains`](GraphNodeRef::output_domains)); if it shares no
/// domain with `C`'s [`input_domains`](GraphNodeRef::input_domains), `factory` is
/// asked for a converter from the producer's preferred domain to one `C` accepts,
/// and it is spliced onto that edge ([`Graph::insert_on_edge`]).
///
/// Converters are caps-transparent (`Identity`), so the subsequent caps solve is
/// unaffected. `factory` returns `None` when it has no converter for a pair, in
/// which case the edge is left as-is and the conflict surfaces later as an
/// [`AllocationConflict`](G2gError::AllocationConflict) (the loud failure M351
/// already gives). The converter elements live in `g2g-plugins`, so the caller
/// supplies the factory (the crate layering keeps `g2g-core` converter-agnostic);
/// `g2g-plugins` provides the CUDA wrapper.
pub fn auto_plug_domain_converters<'a>(
    mut graph: Graph<GraphNodeRef<'a>>,
    factory: &dyn Fn(MemoryDomainKind, MemoryDomainKind) -> Option<GraphNodeRef<'a>>,
) -> Graph<GraphNodeRef<'a>> {
    // Snapshot the original edge count: splicing appends nodes and a `K -> C`
    // edge and only rewires the edge being spliced, so original ids stay valid.
    let original_edges = graph.edges().len();
    for e in 0..original_edges {
        let edge = graph.edges()[e];
        let producer = traced_output_domains(&graph, edge.src.node);
        let consumer = graph
            .element(edge.dst.node)
            .map(|n| n.input_domains())
            .unwrap_or(DomainSet::ALL);
        let (Some(from), Some(to)) = (producer.preferred(), consumer.preferred()) else {
            continue;
        };
        // Nothing to do when the two ends already agree on a domain, unless the
        // only thing they agree on is system memory while both would rather stay
        // on the GPU.
        if !producer.intersect(consumer).is_empty() && !bridges_two_gpus(from, to) {
            continue;
        }
        if let Some(conv) = factory(from, to) {
            graph.insert_on_edge(e, conv);
        }
    }
    graph
}

/// Whether a converter beats the domain the two ends would otherwise agree on
/// (M1017): both prefer to keep the frame on the GPU, but in different GPU
/// domains, so the only domain they share is system memory. Bridging the two GPU
/// domains costs a device-to-device copy where agreeing on system memory costs a
/// download and an upload, so the bridge wins whenever the factory has one (a
/// CUDA decoder feeding a wgpu display sink is the case this exists for). When
/// either end prefers system memory, the shared domain is what it asked for and
/// nothing is spliced.
fn bridges_two_gpus(from: MemoryDomainKind, to: MemoryDomainKind) -> bool {
    from != to && !from.is_system() && !to.is_system()
}

/// Domains the node emits, tracing through structural tee/demux nodes (which
/// forward their single input's domain) to the real producer. Used by the
/// converter auto-plug so a tee fed by a GPU decoder reports the GPU domain on
/// every branch rather than the structural `System` default.
fn traced_output_domains<'a>(graph: &Graph<GraphNodeRef<'a>>, node: NodeId) -> DomainSet {
    if let Some(NodeKind::Tee(_)) = graph.node_kind(node) {
        // A tee/demux is domain-transparent: trace its single input's producer.
        return match graph.edges().iter().find(|e| e.dst.node == node) {
            Some(in_edge) => traced_output_domains(graph, in_edge.src.node),
            None => DomainSet::only(MemoryDomainKind::System),
        };
    }
    graph
        .element(node)
        .map(|n| n.output_domains())
        .unwrap_or(DomainSet::only(MemoryDomainKind::System))
}

/// As [`run_graph`], but posts pipeline [`BusMessage`](crate::BusMessage)s to
/// `bus`: a startup `NegotiationFailed`, per transform / sink a `Buffering`
/// level report each time that element's input link crosses a fill quartile
/// (M87, interior links M843), and a `DurationChanged` when a source first
/// reports its duration (M203), so the app can show a buffering indicator, wait
/// for a full buffer, or size a seek bar.
pub async fn run_graph_with_bus<'a, Clk: PipelineClock>(
    graph: Graph<GraphNodeRef<'a>>,
    clock: &Clk,
    link_capacity: impl Into<LinkCapacity>,
    bus: &BusHandle,
) -> Result<RunStats, G2gError> {
    run_graph_inner(
        graph,
        clock,
        link_capacity,
        Some(bus),
        None,
        None,
        None,
        None,
        None,
        None,
    )
    .await
}

/// As [`run_graph`], but taps live telemetry into `observer` and (optionally)
/// posts events to `bus`, the pairing a dev dashboard consumes: the observer
/// carries the graph topology plus per-element `process()` latency / input-link
/// fill, readable mid-run via [`Observer::snapshot`](crate::runtime::Observer::snapshot)
/// from a concurrent task, while the bus carries the out-of-band events (caps
/// changes surface as `Info`/`NegotiationFailed`, plus `Buffering` / `Qos` /
/// `Eos` / `Error`). Pass `bus: None` for telemetry only.
pub async fn run_graph_observed<'a, Clk: PipelineClock>(
    graph: Graph<GraphNodeRef<'a>>,
    clock: &Clk,
    link_capacity: impl Into<LinkCapacity>,
    observer: &Observer,
    bus: Option<&BusHandle>,
) -> Result<RunStats, G2gError> {
    run_graph_inner(
        graph,
        clock,
        link_capacity,
        bus,
        None,
        None,
        None,
        Some(observer),
        None,
        None,
    )
    .await
}

/// As [`run_graph`], but publishes playback progress into `progress` (M203): the
/// sink arm publishes the stream-time [`position`](PipelineProgress::position) of
/// every buffer it consumes, and the source arm publishes the
/// [`duration`](PipelineProgress::duration) its source reports
/// ([`SourceLoop::query_duration`]). The application polls the handle while the
/// pipeline runs, the `POSITION` / `DURATION` query analog.
///
/// `bus` carries the out-of-band messages of [`run_graph_with_bus`] (the
/// matching `DurationChanged` push notification, and the `ElementError` naming
/// what ended the run); pass `None` for progress alone. A caller that reacts to a
/// failure by re-plugging the graph needs both, since the decision is "which
/// element failed" plus "had anything been presented yet".
pub async fn run_graph_with_progress<'a, Clk: PipelineClock>(
    graph: Graph<GraphNodeRef<'a>>,
    clock: &Clk,
    link_capacity: impl Into<LinkCapacity>,
    progress: &PipelineProgress,
    bus: Option<&BusHandle>,
) -> Result<RunStats, G2gError> {
    run_graph_inner(
        graph,
        clock,
        link_capacity,
        bus,
        None,
        Some(progress),
        None,
        None,
        None,
        None,
    )
    .await
}

/// Coarsen a link fill percent into a 0..=4 quartile band, so an arm posts a
/// `Buffering` message on a meaningful level transition (underrun, quarter
/// steps, full) rather than on every packet.
fn buffering_bucket(percent: u8) -> u8 {
    (percent / 25).min(4)
}

/// Sample an arm's input-link fill and post a `Buffering` level when it crosses
/// a quartile band, naming the element the link feeds so an application can tell
/// an interior link from the one feeding a sink. `last` carries the band across
/// loop iterations.
fn report_buffering(
    bus: Option<&BusHandle>,
    probe: &Probe,
    in_rx: &LinkReceiver,
    last: &mut Option<u8>,
) {
    let Some(b) = bus else { return };
    let pct = in_rx.fill_percent();
    let bucket = buffering_bucket(pct);
    if *last != Some(bucket) {
        *last = Some(bucket);
        b.try_post(BusMessage::Buffering {
            percent: pct,
            element: probe
                .as_deref()
                .map(|p| alloc::string::String::from(p.name())),
        });
    }
}

/// As [`run_graph`], but driven by a [`StateController`] (M78): every `Sink`
/// arm gates on the controller, so the whole DAG (linear / fan-out / fan-in /
/// diamond) honors `NULL → READY → PAUSED → PLAYING`. Preroll aggregates: the
/// async `Paused` transition completes only when *all* sinks have prerolled
/// (the runner calls [`StateController::expect_prerolls`] with the sink count).
pub async fn run_graph_stateful<'a, Clk: PipelineClock>(
    graph: Graph<GraphNodeRef<'a>>,
    clock: &Clk,
    link_capacity: impl Into<LinkCapacity>,
    state: &StateController,
) -> Result<RunStats, G2gError> {
    run_graph_inner(
        graph,
        clock,
        link_capacity,
        None,
        Some(state.clone()),
        None,
        None,
        None,
        None,
        None,
    )
    .await
}

/// As [`run_graph_with_progress`], but with a [`FlightRecorder`] attached
/// (M1016): every edge keeps a bounded ring of its most recent packets while the
/// run goes, and on failure the caller writes them out with
/// [`FlightRecorder::dump_to_dir`] as one replayable recording per edge. An hour
/// of live streaming that ends in an error hands back the last moments of
/// traffic instead of nothing.
///
/// `progress` is optional (the recorder does not need it); pass `None` for the
/// plain [`run_graph`] behavior plus recording.
pub async fn run_graph_recorded<'a, Clk: PipelineClock>(
    graph: Graph<GraphNodeRef<'a>>,
    clock: &Clk,
    link_capacity: impl Into<LinkCapacity>,
    progress: Option<&PipelineProgress>,
    bus: Option<&BusHandle>,
    recorder: &FlightRecorder,
) -> Result<RunStats, G2gError> {
    run_graph_inner(
        graph,
        clock,
        link_capacity,
        bus,
        None,
        progress,
        None,
        None,
        Some(recorder),
        None,
    )
    .await
}

/// [`run_graph_observed`] plus the [`run_graph_recorded`] flight recorder, for a
/// tool that watches a run live and still wants the last packets when it fails.
pub async fn run_graph_observed_recorded<'a, Clk: PipelineClock>(
    graph: Graph<GraphNodeRef<'a>>,
    clock: &Clk,
    link_capacity: impl Into<LinkCapacity>,
    observer: &Observer,
    bus: Option<&BusHandle>,
    recorder: &FlightRecorder,
) -> Result<RunStats, G2gError> {
    run_graph_inner(
        graph,
        clock,
        link_capacity,
        bus,
        None,
        None,
        None,
        Some(observer),
        Some(recorder),
        None,
    )
    .await
}

/// As [`run_graph`], but posts a structured
/// [`BusMessage::NegotiationFailed`](crate::BusMessage::NegotiationFailed) to
/// `bus` on a startup negotiation failure. The convenience wrappers' `_with_bus`
/// variants build a borrowing `Graph` and route through here; `run_graph` passes
/// `None`.
/// Result of negotiating and configuring a graph, shared by the cooperative and
/// thread-per-arm drivers: the DAG-wide M12 folds (latency / clock / allocation)
/// plus the solved per-edge caps, computed once before the arms take the elements.
#[derive(Debug)]
struct Prepared {
    solution: Vec<Caps>,
    feasibility: Vec<Option<CapsSet>>,
    latency: LatencyReport,
    allocation: Option<AllocationParams>,
    clock_priority: ClockPriority,
    base_time_ns: u64,
    /// Every clock the elements offered, kept so the health monitor can re-elect
    /// over the survivors. Empty when no monitor runs.
    clock_candidates: Vec<ClockCandidate>,
    /// The swappable handle the sinks' [`ClockSync`] points at, so a re-election
    /// retargets them. `Some` exactly when the monitor runs.
    elected_clock: Option<alloc::sync::Arc<ElectedClock>>,
    /// Every node's instance name, indexed by `NodeId` (empty for a structural
    /// tee, which carries no element). The arm loop copies these into arm order
    /// so a failed run can name the element that raised the error.
    names: Vec<alloc::string::String>,
}

/// Phases 1-3.5 of the graph runner: name instances + mint probes, probe source
/// caps, solve + configure the whole DAG, run the allocation cascade, and elect
/// the clock. Mutates `vg` (configure / clock-sync / instance names) and returns
/// the per-node probes plus the folded [`Prepared`] stats. Shared verbatim by
/// [`run_graph_inner`] (cooperative) and [`run_graph_threaded`] (thread-per-arm)
/// so both negotiate identically and differ only in how they run the arms.
async fn prepare_graph<'a>(
    vg: &mut ValidatedGraph<GraphNodeRef<'a>>,
    topo: &[NodeId],
    state: &Option<StateController>,
    bus: Option<&BusHandle>,
    clock: &dyn PipelineClock,
    observer: Option<&Observer>,
    clock_monitor: bool,
) -> Result<(Vec<Probe>, Prepared), G2gError> {
    let n = vg.node_count();
    // M78: tell the controller how many sinks must preroll before the async
    // `Paused` transition completes (aggregated `AsyncDone`).
    if let Some(sc) = state {
        let sinks = topo
            .iter()
            .filter(|&&n| matches!(vg.kind(n), NodeKind::Sink))
            .count();
        sc.expect_prerolls(sinks);
    }

    // M179: give every payload instance (source, transform, sink, muxer, demux,
    // session) a log name and log its addition: the launch line's `name=` when it
    // set one (M842), else `<category>N` per category (the GStreamer
    // `videotestsrc0` convention). Done before negotiation
    // so an element's own log lines (e.g. at `configure_pipeline`) already carry
    // the instance name. Naming runs whether or not a sink is installed (it is
    // cheap; the `g2g_info!` is threshold-gated).
    // M399: while naming, mint a measured-latency probe for each interior element
    // (Transform / Sink: the nodes with a `process()`), keyed by its instance name.
    let mut probes: Vec<Probe> = (0..n).map(|_| None).collect();
    // Per-node instance names, captured for the observer tap (empty for unnamed
    // structural tee / muxer nodes). Indexed by `NodeId`, like `probes`.
    let mut names: Vec<alloc::string::String> = alloc::vec![alloc::string::String::new(); n];
    {
        let mut namer = crate::log::InstanceNamer::new();
        for &node in topo {
            // M694: fan-in / fan-out nodes carry a `process()` too, so name and
            // probe them alongside transforms / sinks (a plain broadcast tee has
            // no element and no `process()`, so it stays unnamed / unprobed).
            let category = match vg.kind(node) {
                // A terminal fan-in carries a `Muxer` payload but is a session
                // sink, not a merging mux; name it accordingly. The fan-out
                // source is its receive-side mirror.
                NodeKind::FaninSink(_) => "session",
                NodeKind::FanoutSrc(_) => "session-src",
                _ => match vg.element_mut(node) {
                    Some(GraphNodeRef::Source(src)) => src.log_category(),
                    Some(GraphNodeRef::Element(elem)) => elem.log_category(),
                    Some(GraphNodeRef::Muxer(_)) => "mux",
                    Some(GraphNodeRef::FanoutSource(_)) => "session-src",
                    Some(GraphNodeRef::Demux(_)) => "demux",
                    None => continue, // plain broadcast tee: no element to name
                },
            };
            // M847: a launch line's `log-category=` replaces the type category for
            // this instance's own log lines and their filtering. Applied before
            // naming so the element's first lines already carry it; naming still
            // keys on the type category, so siblings keep counting `<type>N`.
            if let Some(over) = vg.node_log_category(node).map(alloc::string::String::from) {
                match vg.element_mut(node) {
                    Some(GraphNodeRef::Source(src)) => src.set_log_category(over),
                    Some(GraphNodeRef::Element(elem)) => elem.set_log_category(over),
                    Some(GraphNodeRef::Muxer(mux)) => mux.set_log_category(over),
                    Some(GraphNodeRef::FanoutSource(src)) => src.set_log_category(over),
                    Some(GraphNodeRef::Demux(demux)) => demux.set_log_category(over),
                    None => {}
                }
            }
            let name = namer.add(category, vg.node_name(node));
            names[node.0 as usize] = name.clone();
            match vg.element_mut(node) {
                Some(GraphNodeRef::Source(src)) => src.set_instance_name(name.clone()),
                Some(GraphNodeRef::Element(elem)) => elem.set_instance_name(name.clone()),
                Some(GraphNodeRef::Muxer(mux)) => mux.set_instance_name(name.clone()),
                Some(GraphNodeRef::FanoutSource(src)) => src.set_instance_name(name.clone()),
                Some(GraphNodeRef::Demux(demux)) => demux.set_instance_name(name.clone()),
                None => {}
            }
            // Mint a measured-latency probe for every node with a `process()`:
            // transforms, sinks, muxers, and demuxers (a demux is a `Tee`-kind
            // node whose payload is a `Demux` element).
            let has_process = matches!(
                vg.kind(node),
                NodeKind::Transform | NodeKind::Sink | NodeKind::Muxer(_) | NodeKind::FaninSink(_)
            ) || matches!(vg.element(node), Some(GraphNodeRef::Demux(_)));
            if has_process {
                // M851: only an observed run records per-frame journeys; the
                // end-of-run report needs the histograms alone.
                probes[node.0 as usize] = Some(if observer.is_some() {
                    ElementProbe::with_journeys(name)
                } else {
                    ElementProbe::new(name)
                });
            }
        }
    }

    // Dev-tooling tap: hand the observer the topology + a clone of every probe
    // `Arc`, so a concurrent task can read live per-element telemetry while the
    // arms run. No-op (and zero cost) when no observer was supplied.
    if let Some(obs) = observer {
        let roles: Vec<crate::runtime::NodeRole> =
            (0..n).map(|i| vg.kind(NodeId(i as u32)).into()).collect();
        let edges: Vec<crate::runtime::EdgeInfo> = vg
            .edges()
            .iter()
            .map(|e| crate::runtime::EdgeInfo {
                from: e.src.node.0 as usize,
                to: e.dst.node.0 as usize,
                ..Default::default()
            })
            .collect();
        obs.register(names.clone(), roles, probes.clone(), edges);
    }

    // Phase 1: probe each source's produce set (async) into an owned map,
    // releasing the mutable borrow before the constraint phase borrows every
    // node. A source that offers alternatives (a camera's pixel formats) hands
    // over all of them, so the solve picks the one downstream can take.
    let mut source_caps: Vec<Option<CapsSet>> = (0..n).map(|_| None).collect();
    for &node in topo {
        if matches!(vg.kind(node), NodeKind::Source) {
            let GraphNodeRef::Source(src) = vg.element_mut(node).ok_or(G2gError::CapsMismatch)?
            else {
                return Err(G2gError::CapsMismatch);
            };
            source_caps[node.0 as usize] = Some(src.produced_caps().await?);
        }
    }

    // Phase 2: build a per-node constraint and solve the whole DAG, and snapshot
    // each edge's downstream feasibility (D4) for the mid-stream re-solve. The
    // transform/sink constraints borrow their elements immutably (coexisting),
    // so both are computed and the borrows released before configure.
    let (solution, feasibility): (Vec<Caps>, Vec<Option<CapsSet>>) = {
        let constraints = build_node_constraints(vg, &source_caps)?;
        let preferences = build_node_preferences(vg);
        let solution =
            solve_graph_preferred(vg, &constraints, &preferences, &|node| caps_label(vg, node))
                .map_err(|f| {
                    report_nego_failure(bus, f);
                    G2gError::CapsMismatch
                })?;
        let feasibility = graph_downstream_feasibility(vg, &constraints, &solution);
        (solution, feasibility)
    };

    // Phase 3: configure each element with its negotiated caps. Source nodes
    // take their single output edge's caps (no input); transforms and sinks
    // take their input edge's caps.
    for &node in topo {
        match vg.kind(node) {
            NodeKind::Source => {
                let caps = solution[vg.out_edges(node)[0]].clone();
                let GraphNodeRef::Source(src) =
                    vg.element_mut(node).ok_or(G2gError::CapsMismatch)?
                else {
                    return Err(G2gError::CapsMismatch);
                };
                src.configure_pipeline(&caps)?.reject_refixate()?;
            }
            NodeKind::Transform | NodeKind::Sink => {
                let caps = solution[vg.in_edges(node)[0]].clone();
                // A transform also learns its negotiated OUTPUT caps (M185), so a
                // caps-driven transform (videoscale fed by a downstream
                // capsfilter) can take its target from the solve. Sinks have no
                // output edge and skip it.
                let out_caps = vg.out_edges(node).first().map(|&eid| solution[eid].clone());
                let GraphNodeRef::Element(elem) =
                    vg.element_mut(node).ok_or(G2gError::CapsMismatch)?
                else {
                    return Err(G2gError::CapsMismatch);
                };
                elem.configure_pipeline(&caps)?.reject_refixate()?;
                if let Some(out_caps) = out_caps {
                    elem.configure_output(&out_caps)?;
                }
            }
            NodeKind::Tee(_) => {
                // A plain (broadcast) tee carries no element. A demux is a
                // tee-shaped node carrying a `MultiOutputElement`, configured
                // with its single input edge's negotiated caps (the byte stream);
                // each branch retypes later via per-output `CapsChanged`.
                if matches!(vg.element(node), Some(GraphNodeRef::Demux(_))) {
                    let caps = solution[vg.in_edges(node)[0]].clone();
                    let GraphNodeRef::Demux(elem) =
                        vg.element_mut(node).ok_or(G2gError::CapsMismatch)?
                    else {
                        return Err(G2gError::CapsMismatch);
                    };
                    elem.configure_pipeline(&caps)?.reject_refixate()?;
                }
            }
            // A terminal fan-out source has no configure hook: its ports'
            // caps were solved from `output_caps` and each downstream sink
            // configures from its own in-edge.
            NodeKind::FanoutSrc(_) => {}
            NodeKind::Muxer(_) | NodeKind::FaninSink(_) => {
                // Configure each input pad with its in-edge's negotiated caps
                // (a terminal fan-in has no output edge to configure).
                let in_edges: Vec<usize> = vg.in_edges(node).to_vec();
                for &eid in &in_edges {
                    let pad = vg.edge(eid).dst.index as usize;
                    let caps = solution[eid].clone();
                    let GraphNodeRef::Muxer(elem) =
                        vg.element_mut(node).ok_or(G2gError::CapsMismatch)?
                    else {
                        return Err(G2gError::CapsMismatch);
                    };
                    elem.configure_pipeline(pad, &caps)?.reject_refixate()?;
                }
            }
        }
    }

    // Phase 3.5: DAG-wide M12 folds, so the runner reports the same latency /
    // clock / allocation the linear runners do (the convenience wrappers reduce
    // to thin builders over this). Done before Phase 4 takes the elements.
    //
    // Allocation cascade in reverse topo order: each element absorbs the
    // proposal arriving on its output edge(s) (`configure_allocation`), then
    // proposes from its output-link caps; the proposal is stored on its input
    // edge(s) for its upstream to absorb. A tee joins its branch proposals
    // (most-restrictive intersection, loud failure on a domain conflict) onto
    // its single input; a muxer proposes its own per-pad demand onto each input
    // edge (the boundary now crosses at startup). The source's absorbed proposal
    // is the reported `allocation`. For a linear chain this is byte-for-byte the
    // linear runner's sink->source fold.
    let allocation = cascade_allocation(vg, topo, &solution)?;

    // Latency fold + clock election over every element node (tee is structural;
    // a muxer contributes neither, like the fan-in runner).
    let mut latencies: Vec<LatencyReport> = Vec::with_capacity(n);
    let mut clocks: Vec<Option<ClockCandidate>> = Vec::with_capacity(n);
    for &node in topo {
        if let Some(l) = element_latency(vg, node) {
            latencies.push(l);
            clocks.push(element_clock(vg, node));
        }
    }
    let latency = LatencyReport::aggregate(latencies);
    let elected = elect_clock(clocks.iter().cloned());
    let (clock_priority, base_time_ns) = match &elected {
        Some(c) => (c.priority, c.clock.now_ns()),
        None => (ClockPriority::SystemFallback, clock.now_ns()),
    };

    // M1004: when the runner is going to watch the elected clock's health, the
    // sinks read it through a swappable handle instead of directly, so a
    // re-election after a loss retargets them without reaching the elements
    // again (they are already inside their arms by then).
    let elected_clock = elected
        .as_ref()
        .filter(|_| clock_monitor)
        .map(|c| alloc::sync::Arc::new(ElectedClock::new(c.clock.clone())));
    let clock_candidates: Vec<ClockCandidate> = if clock_monitor {
        clocks.into_iter().flatten().collect()
    } else {
        Vec::new()
    };

    // Hand the elected clock + base time to every sink so each presents its
    // frames at their running-time deadline (PTS pacing), the same as the linear
    // runners (M169). Only when a clock was elected; without one the sinks present
    // as fast as backpressure allows. A sink node always holds a
    // `GraphNodeRef::Element` (not a `Source`), so the match below covers them.
    if let Some(c) = &elected {
        let sink_clock: alloc::sync::Arc<dyn PipelineClock + Send + Sync> = match &elected_clock {
            Some(handle) => handle.clone(),
            None => c.clock.clone(),
        };
        // M176: under a state controller, arm one Playing-transition anchor
        // (shared across sinks) so each bases presentation on the play edge,
        // not on startup / its preroll frame; without one, the eager base time
        // stands. Armed once outside the loop; the anchor is cheaply cloned.
        let anchor = state
            .as_ref()
            .map(|sc| sc.arm_play_anchor(sink_clock.clone()));
        for &node in topo {
            if matches!(vg.kind(node), NodeKind::Sink) {
                if let Some(GraphNodeRef::Element(elem)) = vg.element_mut(node) {
                    let sync = match &anchor {
                        Some(a) => {
                            ClockSync::with_play_anchor(sink_clock.clone(), base_time_ns, a.clone())
                        }
                        None => ClockSync::new(sink_clock.clone(), base_time_ns),
                    };
                    elem.set_clock_sync(sync.with_path_latency(latency));
                }
            }
        }
    }

    Ok((
        probes,
        Prepared {
            solution,
            feasibility,
            latency,
            allocation,
            clock_priority,
            base_time_ns,
            clock_candidates,
            elected_clock,
            names,
        },
    ))
}

/// Per-edge bounded channels plus the D4 β re-cascade coordinator, shared by the
/// cooperative and thread-per-arm drivers. Returns the edge sender/receiver
/// slots (taken by the arm loop), the shared leaky-link drop counter, the
/// per-transform `ArmDirective` receivers, and the coordinator + its handle.
struct GraphChannels {
    txs: Vec<Option<LinkSender>>,
    rxs: Vec<Option<LinkReceiver>>,
    dropped: alloc::sync::Arc<spin::Mutex<u64>>,
    arm_ctrl_rx: Vec<Option<Receiver<ArmDirective>>>,
    coord_handle: GraphCoordHandle,
    coordinator: GraphCoordinator,
}

fn build_channels<'a>(
    vg: &ValidatedGraph<GraphNodeRef<'a>>,
    topo: &[NodeId],
    link_capacity: usize,
    instrument: bool,
) -> GraphChannels {
    let n = vg.node_count();
    // Phase 4: one bounded channel per edge, then one arm per node. Each arm
    // takes the senders of its outgoing edges and the receivers of its
    // incoming edges (a tee holds n senders, a sink one receiver, etc.).
    let ne = vg.edge_count();
    let mut txs: Vec<Option<LinkSender>> = Vec::with_capacity(ne);
    let mut rxs: Vec<Option<LinkReceiver>> = Vec::with_capacity(ne);
    // Shared drop counter: leaky links (`LinkPolicy::DropOldest`/`DropNewest`)
    // increment it per dropped frame, so the total surfaces in `RunStats`.
    let dropped = alloc::sync::Arc::new(spin::Mutex::new(0u64));
    for eid in 0..ne {
        // A per-edge depth (a `queue max-size-buffers=N`) overrides the graph-wide
        // default; most edges leave it `None` and take `link_capacity`.
        let cap = vg.edge(eid).capacity.unwrap_or(link_capacity);
        // Transit instrumentation only where it is measured + read: `Block` edges
        // (no drops -> the stamp ring stays aligned) into a transform/sink arm
        // (which pops the stamp). Elsewhere a plain link (zero cost).
        let dst_kind = vg.kind(vg.edge(eid).dst.node);
        let instr_edge = instrument
            && vg.edge(eid).policy == crate::link::LinkPolicy::Block
            && matches!(dst_kind, NodeKind::Transform | NodeKind::Sink);
        let (mut tx, rx) = if instr_edge {
            link_with_transit(cap)
        } else {
            link(cap)
        };
        let policy = vg.edge(eid).policy;
        tx.set_policy(policy);
        if policy != crate::link::LinkPolicy::Block {
            tx.set_drop_counter(dropped.clone());
        }
        // Live per-edge traffic counters, on every edge while an observer is
        // attached (unlike transit, they cost nothing to keep aligned).
        if instrument {
            tx.set_counters(alloc::sync::Arc::new(
                crate::runtime::instrument::EdgeCounters::default(),
            ));
        }
        txs.push(Some(tx));
        rxs.push(Some(rx));
    }

    // D4 β coordinator: one `ArmDirective` channel per interruptible arm, plus
    // the upstream-arm adjacency the coordinator walks. Transforms take an
    // upstream demand; a muxer continues the walk through itself per input pad
    // (M839); a sink is the downstream end of a producer's settled pool. When
    // every such arm finishes (EOS-driven), the report handles drop and the
    // coordinator ends.
    let mut arm_ctrl: Vec<Option<Sender<ArmDirective>>> = (0..n).map(|_| None).collect();
    let mut arm_ctrl_rx: Vec<Option<Receiver<ArmDirective>>> = (0..n).map(|_| None).collect();
    let mut upstream_arms: Vec<Vec<NodeId>> = (0..n).map(|_| Vec::new()).collect();
    for &node in topo {
        if matches!(
            vg.kind(node),
            NodeKind::Transform | NodeKind::Sink | NodeKind::Muxer(_)
        ) {
            let (ctx, crx) = bounded::<ArmDirective>(link_capacity);
            arm_ctrl[node.0 as usize] = Some(ctx);
            arm_ctrl_rx[node.0 as usize] = Some(crx);
        }
    }
    // β reporters are transforms (after a directive) and sinks (on caps change);
    // each forwards to the nearest interior arm feeding its inputs.
    for &node in topo {
        if matches!(vg.kind(node), NodeKind::Transform | NodeKind::Sink) {
            let mut ups: Vec<NodeId> = Vec::new();
            for &ie in vg.in_edges(node) {
                if let Some(u) = nearest_upstream_arm(vg, ie) {
                    if !ups.contains(&u) {
                        ups.push(u);
                    }
                }
            }
            upstream_arms[node.0 as usize] = ups;
        }
    }
    let (coord_tx, coord_rx) = bounded::<Recascade>(link_capacity);
    let coord_handle = GraphCoordHandle { tx: coord_tx };
    let coordinator = GraphCoordinator {
        rx: coord_rx,
        arm_ctrl,
        upstream_arms,
    };

    GraphChannels {
        txs,
        rxs,
        dropped,
        arm_ctrl_rx,
        coord_handle,
        coordinator,
    }
}

/// Start the flight recorder on every edge, through the same per-edge
/// content-inspection slot an observer's preview tap uses. Called once the
/// channels are built and before any packet flows, so each edge's ring opens
/// with its negotiated caps and the element names the run assigned. Generic over
/// the node payload because the cooperative runner borrows its elements and the
/// thread-per-arm one owns them, while the edges this reads are the same.
fn install_flight_recorder<E>(
    recorder: &crate::runtime::FlightRecorder,
    vg: &ValidatedGraph<E>,
    txs: &[Option<LinkSender>],
    solution: &[Caps],
    names: &[alloc::string::String],
) {
    for (eid, tx) in txs.iter().enumerate() {
        let (Some(tx), Some(caps)) = (tx.as_ref(), solution.get(eid)) else {
            continue;
        };
        let edge = vg.edge(eid);
        let label = crate::runtime::flight_recorder::edge_label(
            names,
            edge.src.node.0 as usize,
            edge.dst.node.0 as usize,
        );
        recorder.record_edge(label, caps, &tx.probe);
    }
}

/// Hand `obs` the per-edge taps that only exist once the channels are built: the
/// content-inspection slot, the negotiated caps, and the live traffic counters.
/// Aligned with the edge ids registered during `prepare_graph`.
fn register_edge_taps(
    obs: &Observer,
    txs: &[Option<LinkSender>],
    solution: &[Caps],
    link_capacity: usize,
) {
    obs.set_link_capacity(link_capacity);
    let probes: Vec<crate::runtime::channel::ProbeSlot> = txs
        .iter()
        .map(|o| o.as_ref().map(|s| s.probe.clone()).unwrap_or_default())
        .collect();
    let counters: Vec<Option<alloc::sync::Arc<crate::runtime::instrument::EdgeCounters>>> =
        txs.iter().map(|o| o.as_ref()?.counters.clone()).collect();
    obs.register_edges(probes, solution.to_vec(), counters);
}

#[allow(clippy::too_many_arguments)]
pub(crate) async fn run_graph_inner<'a, Clk: PipelineClock>(
    graph: Graph<GraphNodeRef<'a>>,
    clock: &'a Clk,
    link_capacity: impl Into<LinkCapacity>,
    bus: Option<&BusHandle>,
    state: Option<StateController>,
    progress: Option<&PipelineProgress>,
    copy_policy: Option<crate::copyplan::CopyPolicy>,
    observer: Option<&Observer>,
    recorder: Option<&crate::runtime::FlightRecorder>,
    ticker: Option<&'a dyn DynAsyncClock>,
) -> Result<RunStats, G2gError> {
    // A pipeline clock that can sleep on a deadline is the fan-in tick timer
    // (M880), so every entry point ticks without one of its own. An explicit
    // `ticker` still wins.
    let ticker = ticker.or_else(|| clock.as_ticker());
    let link_capacity: usize = link_capacity.into().get();
    let mut vg = graph.finish().map_err(|_| G2gError::CapsMismatch)?;
    let n = vg.node_count();
    if n < 2 {
        return Err(G2gError::CapsMismatch);
    }
    let topo = vg.topo().to_vec();
    // M882: animated properties are checked against their elements before
    // negotiation, so a bad binding fails before configure opens any device.
    let mut controllers = resolve_controllers(&mut vg, &topo)?;

    let (
        probes,
        Prepared {
            solution,
            feasibility,
            latency,
            allocation,
            clock_priority,
            base_time_ns,
            clock_candidates,
            elected_clock,
            names,
        },
    ) = prepare_graph(
        &mut vg,
        &topo,
        &state,
        bus,
        clock,
        observer,
        bus.is_some() && ticker.is_some(),
    )
    .await?;

    // Enforce the memory-domain copy budget (M617) before any frame flows: the graph
    // is negotiated, so the per-edge domains are known and the copy plan is exact. A
    // zero-copy pipeline (`CopyPolicy::DenyAll`) refuses to start rather than silently
    // paying a host round-trip at runtime. Configure side effects already ran (bound
    // sockets etc.) and are released when the doomed graph drops.
    if let Some(policy) = copy_policy {
        let edge_memory: Vec<crate::memory::MemoryDomainKind> = (0..vg.edge_count())
            .map(|id| {
                let src = vg.edge(id).src.node;
                vg.element(src)
                    .map(|e| e.output_memory())
                    .unwrap_or(crate::memory::MemoryDomainKind::System)
            })
            .collect();
        let plan = copy_plan(&vg, &solution, &edge_memory);
        if plan.check(policy).is_err() {
            return Err(G2gError::CopyBudget);
        }
    }

    let GraphChannels {
        mut txs,
        mut rxs,
        dropped,
        mut arm_ctrl_rx,
        coord_handle,
        coordinator,
    } = build_channels(&vg, &topo, link_capacity, observer.is_some());

    // Dev-tooling edge tap: hand the observer each edge's content-inspection slot
    // (shared with the arm's `SenderSink`), its negotiated caps, and its live
    // traffic counters, so a preview subscriber can sample packets crossing any
    // edge and a dashboard can watch its packet / byte / drop totals. No arm
    // changes needed; the slot is empty (pass-through) until a subscriber
    // installs an interceptor.
    if let Some(obs) = observer {
        register_edge_taps(obs, &txs, &solution, link_capacity);
    }
    // Flight recorder: a bounded ring of recent packets per edge, dumped by the
    // caller once the run has failed. Same per-edge slot as the observer's
    // preview tap, so an unrecorded run stays exactly as cheap as before.
    if let Some(rec) = recorder {
        install_flight_recorder(rec, &vg, &txs, &solution, &names);
    }

    let mut arms: Vec<BoxFuture<'a, Result<u64, G2gError>>> = Vec::with_capacity(n + 1);
    let mut arm_kinds: Vec<NodeKind> = Vec::with_capacity(n);
    // Arm order, not node order: a muxer contributes one forwarder arm per
    // input pad plus its own, so an arm index cannot index `names` directly.
    let mut arm_names: Vec<alloc::string::String> = Vec::with_capacity(n + 1);

    for &node in &topo {
        let kind = vg.kind(node);
        let in_e: Vec<usize> = vg.in_edges(node).to_vec();
        let out_e: Vec<usize> = vg.out_edges(node).to_vec();
        let mut in_rxs: Vec<LinkReceiver> = in_e
            .iter()
            .map(|&e| rxs[e].take().expect("edge rx present"))
            .collect();
        let mut out_txs: Vec<LinkSender> = out_e
            .iter()
            .map(|&e| txs[e].take().expect("edge tx present"))
            .collect();
        let element = vg.take_element(node);

        // A muxer contributes N+1 arms (one forwarder per input pad plus the
        // muxer arm), so it is built before the single-arm match below.
        if let NodeKind::Muxer(_) = kind {
            let Some(GraphNodeRef::Muxer(mux)) = element else {
                return Err(G2gError::CapsMismatch);
            };
            let out_tx = out_txs.pop().expect("muxer output edge");
            let mux_out_caps = solution[out_e[0]].clone();
            let pads: Vec<usize> = in_e
                .iter()
                .map(|&eid| vg.edge(eid).dst.index as usize)
                .collect();
            let beta = mux_beta(&vg, node, &in_e, out_e[0], &solution, coord_handle.clone());
            let mux_ctrl = arm_ctrl_rx[node.0 as usize].take().expect("muxer ctrl rx");
            let input_count = in_rxs.len();
            // Each input pad gets its OWN bounded channel feeding the muxer arm,
            // which drains them round-robin. A single shared FIFO would let a
            // fast input (e.g. a free-running background) monopolize the queue
            // and starve a slower real-time input (e.g. a 30 fps camera): the
            // camera's overlay would freeze and, worse, its EOS would never
            // arrive, hanging the all-inputs-EOS aggregation forever. Forwarders
            // are still indexed by pad so a muxer's `process(pad, ..)` keeps its
            // per-input geometry straight even if pads link out of order.
            let mut pad_rxs: Vec<(usize, Receiver<PipelinePacket>)> =
                Vec::with_capacity(input_count);
            for (in_rx, pad) in in_rxs.into_iter().zip(pads) {
                let (pad_tx, pad_rx) = bounded::<PipelinePacket>(link_capacity);
                let fwd: BoxFuture<'a, Result<u64, G2gError>> =
                    Box::pin(muxer_forwarder(in_rx, pad_tx));
                arms.push(fwd);
                arm_kinds.push(kind);
                arm_names.push(names[node.0 as usize].clone());
                pad_rxs.push((pad, pad_rx));
            }
            // A muxer can opt into runner-level PTS-ordered delivery (the runner
            // merges its inputs by DataFrame PTS); the default drains round-robin
            // in arrival order. The hook picks the arm from
            // `input_pts_ordered` and monomorphizes it over the element.
            let mux_probe = probes[node.0 as usize].clone();
            let mux_control = controllers[node.0 as usize].take();
            let arm: BoxFuture<'a, Result<u64, G2gError>> = mux.drive_muxer_arm(MuxerArmIo {
                parts: MuxerArmParts {
                    pad_rxs,
                    out_tx,
                    input_count,
                    current_output: mux_out_caps,
                    beta,
                    arm_rx: mux_ctrl,
                    probe: mux_probe,
                    control: mux_control,
                },
                ticker,
            });
            arms.push(arm);
            arm_kinds.push(kind);
            arm_names.push(names[node.0 as usize].clone());
            continue;
        }

        let arm: BoxFuture<'a, Result<u64, G2gError>> = match kind {
            NodeKind::Source => {
                let Some(GraphNodeRef::Source(src)) = element else {
                    return Err(G2gError::CapsMismatch);
                };
                let out_tx = out_txs.pop().expect("source output edge");
                Box::pin(source_arm(src, out_tx, bus.cloned(), progress.cloned()))
            }
            NodeKind::Transform => {
                let Some(GraphNodeRef::Element(elem)) = element else {
                    return Err(G2gError::CapsMismatch);
                };
                let in_rx = in_rxs.pop().expect("transform input edge");
                let out_tx = out_txs.pop().expect("transform output edge");
                let out_edge = out_e[0];
                let arm_rx = arm_ctrl_rx[node.0 as usize]
                    .take()
                    .expect("transform ctrl rx");
                let out_caps = solution[out_edge].clone();
                let downstream_feasible = feasibility[out_edge].clone();
                elem.drive_transform_arm(TransformArmIo {
                    in_rx,
                    out_tx,
                    arm_rx,
                    coord: coord_handle.clone(),
                    node,
                    out_caps,
                    downstream_feasible,
                    mode: branch_mode(&vg, node),
                    bus: bus.cloned(),
                    probe: probes[node.0 as usize].clone(),
                    control: controllers[node.0 as usize].take(),
                })
            }
            NodeKind::Sink => {
                let Some(GraphNodeRef::Element(elem)) = element else {
                    return Err(G2gError::CapsMismatch);
                };
                let in_rx = in_rxs.pop().expect("sink input edge");
                let arm_rx = arm_ctrl_rx[node.0 as usize].take().expect("sink ctrl rx");
                advertise_orientation(&in_rx, elem.absorbs_orientation());
                elem.drive_sink_arm(SinkArmIo {
                    in_rx,
                    arm_rx,
                    coord: coord_handle.clone(),
                    node,
                    mode: branch_mode(&vg, node),
                    bus: bus.cloned(),
                    state: state.clone(),
                    progress: progress.cloned(),
                    probe: probes[node.0 as usize].clone(),
                    control: controllers[node.0 as usize].take(),
                })
            }
            NodeKind::Tee(_) => {
                let in_rx = in_rxs.pop().expect("tee input edge");
                // A tee-shaped node carrying a demux element routes per-output;
                // a plain tee broadcasts. Under `AllowBranchDrop` the broadcast
                // tolerates a branch that has dropped out (closed its channel).
                let branch_drop = vg.fanout_policy(node) == FanOutPolicy::AllowBranchDrop;
                match element {
                    Some(GraphNodeRef::Demux(demux)) => demux.drive_demux_arm(DemuxArmIo {
                        in_rx,
                        out_txs: demux_out_txs_by_port(&vg, node, out_txs),
                        probe: probes[node.0 as usize].clone(),
                    }),
                    _ => Box::pin(tee_arm(in_rx, out_txs, branch_drop)),
                }
            }
            NodeKind::FaninSink(_) => {
                let Some(GraphNodeRef::Muxer(session)) = element else {
                    return Err(G2gError::CapsMismatch);
                };
                let (pad_rxs, reverse) = fanin_sink_pads(&vg, node, in_rxs, &*session);
                session.drive_fanin_sink_arm(FaninSinkArmIo {
                    pad_rxs,
                    reverse,
                    probe: probes[node.0 as usize].clone(),
                })
            }
            NodeKind::FanoutSrc(_) => {
                let Some(GraphNodeRef::FanoutSource(source)) = element else {
                    return Err(G2gError::CapsMismatch);
                };
                let sinks = fanout_src_ports(&vg, node, out_txs);
                Box::pin(fanout_src_arm(source, sinks))
            }
            NodeKind::Muxer(_) => unreachable!("muxer handled above"),
        };
        arms.push(arm);
        arm_kinds.push(kind);
        arm_names.push(names[node.0 as usize].clone());
    }

    // Drop the template handle so the coordinator can end once every arm's
    // clone drops; append the coordinator as the final arm. Its result is the
    // count of re-cascade events it observed.
    drop(coord_handle);
    let coord_arm_index = arms.len();
    arms.push(Box::pin(async move { Ok(coordinator.run().await) }));

    // M1004: with a swappable elected clock installed, the health monitor runs
    // alongside the arms and is dropped once they finish (it never ends itself).
    let results = match (elected_clock, bus, ticker) {
        (Some(handle), Some(b), Some(t)) => {
            let monitor = clock_health_monitor(clock_candidates, handle, b.clone(), t);
            match select2(join_all(arms), monitor).await {
                Either::Left(results) => results,
                Either::Right(never) => match never {},
            }
        }
        _ => join_all(arms).await,
    };
    fold_run_stats(
        results,
        &arm_kinds,
        &arm_names,
        coord_arm_index,
        &dropped,
        &probes,
        latency,
        allocation,
        clock_priority,
        base_time_ns,
        bus,
    )
}

/// Fold each arm's per-node frame count into the final [`RunStats`], shared by
/// the cooperative ([`run_graph_inner`]) and thread-per-arm
/// ([`run_graph_threaded`]) drivers, which differ only in how the arms are run
/// (one executor vs one OS thread each), not in how their results aggregate.
/// `results` is in arm order (source / transform / sink / muxer arms, then the
/// coordinator last at `coord_arm_index`); `arm_kinds` labels every arm except
/// the coordinator.
#[allow(clippy::too_many_arguments)]
fn fold_run_stats(
    results: Vec<Result<u64, G2gError>>,
    arm_kinds: &[NodeKind],
    arm_names: &[alloc::string::String],
    coord_arm_index: usize,
    dropped: &alloc::sync::Arc<spin::Mutex<u64>>,
    probes: &[Probe],
    latency: LatencyReport,
    allocation: Option<AllocationParams>,
    clock_priority: ClockPriority,
    base_time_ns: u64,
    bus: Option<&BusHandle>,
) -> Result<RunStats, G2gError> {
    // M81: surface a substantive arm error over a secondary `Shutdown` (a real
    // error in one node closes links, which surfaces as `Shutdown` on the
    // others; reporting the first-in-topo-order error would often mask the
    // cause).
    let failed = |only_substantive: bool| {
        results
            .iter()
            .enumerate()
            .filter_map(|(i, r)| r.as_ref().err().map(|e| (i, e)))
            .find(|(_, e)| !only_substantive || **e != G2gError::Shutdown)
    };
    if let Some((arm, e)) = failed(true).or_else(|| failed(false)) {
        let name = arm_names.get(arm).map(|n| n.as_str());
        crate::log::report_element_failure(name, e);
        // Named on the bus too: the returned error carries no element identity,
        // so an application that reacts per element has nothing else to key on.
        if let (Some(bus), Some(name)) = (bus, name.filter(|n| !n.is_empty())) {
            bus.try_post(crate::bus::BusMessage::ElementError {
                element: alloc::string::String::from(name),
                error: e.clone(),
            });
        }
        return Err(e.clone());
    }
    let mut counts = Vec::with_capacity(results.len());
    for r in results {
        counts.push(r?);
    }
    let coordinator_events = counts[coord_arm_index];
    let mut emitted = 0u64;
    let mut consumed = 0u64;
    for (kind, &count) in arm_kinds.iter().zip(counts.iter()) {
        match kind {
            NodeKind::Source | NodeKind::FanoutSrc(_) => emitted += count,
            NodeKind::Sink | NodeKind::FaninSink(_) => consumed += count,
            _ => {}
        }
    }

    let frames_dropped = *dropped.lock();
    // M399: every arm has joined, so its probe is no longer being written; snapshot
    // each interior element's measured latency + fill into the report.
    let per_element = snapshot_all(probes);
    Ok(RunStats {
        frames_emitted: emitted,
        frames_consumed: consumed,
        frames_dropped,
        latency,
        allocation,
        clock_priority,
        base_time_ns,
        coordinator_events,
        per_element,
    })
}

/// A graph arm's future, built and driven entirely on one worker thread. It is
/// deliberately **not** `Send`: the thread-per-arm runner never moves a future
/// between threads (only the element + channels, all `Send`, cross at setup), so
/// an element whose future is `!Send` (a hardware decoder holding a raw context)
/// runs unchanged, exactly as under the cooperative runner.
#[cfg(all(feature = "std", feature = "multi-thread"))]
pub type LocalArmFuture =
    core::pin::Pin<alloc::boxed::Box<dyn core::future::Future<Output = Result<u64, G2gError>>>>;

/// Executor abstraction for [`run_graph_threaded`]. The runner hands each graph
/// node's arm to `spawn_arm` as a `Send` builder closure; the spawner runs the
/// builder on a dedicated worker thread and drives the [`LocalArmFuture`] it
/// returns to completion there, resolving the returned handle with the arm's
/// frame count (or its error). This is the GStreamer streaming-thread model: one
/// OS thread per element, so CPU-bound stages (software decode/encode) overlap
/// across cores instead of serialising on one cooperative executor.
///
/// `g2g-core` stays executor-agnostic; `g2g-plugins` supplies a tokio-backed
/// `ThreadSpawner`. Only the element and its channels (all `Send`) cross the
/// thread boundary; the future stays put, so elements need no `Send` future.
#[cfg(all(feature = "std", feature = "multi-thread"))]
pub trait GraphSpawner {
    /// Run `build` on a worker thread and drive its future there; the returned
    /// handle resolves (on the caller thread) once that arm finishes.
    fn spawn_arm(
        &self,
        build: alloc::boxed::Box<dyn FnOnce() -> LocalArmFuture + Send>,
    ) -> BoxFuture<'static, Result<u64, G2gError>>;
}

/// Identifies the calling OS thread for
/// [`BusMessage::StreamStatus`](crate::BusMessage::StreamStatus). `ThreadId` has
/// no stable numeric form on the MSRV (`as_u64` is unstable), so hash it: only
/// equality between an enter and its leave is meaningful.
#[cfg(all(feature = "std", feature = "multi-thread"))]
fn current_thread_id() -> u64 {
    use core::hash::{Hash, Hasher};
    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    std::thread::current().id().hash(&mut hasher);
    hasher.finish()
}

/// Bracket one arm's builder with its [`BusMessage::StreamStatus`](crate::BusMessage::StreamStatus)
/// enter / leave pair. The builder itself runs on the worker thread, so posting
/// there names the thread the arm will be driven on; the leave rides on the
/// arm's own future so it lands when that arm finishes. Without a bus the
/// builder is returned untouched.
#[cfg(all(feature = "std", feature = "multi-thread"))]
fn with_stream_status(
    bus: Option<BusHandle>,
    build: alloc::boxed::Box<dyn FnOnce() -> LocalArmFuture + Send>,
) -> alloc::boxed::Box<dyn FnOnce() -> LocalArmFuture + Send> {
    let Some(bus) = bus else {
        return build;
    };
    alloc::boxed::Box::new(move || -> LocalArmFuture {
        let thread_id = current_thread_id();
        bus.try_post(BusMessage::StreamStatus {
            entered: true,
            thread_id,
        });
        let arm = build();
        Box::pin(async move {
            let result = arm.await;
            bus.try_post(BusMessage::StreamStatus {
                entered: false,
                thread_id,
            });
            result
        })
    })
}

/// Thread-per-arm sibling of [`run_graph_inner`]: negotiates the graph
/// identically (shared [`prepare_graph`] / [`build_channels`] / [`fold_run_stats`]),
/// then hands each arm to `spawner` to run on its own OS thread rather than
/// cooperatively multiplexing them on the caller's executor. The graph must own
/// its elements (`Graph<GraphNode>`, i.e. `'static`) so each arm can move its
/// element onto a worker thread. All `vg`-borrowing / reference-typed inputs are
/// resolved to owned values *before* each arm's builder closure, since the
/// closure runs on another thread and may not borrow `vg`, `bus`, or `progress`.
/// `ticker` follows that rule too: the fan-in arm's deadline clock is a shared
/// [`Arc`](alloc::sync::Arc), not a borrow, so each muxer builder can carry it
/// onto its thread.
#[cfg(all(feature = "std", feature = "multi-thread"))]
#[allow(clippy::too_many_arguments)]
pub(crate) async fn run_graph_threaded_inner<S: GraphSpawner>(
    graph: Graph<GraphNode>,
    clock: &dyn PipelineClock,
    link_capacity: impl Into<LinkCapacity>,
    bus: Option<&BusHandle>,
    state: Option<StateController>,
    progress: Option<&PipelineProgress>,
    observer: Option<&Observer>,
    recorder: Option<&FlightRecorder>,
    ticker: Option<alloc::sync::Arc<dyn DynAsyncClock + Send + Sync>>,
    spawner: &S,
) -> Result<RunStats, G2gError> {
    // Same rule as the cooperative runner (M880), through the shared handle the
    // arm threads can own: a pipeline clock that can sleep on a deadline is the
    // fan-in tick timer, so every threaded entry point ticks without one of its
    // own. An explicit `ticker` still wins.
    let ticker = ticker.or_else(|| clock.shared_ticker());
    let link_capacity: usize = link_capacity.into().get();
    let mut vg = graph.finish().map_err(|_| G2gError::CapsMismatch)?;
    let n = vg.node_count();
    if n < 2 {
        return Err(G2gError::CapsMismatch);
    }
    let topo = vg.topo().to_vec();
    // M882: same pre-negotiation check as the cooperative runner; a resolved
    // controller is owned data, so it rides each arm's builder closure onto its
    // worker thread like the element itself.
    let mut controllers = resolve_controllers(&mut vg, &topo)?;

    let (
        probes,
        Prepared {
            solution,
            feasibility,
            latency,
            allocation,
            clock_priority,
            base_time_ns,
            clock_candidates,
            elected_clock,
            names,
        },
    ) = prepare_graph(
        &mut vg,
        &topo,
        &state,
        bus,
        clock,
        observer,
        bus.is_some() && ticker.is_some(),
    )
    .await?;

    let GraphChannels {
        mut txs,
        mut rxs,
        dropped,
        mut arm_ctrl_rx,
        coord_handle,
        coordinator,
    } = build_channels(&vg, &topo, link_capacity, observer.is_some());

    // Dev-tooling edge tap: same as the cooperative path.
    if let Some(obs) = observer {
        register_edge_taps(obs, &txs, &solution, link_capacity);
    }
    // Flight recorder: installed here, before the arms move onto their threads,
    // so each ring is shared with the worker that fills it (see the cooperative
    // path for the mechanism).
    if let Some(rec) = recorder {
        install_flight_recorder(rec, &vg, &txs, &solution, &names);
    }

    // One `spawn_arm` handle per arm (mirrors the cooperative `arms` vec). Each
    // handle resolves on this thread once its worker thread finishes.
    let mut handles: Vec<BoxFuture<'static, Result<u64, G2gError>>> = Vec::with_capacity(n + 1);
    let mut arm_kinds: Vec<NodeKind> = Vec::with_capacity(n);
    // Arm order, not node order: a muxer contributes one forwarder arm per
    // input pad plus its own, so an arm index cannot index `names` directly.
    let mut arm_names: Vec<alloc::string::String> = Vec::with_capacity(n + 1);

    for &node in &topo {
        let kind = vg.kind(node);
        let in_e: Vec<usize> = vg.in_edges(node).to_vec();
        let out_e: Vec<usize> = vg.out_edges(node).to_vec();
        let mut in_rxs: Vec<LinkReceiver> = in_e
            .iter()
            .map(|&e| rxs[e].take().expect("edge rx present"))
            .collect();
        let mut out_txs: Vec<LinkSender> = out_e
            .iter()
            .map(|&e| txs[e].take().expect("edge tx present"))
            .collect();
        let element = vg.take_element(node);

        if let NodeKind::Muxer(_) = kind {
            let Some(GraphNodeRef::Muxer(mux)) = element else {
                return Err(G2gError::CapsMismatch);
            };
            let out_tx = out_txs.pop().expect("muxer output edge");
            let mux_out_caps = solution[out_e[0]].clone();
            let pads: Vec<usize> = in_e
                .iter()
                .map(|&eid| vg.edge(eid).dst.index as usize)
                .collect();
            let beta = mux_beta(&vg, node, &in_e, out_e[0], &solution, coord_handle.clone());
            let mux_ctrl = arm_ctrl_rx[node.0 as usize].take().expect("muxer ctrl rx");
            let input_count = in_rxs.len();
            let mut pad_rxs: Vec<(usize, Receiver<PipelinePacket>)> =
                Vec::with_capacity(input_count);
            for (in_rx, pad) in in_rxs.into_iter().zip(pads) {
                let (pad_tx, pad_rx) = bounded::<PipelinePacket>(link_capacity);
                let build: alloc::boxed::Box<dyn FnOnce() -> LocalArmFuture + Send> =
                    alloc::boxed::Box::new(move || -> LocalArmFuture {
                        Box::pin(muxer_forwarder(in_rx, pad_tx))
                    });
                handles.push(spawner.spawn_arm(with_stream_status(bus.cloned(), build)));
                arm_kinds.push(kind);
                arm_names.push(names[node.0 as usize].clone());
                pad_rxs.push((pad, pad_rx));
            }
            let mux_probe = probes[node.0 as usize].clone();
            let mux_ticker = ticker.clone();
            let mux_control = controllers[node.0 as usize].take();
            let build: alloc::boxed::Box<dyn FnOnce() -> LocalArmFuture + Send> =
                alloc::boxed::Box::new(move || -> LocalArmFuture {
                    mux.drive_muxer_arm_owned_tick(MuxerArmOwnedTickIo {
                        parts: MuxerArmParts {
                            pad_rxs,
                            out_tx,
                            input_count,
                            current_output: mux_out_caps,
                            beta,
                            arm_rx: mux_ctrl,
                            probe: mux_probe,
                            control: mux_control,
                        },
                        ticker: mux_ticker,
                    })
                });
            handles.push(spawner.spawn_arm(with_stream_status(bus.cloned(), build)));
            arm_kinds.push(kind);
            arm_names.push(names[node.0 as usize].clone());
            continue;
        }

        let build: alloc::boxed::Box<dyn FnOnce() -> LocalArmFuture + Send> = match kind {
            NodeKind::Source => {
                let Some(GraphNodeRef::Source(src)) = element else {
                    return Err(G2gError::CapsMismatch);
                };
                let out_tx = out_txs.pop().expect("source output edge");
                let bus_c = bus.cloned();
                let prog_c = progress.cloned();
                alloc::boxed::Box::new(move || -> LocalArmFuture {
                    Box::pin(source_arm(src, out_tx, bus_c, prog_c))
                })
            }
            NodeKind::Transform => {
                let Some(GraphNodeRef::Element(elem)) = element else {
                    return Err(G2gError::CapsMismatch);
                };
                let in_rx = in_rxs.pop().expect("transform input edge");
                let out_tx = out_txs.pop().expect("transform output edge");
                let out_edge = out_e[0];
                let arm_rx = arm_ctrl_rx[node.0 as usize]
                    .take()
                    .expect("transform ctrl rx");
                let out_caps = solution[out_edge].clone();
                let downstream_feasible = feasibility[out_edge].clone();
                let bm = branch_mode(&vg, node);
                let bus_c = bus.cloned();
                let probe = probes[node.0 as usize].clone();
                let ch = coord_handle.clone();
                let control = controllers[node.0 as usize].take();
                alloc::boxed::Box::new(move || -> LocalArmFuture {
                    elem.drive_transform_arm(TransformArmIo {
                        in_rx,
                        out_tx,
                        arm_rx,
                        coord: ch,
                        node,
                        out_caps,
                        downstream_feasible,
                        mode: bm,
                        bus: bus_c,
                        probe,
                        control,
                    })
                })
            }
            NodeKind::Sink => {
                let Some(GraphNodeRef::Element(elem)) = element else {
                    return Err(G2gError::CapsMismatch);
                };
                let in_rx = in_rxs.pop().expect("sink input edge");
                advertise_orientation(&in_rx, elem.absorbs_orientation());
                let bm = branch_mode(&vg, node);
                let bus_c = bus.cloned();
                let state_c = state.clone();
                let prog_c = progress.cloned();
                let probe = probes[node.0 as usize].clone();
                let ch = coord_handle.clone();
                let arm_rx = arm_ctrl_rx[node.0 as usize].take().expect("sink ctrl rx");
                let control = controllers[node.0 as usize].take();
                alloc::boxed::Box::new(move || -> LocalArmFuture {
                    elem.drive_sink_arm(SinkArmIo {
                        in_rx,
                        arm_rx,
                        coord: ch,
                        node,
                        mode: bm,
                        bus: bus_c,
                        state: state_c,
                        progress: prog_c,
                        probe,
                        control,
                    })
                })
            }
            NodeKind::Tee(_) => {
                let in_rx = in_rxs.pop().expect("tee input edge");
                let branch_drop = vg.fanout_policy(node) == FanOutPolicy::AllowBranchDrop;
                match element {
                    Some(GraphNodeRef::Demux(demux)) => {
                        let demux_probe = probes[node.0 as usize].clone();
                        let out_txs = demux_out_txs_by_port(&vg, node, out_txs);
                        alloc::boxed::Box::new(move || -> LocalArmFuture {
                            demux.drive_demux_arm(DemuxArmIo {
                                in_rx,
                                out_txs,
                                probe: demux_probe,
                            })
                        })
                    }
                    _ => alloc::boxed::Box::new(move || -> LocalArmFuture {
                        Box::pin(tee_arm(in_rx, out_txs, branch_drop))
                    }),
                }
            }
            NodeKind::FaninSink(_) => {
                let Some(GraphNodeRef::Muxer(session)) = element else {
                    return Err(G2gError::CapsMismatch);
                };
                let (pad_rxs, reverse) = fanin_sink_pads(&vg, node, in_rxs, &*session);
                let probe = probes[node.0 as usize].clone();
                alloc::boxed::Box::new(move || -> LocalArmFuture {
                    session.drive_fanin_sink_arm(FaninSinkArmIo {
                        pad_rxs,
                        reverse,
                        probe,
                    })
                })
            }
            NodeKind::FanoutSrc(_) => {
                let Some(GraphNodeRef::FanoutSource(source)) = element else {
                    return Err(G2gError::CapsMismatch);
                };
                let sinks = fanout_src_ports(&vg, node, out_txs);
                alloc::boxed::Box::new(move || -> LocalArmFuture {
                    Box::pin(fanout_src_arm(source, sinks))
                })
            }
            NodeKind::Muxer(_) => unreachable!("muxer handled above"),
        };
        handles.push(spawner.spawn_arm(with_stream_status(bus.cloned(), build)));
        arm_kinds.push(kind);
        arm_names.push(names[node.0 as usize].clone());
    }

    // Drop the template handle so the coordinator ends once every arm's clone
    // drops; append the coordinator as the final arm on its own thread.
    drop(coord_handle);
    let coord_arm_index = handles.len();
    handles.push(spawner.spawn_arm(with_stream_status(
        bus.cloned(),
        alloc::boxed::Box::new(move || -> LocalArmFuture {
            Box::pin(async move { Ok(coordinator.run().await) })
        }),
    )));

    // M1004: same clock-health watch as the cooperative runner, driven on the
    // caller's executor rather than a worker thread (it only sleeps and reads).
    let results = match (elected_clock, bus, &ticker) {
        (Some(handle), Some(b), Some(t)) => {
            let monitor = clock_health_monitor(clock_candidates, handle, b.clone(), &**t);
            match select2(join_all(handles), monitor).await {
                Either::Left(results) => results,
                Either::Right(never) => match never {},
            }
        }
        _ => join_all(handles).await,
    };
    fold_run_stats(
        results,
        &arm_kinds,
        &arm_names,
        coord_arm_index,
        &dropped,
        &probes,
        latency,
        allocation,
        clock_priority,
        base_time_ns,
        bus,
    )
}

/// Run a DAG with one OS thread per arm via `spawner` (opt-in multicore; the
/// GStreamer streaming-thread model). Cooperative [`run_graph`] stays the default
/// for lowest latency and the `no_std` / wasm executors; this trades a per-stage
/// thread handoff for CPU-bound stages overlapping across cores.
///
/// Fan-in arms tick exactly as they do cooperatively (M953): when `clock` hands
/// out a shared timer ([`PipelineClock::shared_ticker`](crate::PipelineClock::shared_ticker),
/// which the wall clock does), an element declaring a
/// [`tick_interval_ns`](crate::MultiInputElement::tick_interval_ns) receives
/// [`PipelinePacket::Tick`] on that period while its inputs are silent.
#[cfg(all(feature = "std", feature = "multi-thread"))]
pub async fn run_graph_threaded<Clk: PipelineClock, S: GraphSpawner>(
    graph: Graph<GraphNode>,
    clock: &Clk,
    link_capacity: impl Into<LinkCapacity>,
    spawner: &S,
) -> Result<RunStats, G2gError> {
    run_graph_threaded_inner(
        graph,
        clock,
        link_capacity,
        None,
        None,
        None,
        None,
        None,
        None,
        spawner,
    )
    .await
}

/// As [`run_graph_threaded`], but takes the arms' **deadline tick** timer
/// explicitly (M879): `clock` is both the pipeline clock and the timer the arms
/// sleep on, so a fan-in element declaring a
/// [`tick_interval_ns`](crate::MultiInputElement::tick_interval_ns) receives
/// [`PipelinePacket::Tick`] on that period even while its inputs are silent.
///
/// The clock arrives as a shared handle rather than a borrow because each arm's
/// builder closure moves onto its own OS thread: `Arc::new(my_clock)` (any
/// [`AsyncClock`](crate::AsyncClock) that is `Send + Sync`) coerces to it. This is
/// the entry for a clock that cannot hand out such a handle from `&self`, which is
/// what [`run_graph_threaded`] reads
/// ([`PipelineClock::shared_ticker`](crate::PipelineClock::shared_ticker)).
#[cfg(all(feature = "std", feature = "multi-thread"))]
pub async fn run_graph_threaded_ticked<S: GraphSpawner>(
    graph: Graph<GraphNode>,
    clock: alloc::sync::Arc<dyn DynAsyncClock + Send + Sync>,
    link_capacity: impl Into<LinkCapacity>,
    spawner: &S,
) -> Result<RunStats, G2gError> {
    run_graph_threaded_inner(
        graph,
        &clock,
        link_capacity,
        None,
        None,
        None,
        None,
        None,
        Some(clock.clone()),
        spawner,
    )
    .await
}

/// As [`run_graph_threaded`], but posts pipeline messages to `bus` (the
/// thread-per-arm analog of [`run_graph_with_bus`]). This runner adds the
/// [`StreamStatus`](crate::BusMessage::StreamStatus) enter / leave pair per arm
/// thread, which the cooperative runner has no equivalent of.
#[cfg(all(feature = "std", feature = "multi-thread"))]
pub async fn run_graph_threaded_with_bus<Clk: PipelineClock, S: GraphSpawner>(
    graph: Graph<GraphNode>,
    clock: &Clk,
    link_capacity: impl Into<LinkCapacity>,
    bus: &BusHandle,
    spawner: &S,
) -> Result<RunStats, G2gError> {
    run_graph_threaded_inner(
        graph,
        clock,
        link_capacity,
        Some(bus),
        None,
        None,
        None,
        None,
        None,
        spawner,
    )
    .await
}

/// As [`run_graph_threaded`], but publishes playback progress (the thread-per-arm
/// analog of [`run_graph_with_progress`]).
#[cfg(all(feature = "std", feature = "multi-thread"))]
pub async fn run_graph_threaded_with_progress<Clk: PipelineClock, S: GraphSpawner>(
    graph: Graph<GraphNode>,
    clock: &Clk,
    link_capacity: impl Into<LinkCapacity>,
    progress: &PipelineProgress,
    bus: Option<&BusHandle>,
    spawner: &S,
) -> Result<RunStats, G2gError> {
    run_graph_threaded_inner(
        graph,
        clock,
        link_capacity,
        bus,
        None,
        Some(progress),
        None,
        None,
        None,
        spawner,
    )
    .await
}

/// As [`run_graph_threaded`], but taps live telemetry into `observer` (the
/// thread-per-arm analog of [`run_graph_observed`]): a concurrent task reads
/// per-element `process()` latency / input-link fill mid-run via
/// [`Observer::snapshot`](crate::runtime::Observer::snapshot) while the arms run
/// on their own OS threads.
#[cfg(all(feature = "std", feature = "multi-thread"))]
pub async fn run_graph_threaded_observed<Clk: PipelineClock, S: GraphSpawner>(
    graph: Graph<GraphNode>,
    clock: &Clk,
    link_capacity: impl Into<LinkCapacity>,
    observer: &Observer,
    spawner: &S,
) -> Result<RunStats, G2gError> {
    run_graph_threaded_inner(
        graph,
        clock,
        link_capacity,
        None,
        None,
        None,
        Some(observer),
        None,
        None,
        spawner,
    )
    .await
}

/// [`run_graph_recorded`]'s thread-per-arm twin: the same [`FlightRecorder`],
/// with each arm on its own OS thread. The rings are shared with the worker
/// threads that fill them, so a heavy multicore pipeline (the kind that runs
/// under [`run_graph_threaded`] in the first place) leaves the same replayable
/// per-edge recording behind when it fails.
///
/// `progress` is optional, as in [`run_graph_recorded`].
#[cfg(all(feature = "std", feature = "multi-thread"))]
pub async fn run_graph_threaded_recorded<Clk: PipelineClock, S: GraphSpawner>(
    graph: Graph<GraphNode>,
    clock: &Clk,
    link_capacity: impl Into<LinkCapacity>,
    progress: Option<&PipelineProgress>,
    bus: Option<&BusHandle>,
    recorder: &FlightRecorder,
    spawner: &S,
) -> Result<RunStats, G2gError> {
    run_graph_threaded_inner(
        graph,
        clock,
        link_capacity,
        bus,
        None,
        progress,
        None,
        Some(recorder),
        None,
        spawner,
    )
    .await
}

/// Zero-dependency [`GraphSpawner`]: each arm runs on its own `std` thread driven
/// by the park-based [`block_on`](crate::runtime::block_on). Dependency-free and
/// sufficient for graphs of pure-core elements (core channels + clock). Elements
/// that need a tokio reactor (network sources) require a tokio-backed spawner
/// instead (`g2g-plugins`' `TokioThreadSpawner`), since `block_on` provides no
/// I/O driver.
#[cfg(all(feature = "std", feature = "multi-thread"))]
#[derive(Debug, Default, Clone, Copy)]
pub struct ThreadSpawner;

#[cfg(all(feature = "std", feature = "multi-thread"))]
impl GraphSpawner for ThreadSpawner {
    fn spawn_arm(
        &self,
        build: alloc::boxed::Box<dyn FnOnce() -> LocalArmFuture + Send>,
    ) -> BoxFuture<'static, Result<u64, G2gError>> {
        // A capacity-1 channel is the handle: the worker delivers its one result,
        // the caller awaits it. Cross-thread wake is via the channel's waker.
        let (tx, rx) = crate::runtime::channel::bounded::<Result<u64, G2gError>>(1);
        std::thread::spawn(move || {
            let result = crate::runtime::block_on(build());
            // Best-effort: a dropped handle (the join aborted after a sibling's
            // error) just discards the result.
            let _ = tx.try_send(result);
        });
        Box::pin(async move { rx.recv().await.unwrap_or(Err(G2gError::Shutdown)) })
    }
}

/// Each node's declared per-alternative costs, indexed by node id, for the
/// solver's minimum-cost fixation. Source, transform and sink nodes declare
/// them; a muxer / demux node is not part of a linear chain, which is the only
/// topology the cost minimization covers. A source's costs index its produce
/// set, which without them is read in its own order.
fn build_node_preferences(vg: &ValidatedGraph<GraphNodeRef<'_>>) -> Vec<Option<CapsPreferences>> {
    (0..vg.node_count())
        .map(|i| {
            let node = NodeId(i as u32);
            match vg.kind(node) {
                NodeKind::Source => source_ref(vg, node).and_then(|s| s.caps_preferences()),
                NodeKind::Transform | NodeKind::Sink => {
                    element_ref(vg, node).and_then(|e| e.caps_preferences())
                }
                _ => None,
            }
        })
        .collect()
}

/// View a node's payload as a source. `None` for any other node kind.
fn source_ref<'g, 'a>(
    vg: &'g ValidatedGraph<GraphNodeRef<'a>>,
    node: NodeId,
) -> Option<&'g (dyn DynSourceLoop + 'a)> {
    match vg.element(node)? {
        GraphNodeRef::Source(src) => Some(&**src),
        GraphNodeRef::Element(_)
        | GraphNodeRef::Muxer(_)
        | GraphNodeRef::FanoutSource(_)
        | GraphNodeRef::Demux(_) => None,
    }
}

/// View a node's payload as a transform/sink element. `None` for a source or a
/// muxer (whose constraints the runner builds from their own trait methods).
fn element_ref<'g, 'a>(
    vg: &'g ValidatedGraph<GraphNodeRef<'a>>,
    node: NodeId,
) -> Option<&'g (dyn DynAsyncElement + 'a)> {
    match vg.element(node)? {
        GraphNodeRef::Element(elem) => Some(&**elem),
        GraphNodeRef::Source(_)
        | GraphNodeRef::Muxer(_)
        | GraphNodeRef::FanoutSource(_)
        | GraphNodeRef::Demux(_) => None,
    }
}

/// Build the per-node solver constraints for a validated graph, given each
/// source's probed produce set (indexed by node id, `None` for non-sources). The
/// constraints borrow their elements immutably, so the returned vec must be
/// dropped before any `&mut` borrow (configure). Shared by the runner's Phase 2
/// and the negotiate-only tooling path ([`negotiate_graph`]).
fn build_node_constraints<'g, 'a>(
    vg: &'g ValidatedGraph<GraphNodeRef<'a>>,
    source_caps: &[Option<CapsSet>],
) -> Result<Vec<NodeConstraint<'g>>, G2gError> {
    let mut constraints: Vec<NodeConstraint<'g>> = Vec::with_capacity(vg.node_count());
    for (i, src_caps) in source_caps.iter().enumerate() {
        let node = NodeId(i as u32);
        let nc = match vg.kind(node) {
            NodeKind::Source => {
                let set = src_caps.clone().ok_or(G2gError::CapsMismatch)?;
                NodeConstraint::Element(CapsConstraint::Produces(set))
            }
            NodeKind::Transform => {
                let elem = element_ref(vg, node).ok_or(G2gError::CapsMismatch)?;
                NodeConstraint::Element(elem.caps_constraint_as_transform())
            }
            NodeKind::Sink => {
                let elem = element_ref(vg, node).ok_or(G2gError::CapsMismatch)?;
                NodeConstraint::Element(elem.caps_constraint_as_sink())
            }
            // A plain (broadcast) tee is structural; the solver couples its
            // branches via `IdentityAny`. A demux that declares per-port caps
            // (M380) instead negotiates each branch against its port: build a
            // `Demux` constraint so a downstream decoder configures at startup.
            NodeKind::Tee(n) => {
                let ports: Vec<Option<Caps>> = match vg.element(node) {
                    Some(GraphNodeRef::Demux(elem)) => {
                        (0..n as usize).map(|p| elem.port_output_caps(p)).collect()
                    }
                    _ => Vec::new(),
                };
                if !ports.is_empty() && ports.iter().all(Option::is_some) {
                    let input = match vg.element(node) {
                        Some(GraphNodeRef::Demux(elem)) => elem.caps_constraint_as_input(),
                        _ => CapsConstraint::AcceptsAny,
                    };
                    let ports = ports
                        .into_iter()
                        .map(|c| CapsConstraint::Produces(CapsSet::one(c.expect("all Some"))))
                        .collect();
                    NodeConstraint::Demux { input, ports }
                } else {
                    NodeConstraint::Element(CapsConstraint::IdentityAny)
                }
            }
            NodeKind::Muxer(_) => {
                let GraphNodeRef::Muxer(elem) = vg.element(node).ok_or(G2gError::CapsMismatch)?
                else {
                    return Err(G2gError::CapsMismatch);
                };
                let inputs: Vec<CapsConstraint<'g>> = (0..elem.input_count())
                    .map(|pad| elem.caps_constraint_as_input(pad))
                    .collect();
                let follows = elem.output_follows_input();
                // An identity-passthrough mux derives its output from a pad, so it
                // need not (and may be unable to) declare output caps up front;
                // only ask for them in the independent-output case.
                let output = match follows {
                    Some(_) => CapsConstraint::AcceptsAny,
                    None => elem
                        .caps_constraint_for_output()
                        .map_err(|_| G2gError::CapsMismatch)?,
                };
                NodeConstraint::Muxer {
                    inputs,
                    output,
                    follows,
                }
            }
            // A terminal fan-out source produces per-port caps: the demux
            // constraint shape with the input half inert (no input edge).
            NodeKind::FanoutSrc(n) => {
                let GraphNodeRef::FanoutSource(elem) =
                    vg.element(node).ok_or(G2gError::CapsMismatch)?
                else {
                    return Err(G2gError::CapsMismatch);
                };
                let ports: Vec<CapsConstraint<'g>> = (0..n as usize)
                    .map(|p| {
                        elem.output_caps(p)
                            .map(|c| CapsConstraint::Produces(CapsSet::one(c)))
                    })
                    .collect::<Result<_, _>>()?;
                NodeConstraint::Demux {
                    input: CapsConstraint::AcceptsAny,
                    ports,
                }
            }
            // A terminal fan-in reuses the muxer constraint shape with the
            // output half inert (no output edge exists to couple).
            NodeKind::FaninSink(_) => {
                let GraphNodeRef::Muxer(elem) = vg.element(node).ok_or(G2gError::CapsMismatch)?
                else {
                    return Err(G2gError::CapsMismatch);
                };
                let inputs: Vec<CapsConstraint<'g>> = (0..elem.input_count())
                    .map(|pad| elem.caps_constraint_as_input(pad))
                    .collect();
                NodeConstraint::Muxer {
                    inputs,
                    output: CapsConstraint::AcceptsAny,
                    follows: None,
                }
            }
        };
        constraints.push(nc);
    }
    Ok(constraints)
}

/// A node's label for the caps explainer and the DOT dump: the element's log
/// category (e.g. `h264parse`), falling back to the structural kind for a tee
/// (which carries no element).
fn caps_label(vg: &ValidatedGraph<GraphNodeRef<'_>>, node: NodeId) -> alloc::string::String {
    match vg.element(node) {
        Some(e) => e.log_category().to_string(),
        None => match vg.kind(node) {
            NodeKind::Tee(_) => "tee".to_string(),
            k => alloc::format!("{k:?}"),
        },
    }
}

/// Run startup caps negotiation only, without running the pipeline: validate the
/// graph, probe each source's caps (Phase 1, async), and solve the whole-graph
/// CSP (Phase 2), returning the validated graph, the fixated caps per edge, and
/// each edge's memory domain (all indexed by edge id, as
/// [`crate::dot::DotAnnotations`] expects). The per-edge domain is the producing
/// node's [`output_memory`](GraphNodeRef::output_memory) (M285), so a GPU /
/// zero-copy link shows up in the dump. For tooling that wants the *chosen* caps
/// without moving data, e.g. `g2g-launch --dot`. It performs the same
/// source-caps probing the runner does, so a source that connects on
/// `intercept_caps` (a live ingress) will do so here too; a negotiation failure
/// returns `CapsMismatch` (the caller can fall back to a topology-only dump).
pub async fn negotiate_graph<'a>(
    graph: Graph<GraphNodeRef<'a>>,
) -> Result<
    (
        ValidatedGraph<GraphNodeRef<'a>>,
        Vec<Caps>,
        Vec<crate::memory::MemoryDomainKind>,
    ),
    G2gError,
> {
    negotiate_graph_explained(graph).await.map_err(|e| match e {
        NegotiateError::Setup(err) => err,
        NegotiateError::Solve(_) => G2gError::CapsMismatch,
    })
}

/// Why [`negotiate_graph_explained`] could not negotiate a graph. `Setup` is a
/// structural / I/O failure before the solve (too few nodes, a bad source, a
/// source caps-probe error); `Solve` carries the structured
/// [`NegotiationFailure`] naming the conflicting link, which the opaque
/// [`negotiate_graph`] flattens to `CapsMismatch`.
#[derive(Debug)]
pub enum NegotiateError {
    Setup(G2gError),
    Solve(NegotiationFailure),
}

/// As [`negotiate_graph`], but preserves the structured [`NegotiationFailure`]
/// on a solve conflict (for the caps-negotiation explainer / `validate`
/// tooling). `negotiate_graph` is the opaque wrapper over this.
pub async fn negotiate_graph_explained<'a>(
    graph: Graph<GraphNodeRef<'a>>,
) -> Result<
    (
        ValidatedGraph<GraphNodeRef<'a>>,
        Vec<Caps>,
        Vec<crate::memory::MemoryDomainKind>,
    ),
    NegotiateError,
> {
    let mut vg = graph
        .finish()
        .map_err(|_| NegotiateError::Setup(G2gError::CapsMismatch))?;
    let n = vg.node_count();
    if n < 2 {
        return Err(NegotiateError::Setup(G2gError::CapsMismatch));
    }
    let topo = vg.topo().to_vec();

    // Phase 1: probe each source's produce set (async), releasing the mutable
    // borrow before the constraint phase borrows every node immutably.
    let mut source_caps: Vec<Option<CapsSet>> = (0..n).map(|_| None).collect();
    for &node in &topo {
        if matches!(vg.kind(node), NodeKind::Source) {
            let GraphNodeRef::Source(src) = vg
                .element_mut(node)
                .ok_or(NegotiateError::Setup(G2gError::CapsMismatch))?
            else {
                return Err(NegotiateError::Setup(G2gError::CapsMismatch));
            };
            source_caps[node.0 as usize] =
                Some(src.produced_caps().await.map_err(NegotiateError::Setup)?);
        }
    }

    // Phase 2: build constraints and solve. Scope the immutable borrow so `vg`
    // moves out cleanly in the return.
    let solution = {
        let constraints =
            build_node_constraints(&vg, &source_caps).map_err(NegotiateError::Setup)?;
        let preferences = build_node_preferences(&vg);
        solve_graph_preferred(&vg, &constraints, &preferences, &|node| {
            caps_label(&vg, node)
        })
        .map_err(NegotiateError::Solve)?
    };

    // Phase 3.5's allocation cascade, run for its effect on the elements: it is
    // what settles a multi-domain producer (a decoder that can keep frames on
    // the GPU or download them) on the domain its consumer asked for. Reading
    // `output_memory` without it reports every producer's default preference, so
    // the dump would call a downloading link a GPU link.
    cascade_allocation(&mut vg, &topo, &solution).map_err(NegotiateError::Setup)?;

    // Per-edge memory domain: the domain of the node producing onto that edge.
    let edge_memory: Vec<crate::memory::MemoryDomainKind> = (0..vg.edge_count())
        .map(|id| {
            let src = vg.edge(id).src.node;
            vg.element(src)
                .map(|e| e.output_memory())
                .unwrap_or(crate::memory::MemoryDomainKind::System)
        })
        .collect();

    Ok((vg, solution, edge_memory))
}

/// Build the [`CopyPlan`](crate::copyplan::CopyPlan) for a negotiated graph from the
/// three arrays [`negotiate_graph`] returns (the validated graph, per-edge fixated
/// caps, and per-edge memory domain). Extracts each node's label + output domain and
/// each edge's producer/consumer/domain/caps into the flat profiles the pure
/// analysis works over, so tooling (e.g. `g2g-launch --copy-plan`) or a graph-level
/// copy budget can inspect the memory-domain path before running.
pub fn copy_plan(
    vg: &ValidatedGraph<GraphNodeRef<'_>>,
    edge_caps: &[Caps],
    edge_memory: &[crate::memory::MemoryDomainKind],
) -> crate::copyplan::CopyPlan {
    let nodes: Vec<crate::copyplan::NodeProfile> = (0..vg.node_count())
        .map(|i| {
            let node = NodeId(i as u32);
            crate::copyplan::NodeProfile {
                label: caps_label(vg, node),
                out_domain: vg
                    .element(node)
                    .map(|e| e.output_memory())
                    .unwrap_or(crate::memory::MemoryDomainKind::System),
            }
        })
        .collect();
    let edges: Vec<crate::copyplan::EdgeProfile> = (0..vg.edge_count())
        .map(|id| {
            let e = vg.edge(id);
            crate::copyplan::EdgeProfile {
                src: e.src.node.0 as usize,
                dst: e.dst.node.0 as usize,
                domain: edge_memory[id],
                caps: edge_caps[id].clone(),
            }
        })
        .collect();
    crate::copyplan::CopyPlan::analyze(&nodes, &edges)
}

/// Run the allocation cascade over a solved graph, in reverse topo order: each
/// element absorbs the proposal arriving on its output edge(s)
/// (`configure_allocation`), then proposes from its output-link caps, and the
/// proposal is stored on its input edge(s) for its upstream to absorb. A tee
/// joins its branch proposals (most-restrictive intersection, loud failure on a
/// domain conflict) onto its single input; a muxer proposes its own per-pad
/// demand onto each input edge. For a linear chain this is byte-for-byte the
/// linear runner's sink->source fold.
///
/// Returns the source's absorbed proposal (what the runner reports as its
/// `allocation`). Shared with [`negotiate_graph`], which runs it
/// purely to settle each element's domain so a graph dump reports the negotiated
/// memory domain rather than every producer's default preference.
fn cascade_allocation(
    vg: &mut ValidatedGraph<GraphNodeRef<'_>>,
    topo: &[NodeId],
    solution: &[Caps],
) -> Result<Option<AllocationParams>, G2gError> {
    let nee = vg.edge_count();
    let mut edge_proposal: Vec<Option<AllocationParams>> = (0..nee).map(|_| None).collect();
    let mut allocation: Option<AllocationParams> = None;
    for &node in topo.iter().rev() {
        match vg.kind(node) {
            NodeKind::Sink => {
                let in_e = vg.in_edges(node)[0];
                let caps = solution[in_e].clone();
                edge_proposal[in_e] = element_propose(vg, node, &caps, MetaRequests::new());
            }
            NodeKind::Transform => {
                let in_e = vg.in_edges(node)[0];
                let out_e = vg.out_edges(node)[0];
                // A transform is a memory-domain pass-through here: it forwards the
                // downstream proposal to its own pool and re-proposes upstream
                // unchanged. Domain capability is enforced at the buffer-pool
                // origin (the source) and at the sibling join (the tee), not at
                // every hop, so a GPU proposal merely passing through a plain
                // transform is not rejected against its System default (M351).
                if let Some(p) = edge_proposal[out_e] {
                    element_configure_alloc(vg, node, &p);
                }
                let caps = solution[out_e].clone();
                let downstream = edge_meta_requests(edge_proposal[out_e]);
                edge_proposal[in_e] = element_propose(vg, node, &caps, downstream);
            }
            NodeKind::Tee(_) => {
                let in_e = vg.in_edges(node)[0];
                // The first branch seeds the join: a `None` from there on means a
                // branch that asked for nothing, which is not the same as no
                // branch yet (it vetoes a demand needing every consumer).
                let mut joined: Option<AllocationParams> = None;
                for (i, &oe) in vg.out_edges(node).iter().enumerate() {
                    joined = match i {
                        0 => edge_proposal[oe],
                        _ => join_alloc(joined, edge_proposal[oe])?,
                    };
                }
                // A plain tee carries no element and narrows nothing; a demux
                // node folds in the domains it declared it can parse, so a
                // System-only demuxer makes a GPU producer download (M1039).
                edge_proposal[in_e] = narrow_to_input_domains(joined, node_input_domains(vg, node));
            }
            NodeKind::Source => {
                let out_e = vg.out_edges(node)[0];
                if let Some(p) = edge_proposal[out_e] {
                    // M351: reconcile against the source's emittable domains, the
                    // upstream end of the two-sided negotiation. The reconciled
                    // proposal is what the source allocates and what `RunStats`
                    // reports.
                    let can = node_output_domains(vg, node);
                    // A proposal carrying only metadata demand accepts every
                    // domain, so reconciling it would let a metadata request pick
                    // the source's memory domain: pass it through untouched.
                    let resolved = match p.constrains_pool() {
                        true => p.resolve_for_producer(can)?,
                        false => p,
                    };
                    if let GraphNodeRef::Source(src) =
                        vg.element_mut(node).ok_or(G2gError::CapsMismatch)?
                    {
                        src.configure_allocation(&resolved);
                    }
                    allocation = Some(resolved);
                }
            }
            // A terminal fan-out source exposes no allocation hook (its
            // outputs are network-generated System bytes).
            NodeKind::FanoutSrc(_) => {}
            NodeKind::Muxer(_) | NodeKind::FaninSink(_) => {
                // A muxer / terminal fan-in asks each input pad for the allocation
                // it wants (most are content-agnostic and propose nothing), storing
                // it on that input edge so the demand crosses the boundary and
                // re-cascades up the branch like any other downstream proposal. The
                // muxer's own output edge proposal is not absorbed here: a container
                // muxer's byte output has no memory-domain tie to its inputs, and a
                // terminal fan-in has no output at all.
                //
                // M976: downstream *metadata* demand does cross the output, on its
                // own, because it describes the frames the fan-in writes (a GPU
                // compositor deciding whether to declare its row padding). Nothing
                // is called when no demand was declared.
                let out_demand = vg
                    .out_edges(node)
                    .first()
                    .map(|&out_e| edge_meta_requests(edge_proposal[out_e]))
                    .unwrap_or_default();
                if !out_demand.is_empty() {
                    if let Some(GraphNodeRef::Muxer(mux)) = vg.element_mut(node) {
                        mux.configure_allocation_for_output(&AllocationParams::meta_demand(
                            out_demand,
                        ));
                    }
                }
                let accepts = node_input_domains(vg, node);
                if let Some(GraphNodeRef::Muxer(mux)) = vg.element(node) {
                    for &in_e in vg.in_edges(node) {
                        let pad = vg.edge(in_e).dst.index as usize;
                        let caps = solution[in_e].clone();
                        edge_proposal[in_e] = narrow_to_input_domains(
                            mux.propose_allocation_for_input(pad, &caps),
                            accepts,
                        );
                    }
                }
            }
        }
    }
    Ok(allocation)
}

/// A transform/sink node's allocation proposal from `caps` (its output-link caps
/// for a transform, its input-link caps for a sink), carrying the node's own meta
/// requests plus `downstream`'s onward up the cascade (M976). `None` for other
/// kinds.
fn element_propose(
    vg: &ValidatedGraph<GraphNodeRef<'_>>,
    node: NodeId,
    caps: &Caps,
    downstream: MetaRequests,
) -> Option<AllocationParams> {
    match vg.element(node) {
        Some(GraphNodeRef::Element(elem)) => with_meta_demand(
            narrow_to_input_domains(elem.propose_allocation(caps), elem.input_domains()),
            elem.meta_requests().carry_upstream(downstream),
        ),
        _ => None,
    }
}

/// Fold the domains an element declared it can take into the proposal it hands
/// its producer. Without this an element that accepts one domain but proposes
/// nothing leaves its producer free to pick any domain, and the mismatch only
/// surfaces as an `UnsupportedDomain` on the first frame; the declaration used
/// to reach nothing but the converter auto-plug. The all-domains default (what
/// an element that never thought about memory reports) narrows nothing, so a
/// graph of such elements cascades exactly as before.
fn narrow_to_input_domains(
    proposal: Option<AllocationParams>,
    accepts: DomainSet,
) -> Option<AllocationParams> {
    if accepts == DomainSet::ALL {
        return proposal;
    }
    let narrowed = match proposal {
        Some(p) => p.accepts.intersect(accepts),
        None => accepts,
    };
    // An element whose own proposal contradicts its declaration is
    // self-inconsistent: keep the explicit proposal, so the producer-side
    // reconcile reports the conflict rather than this silently picking a side.
    let Some(domain) = narrowed.preferred() else {
        return proposal;
    };
    Some(AllocationParams {
        domain,
        accepts: narrowed,
        ..proposal.unwrap_or_default()
    })
}

/// The metadata demand an edge's stored proposal carries, empty when there is
/// none.
fn edge_meta_requests(proposal: Option<AllocationParams>) -> MetaRequests {
    proposal.map(|p| p.meta_requests).unwrap_or_default()
}

/// The set of memory domains a node accepts on its inputs, for folding into the
/// proposal it hands upstream. A node without an element (a structural tee)
/// requires nothing.
fn node_input_domains(vg: &ValidatedGraph<GraphNodeRef<'_>>, node: NodeId) -> DomainSet {
    vg.element(node)
        .map(|n| n.input_domains())
        .unwrap_or(DomainSet::ALL)
}

/// The set of memory domains a node can emit (M351), for reconciling a
/// downstream allocation proposal against the producer's real capability.
/// Missing nodes report a System singleton (the conservative default).
fn node_output_domains(
    vg: &ValidatedGraph<GraphNodeRef<'_>>,
    node: NodeId,
) -> crate::memory::DomainSet {
    vg.element(node)
        .map(|n| n.output_domains())
        .unwrap_or(crate::memory::DomainSet::only(
            crate::memory::MemoryDomainKind::System,
        ))
}

/// Apply a downstream-derived allocation proposal to a transform's own pool.
fn element_configure_alloc(
    vg: &mut ValidatedGraph<GraphNodeRef<'_>>,
    node: NodeId,
    params: &AllocationParams,
) {
    if let Some(GraphNodeRef::Element(elem)) = vg.element_mut(node) {
        elem.configure_allocation(params);
    }
}

/// A node's latency contribution. `None` for structural (tee) and muxer nodes.
fn element_latency(vg: &ValidatedGraph<GraphNodeRef<'_>>, node: NodeId) -> Option<LatencyReport> {
    match vg.element(node) {
        Some(GraphNodeRef::Source(src)) => Some(src.latency()),
        Some(GraphNodeRef::Element(elem)) => Some(elem.latency()),
        _ => None,
    }
}

/// How often the runner reads the elected clock's health. Coarse on purpose:
/// losing a grandmaster is a seconds-scale event, and the check costs a lock on
/// the clock.
const CLOCK_HEALTH_PERIOD_NS: u64 = 1_000_000_000;

/// Watch the elected clock and re-elect when it loses the reference it
/// disciplines to (M1004). Runs alongside the graph's arms, sleeping on the same
/// timer the fan-in arms tick on so it stays executor-agnostic, and posts
/// [`BusMessage::ClockLost`] on each healthy -> unhealthy edge. It then elects
/// again over the candidates that are still healthy and retargets `elected`, which
/// is what every sink's [`ClockSync`] reads through. With no healthy candidate
/// left the pipeline keeps the clock it has: it still tells time, it is just no
/// longer disciplined, and a later re-lock is picked up by the same check.
///
/// Never returns; the runner drops it once the arms have finished.
async fn clock_health_monitor(
    candidates: Vec<ClockCandidate>,
    elected: alloc::sync::Arc<ElectedClock>,
    bus: BusHandle,
    ticker: &dyn DynAsyncClock,
) -> core::convert::Infallible {
    let mut was_healthy = true;
    loop {
        let deadline = ticker.now_ns().saturating_add(CLOCK_HEALTH_PERIOD_NS);
        ticker.sleep_until_ns(deadline).await;
        let healthy = elected.healthy();
        if healthy || !was_healthy {
            was_healthy = healthy;
            continue;
        }
        was_healthy = false;
        bus.try_post(BusMessage::ClockLost);
        if let Some(c) = elect_clock(
            candidates
                .iter()
                .filter(|c| c.clock.healthy())
                .cloned()
                .map(Some),
        ) {
            elected.swap(c.clock.clone());
            was_healthy = elected.healthy();
        }
    }
}

/// A node's offered clock for the pipeline clock election.
fn element_clock(vg: &ValidatedGraph<GraphNodeRef<'_>>, node: NodeId) -> Option<ClockCandidate> {
    match vg.element(node) {
        Some(GraphNodeRef::Source(src)) => src.provide_clock(),
        Some(GraphNodeRef::Element(elem)) => elem.provide_clock(),
        _ => None,
    }
}

/// Join two allocation proposals at a tee's input. Both branches consume the
/// one upstream producer, so the result is the most-restrictive per-parameter
/// intersection ([`AllocationParams::join`]): the larger size, count, and
/// alignment, with a matching memory domain. Divergent domains are an empty
/// intersection and fail loud with [`G2gError::AllocationConflict`] (no single
/// pool can satisfy, say, a CUDA branch and a D3D11 branch at once).
/// M976: a branch that proposes nothing still counts as a branch that requested
/// nothing, so a demand needing every consumer dies against it; and a join left
/// carrying neither pool constraints nor demand collapses back to `None`, so a
/// died-out request leaves the cascade exactly as it found it.
fn join_alloc(
    a: Option<AllocationParams>,
    b: Option<AllocationParams>,
) -> Result<Option<AllocationParams>, G2gError> {
    let joined = match (a, b) {
        (Some(x), Some(y)) => Some(x.join(y)?),
        (Some(x), None) => {
            Some(x.with_meta_requests(x.meta_requests.join_branches(MetaRequests::new())))
        }
        (None, Some(y)) => {
            Some(y.with_meta_requests(y.meta_requests.join_branches(MetaRequests::new())))
        }
        (None, None) => None,
    };
    Ok(joined.filter(|p| p.constrains_pool() || !p.meta_requests.is_empty()))
}

/// Re-solve one muxer input pad against the boundary's new caps (MX-1).
fn solve_mux_input_dyn(
    new_caps: &Caps,
    mux: &dyn DynMultiInputElement,
    pad: usize,
) -> Result<Caps, G2gError> {
    let src_c = CapsConstraint::LegacySource(new_caps.clone());
    let mux_c = mux.caps_constraint_as_input(pad);
    let links = solve_linear(&[&src_c, &mux_c]).map_err(|_| G2gError::CapsMismatch)?;
    links.last().cloned().ok_or(G2gError::CapsMismatch)
}

/// Re-derive the merged muxer output from its current per-input config (MX-2).
fn solve_mux_output_dyn(mux: &dyn DynMultiInputElement) -> Result<Caps, G2gError> {
    let mux_c = mux
        .caps_constraint_for_output()
        .map_err(|_| G2gError::CapsMismatch)?;
    let sink_c = CapsConstraint::AcceptsAny;
    let links = solve_linear(&[&mux_c, &sink_c]).map_err(|_| G2gError::CapsMismatch)?;
    links.last().cloned().ok_or(G2gError::CapsMismatch)
}

async fn source_arm<'a>(
    mut src: Box<dyn DynSourceLoop + 'a>,
    out_tx: LinkSender,
    bus: Option<BusHandle>,
    progress: Option<PipelineProgress>,
) -> Result<u64, G2gError> {
    // M206: announce the stream start before any data, one per source, so an
    // application can bracket each stream's lifetime (StreamStart .. Eos).
    if let Some(b) = &bus {
        b.try_post(BusMessage::StreamStart);
    }
    // M203: publish the source's duration (if it knows one) before producing,
    // so a `DURATION` query is answerable from the first poll, and push-notify a
    // change on the bus. Polled once here; a source that discovers its length
    // mid-stream is a follow-up (it would publish through the handle directly).
    if let Some(duration_ns) = src.query_duration() {
        let changed = progress
            .as_ref()
            .map(|p| p.publish_duration(duration_ns))
            .unwrap_or(true);
        if changed {
            if let Some(b) = &bus {
                b.try_post(BusMessage::DurationChanged { duration_ns });
            }
        }
    }
    let mut adapter = SenderSink::new(out_tx);
    // M81: open the stream with a SEGMENT ahead of the source's data, so every
    // downstream branch maps timestamps to running time from the first frame.
    let _ = adapter
        .push(PipelinePacket::Segment(Segment::new()))
        .await?;
    src.run(&mut adapter).await
}

/// An interior transform arm. Besides forwarding data, it (D4) selects on a β
/// `ArmDirective` channel alongside its data link so an upstream re-cascade
/// reaches it while parked on data, and on a mid-stream `CapsChanged` it steers
/// its forwarded output toward a downstream-acceptable shape using its
/// `downstream_feasible` snapshot (Caps-α), failing loud via a reverse
/// reconfigure if downstream positively rejects every output it can produce.
/// Everything a transform arm needs besides its element (M1000). Opaque on
/// purpose: only the runner can build one, so the `drive_transform_arm` hook
/// on `DynAsyncElement` stays implementable only through the blanket impl.
#[doc(hidden)]
#[allow(missing_debug_implementations)]
pub struct TransformArmIo {
    pub(crate) in_rx: LinkReceiver,
    pub(crate) out_tx: LinkSender,
    pub(crate) arm_rx: Receiver<ArmDirective>,
    pub(crate) coord: GraphCoordHandle,
    pub(crate) node: NodeId,
    pub(crate) out_caps: Caps,
    pub(crate) downstream_feasible: Option<CapsSet>,
    pub(crate) mode: BranchMode,
    pub(crate) bus: Option<BusHandle>,
    pub(crate) probe: Probe,
    pub(crate) control: Option<ArmController>,
}

/// As [`TransformArmIo`], for the sink arm.
#[doc(hidden)]
#[allow(missing_debug_implementations)]
pub struct SinkArmIo {
    pub(crate) in_rx: LinkReceiver,
    pub(crate) arm_rx: Receiver<ArmDirective>,
    pub(crate) coord: GraphCoordHandle,
    pub(crate) node: NodeId,
    pub(crate) mode: BranchMode,
    pub(crate) bus: Option<BusHandle>,
    pub(crate) state: Option<StateController>,
    pub(crate) progress: Option<PipelineProgress>,
    pub(crate) probe: Probe,
    pub(crate) control: Option<ArmController>,
}

/// Monomorphized over the element type by the `drive_transform_arm` blanket
/// hook (M1000), so the per-frame `process` future is the element's own
/// unboxed state machine, not a `Box<dyn Future>`.
#[doc(hidden)]
pub async fn transform_arm<E: AsyncElement>(
    mut elem: E,
    io: TransformArmIo,
) -> Result<u64, G2gError> {
    let TransformArmIo {
        in_rx,
        out_tx,
        arm_rx,
        coord,
        node,
        mut out_caps,
        downstream_feasible,
        mode,
        bus,
        probe,
        control,
    } = io;
    let mut adapter = SenderSink::new(out_tx);
    // M947: charge time spent blocked on the downstream link to this element's
    // push-wait, not to its `process()` compute.
    adapter.set_push_wait_probe(probe.clone());
    // M175: relay a downstream QoS report (seen on this transform's output link)
    // onto its input link, so it reaches the source/decoder one hop at a time
    // through any number of generic transforms, not just the sink's direct
    // upstream. The element's `process` is unaffected. M720 extends the same
    // hop to keyframe requests and bitrate targets when the element does not
    // consume them itself, so a PLI / BWE estimate crosses a parser between
    // the encoder and a WebRTC sink. M997 does the same for QoS: an element
    // that sheds work on a report (a decoder skipping non-reference frames)
    // observes it instead.
    if !elem.handles_qos() {
        adapter.relay_qos_to(in_rx.qos_slot());
    }
    adapter.relay_reconfigure_to(
        in_rx.reconfigure_slot(),
        ReconfigureAnswered {
            keyframe: elem.handles_keyframe_requests(),
            orientation: elem.handles_orientation(),
        },
    );
    if !elem.handles_bitrate_requests() {
        adapter.relay_bitrate_to(in_rx.bitrate_slot());
    }
    let mut control_open = true;
    let mut last_buffer_bucket: Option<u8> = None;
    loop {
        // M843: interior links report their fill too, so an application can see
        // *where* a pipeline is starved or backed up, not just at the sink.
        report_buffering(bus.as_ref(), &probe, &in_rx, &mut last_buffer_bucket);
        let packet = if control_open {
            match select2(arm_rx.recv(), in_rx.recv()).await {
                Either::Left(Some(ArmDirective::Recascade(params))) => {
                    // β: absorb the downstream proposal, re-derive our own from
                    // our output caps, and report it so the cascade continues to
                    // our upstream neighbour.
                    elem.configure_allocation(&params);
                    let proposal = with_meta_demand(
                        elem.propose_allocation(&out_caps),
                        elem.meta_requests().carry_upstream(params.meta_requests),
                    );
                    coord
                        .report(Recascade {
                            node,
                            route: RecascadeRoute::Upstream,
                            proposal,
                        })
                        .await;
                    continue;
                }
                Either::Left(Some(ArmDirective::ProducerAllocation(params))) => {
                    // M839: an upstream muxer settled a new pool on our input
                    // link. Absorb it and stay quiet: re-proposing here would
                    // walk straight back into the boundary it came from.
                    elem.configure_allocation(&params);
                    continue;
                }
                Either::Left(None) => {
                    control_open = false;
                    continue;
                }
                Either::Right(packet) => packet,
            }
        } else {
            in_rx.recv().await
        };
        match packet {
            Some(PipelinePacket::Eos) => {
                elem.process(PipelinePacket::Eos, &mut adapter).await?;
                // M909: skip our push when the element's catch-all arm already
                // forwarded the sentinel, so the sink sees exactly one `Eos`.
                if !adapter.eos_forwarded() {
                    adapter.push(PipelinePacket::Eos).await?;
                }
                // Drop our report handle so the coordinator can wind down once
                // every arm exits, then drain any tail-end re-cascade directive
                // still in flight (a β triggered by the final pre-EOS frames).
                // Dropping before the drain decouples wind-down from the drain,
                // so no arm blocks holding the last handle.
                drop(coord);
                while let Some(directive) = arm_rx.recv().await {
                    elem.configure_allocation(directive.params());
                }
                return Ok(0);
            }
            Some(PipelinePacket::CapsChanged(new_caps)) => {
                // Caps-α: derive the forwarded output from this element's
                // constraint steered by the downstream feasibility snapshot.
                // `Defer` keeps the prior behavior (forward the incoming caps);
                // `Infeasible` fails the run loud (no producer renegotiates).
                let (forward_caps, output_resolved) = {
                    let constraint = elem.caps_constraint_as_transform();
                    match resolve_forward_output(
                        &constraint,
                        &new_caps,
                        downstream_feasible.as_ref(),
                        Some(&out_caps),
                    ) {
                        ForwardResolve::Fixed(caps) => (caps, true),
                        ForwardResolve::Defer => (new_caps.clone(), false),
                        ForwardResolve::Infeasible(failure) => {
                            // A genuinely infeasible re-solve: the refined caps
                            // have no solution against the downstream chain, and
                            // no runtime producer renegotiates its output caps, so
                            // there is nothing to reverse-reconfigure into. A tee
                            // branch (`FailLoud`) and a single-producer chain
                            // (`Reconfigure`) both fail the run loud, naming the
                            // conflict, rather than flowing stale caps; `Drop` (a
                            // tee) ends this branch while its siblings continue.
                            match mode {
                                BranchMode::FailLoud | BranchMode::Reconfigure => {
                                    report_runtime_caps_conflict(&new_caps);
                                    report_nego_failure(bus.as_ref(), failure);
                                    return Err(G2gError::CapsMismatch);
                                }
                                BranchMode::Drop => {
                                    report_nego_failure(bus.as_ref(), failure);
                                    return Ok(0);
                                }
                            }
                        }
                    }
                };
                let instance = probe.as_deref().map(|p| p.name());
                log_caps_forward(instance, &new_caps, &forward_caps, output_resolved);
                match log_caps_rejected(instance, &new_caps, elem.configure_pipeline(&new_caps))? {
                    ConfigureOutcome::Accepted => {
                        // M188: re-resolve a caps-driven transform's output target
                        // on the mid-stream change too (matches startup, line ~421
                        // / the linear coordinator arm). No-op for property-driven
                        // or passthrough elements. Skipped on a Defer: there
                        // `forward_caps` is the incoming INPUT caps, not this
                        // element's output, and `configure_output`'s contract is
                        // output caps only (a strict transform rightly rejects
                        // an input-shaped set, e.g. OpusDec fed `audio/x-opus`).
                        if output_resolved {
                            log_caps_rejected(
                                instance,
                                &forward_caps,
                                elem.configure_output(&forward_caps),
                            )?;
                        }
                        realloc_local_dyn(&mut elem as &mut dyn DynAsyncElement, &forward_caps);
                        // On a Defer `forward_caps` is the incoming INPUT caps:
                        // keep the last known real output as the shape to steer
                        // future re-solves (and allocation proposals) by.
                        if output_resolved {
                            out_caps = forward_caps.clone();
                        }
                        elem.process(PipelinePacket::CapsChanged(forward_caps), &mut adapter)
                            .await?;
                    }
                    ConfigureOutcome::ReFixate(counter) => {
                        in_rx.request_reconfigure(Reconfigure::Propose(counter));
                    }
                }
            }
            Some(packet) => {
                // M759: if this element opts into metadata auto-application,
                // derive the propagated set from this input frame and stash it on
                // the adapter, which attaches it to fresh (meta-empty) outputs.
                // Recomputed per input frame: exact for a 1-in-1-out transform,
                // most-recent-input association for a pipelined one. Cloning the
                // set is cheap (Arc refcount bumps); an empty result clears the
                // stash so a Drop verdict never leaks a stale set.
                #[cfg(feature = "metadata")]
                if let (Some(t), PipelinePacket::DataFrame(frame)) =
                    (elem.meta_transform(), &packet)
                {
                    if frame.meta.is_empty() {
                        adapter.set_meta_stash(None);
                    } else {
                        let mut propagated = frame.meta.clone();
                        propagated.propagate(t);
                        adapter.set_meta_stash((!propagated.is_empty()).then_some(propagated));
                    }
                }
                // M399: time the data-frame `process()` and sample input fill;
                // control packets (segment/flush) are excluded so the histogram
                // reflects real per-frame work, not cheap signalling.
                let seq = match &packet {
                    PipelinePacket::DataFrame(f) => Some(f.sequence),
                    _ => None,
                };
                let timed = probe.as_deref().filter(|_| seq.is_some());
                let mut wait_ns = 0;
                if let Some(p) = timed {
                    p.record_fill(in_rx.fill_percent());
                    // Queue-residency of this frame on the input link (M684).
                    if let Some(t) = in_rx.pop_transit_ns() {
                        p.record_transit(t);
                        wait_ns = t;
                    }
                }
                // M882: animated properties are sampled at this frame's PTS, so
                // the element processes it under the values that frame's time
                // calls for.
                apply_control(
                    control.as_ref(),
                    &mut elem as &mut dyn DynAsyncElement,
                    &packet,
                    &probe,
                )?;
                let t0 = ElementProbe::mark();
                elem.process(packet, &mut adapter).await?;
                if let (Some(p), Some(seq)) = (timed, seq) {
                    let push_wait_ns = p.record_proc_since(t0);
                    // M851: this frame's own wait + work, joined across stages
                    // by sequence id at snapshot time. M951: the push-wait the
                    // call just banked is charged to this visit too, so the
                    // journey's work segment is compute.
                    p.record_visit(seq, wait_ns, t0, push_wait_ns);
                }
                // M1036: renegotiation a transform originates rather than
                // relays (a decoder that read a new resolution out of the
                // bitstream); store it on the input link, where the upstream
                // producer observes it as `PushOutcome::Reconfigure`.
                if let Some(reconf) = elem.take_reconfigure() {
                    in_rx.request_reconfigure(reconf);
                }
            }
            None => return Ok(0),
        }
    }
}

/// A sink arm. On a mid-stream `CapsChanged` (D4) it re-solves its input against
/// its declared constraint; on accept it re-derives its own pool and reports the
/// proposal so the β cascade walks one hop upstream. A re-solve failure surfaces
/// loud as a reverse reconfigure into the boundary that emitted the change.
/// M839: it also selects on the β control channel, so the pool an upstream muxer
/// settles on for its merged output reaches the element that reads it.
/// Monomorphized over the element type by the `drive_sink_arm` blanket hook
/// (M1000); see [`transform_arm`].
#[doc(hidden)]
pub async fn sink_arm<E: AsyncElement>(mut elem: E, io: SinkArmIo) -> Result<u64, G2gError> {
    let probe = io.probe.clone();
    let result = sink_arm_loop(&mut elem, io).await;
    // A paced sink's presented / dropped counters, read here (after the loop
    // ended, on any exit) so the snapshot taken once every arm joined sees a
    // settled value.
    if let (Some(p), Some(stats)) = (&probe, elem.presentation_stats()) {
        p.set_presentation(stats);
    }
    result
}

async fn sink_arm_loop<E: AsyncElement>(elem: &mut E, io: SinkArmIo) -> Result<u64, G2gError> {
    let SinkArmIo {
        in_rx,
        arm_rx,
        coord,
        node,
        mode,
        bus,
        state,
        progress,
        probe,
        control,
    } = io;
    let mut null = NullSink;
    let mut consumed = 0u64;
    let mut prerolled_self = false;
    let mut last_buffer_bucket: Option<u8> = None;
    // M203: the segment in force, so a buffer's PTS maps to stream-time position.
    let mut current_segment: Option<Segment> = None;
    // M360 re-preroll: generation this arm last prerolled at, and whether it is
    // draining stale pre-seek frames (paused flushing seek) until the `Flush`.
    let mut preroll_gen = state.as_ref().map_or(0, |sc| sc.preroll_generation());
    let mut flushing = false;
    // M839: while the coordinator lives, race the data link against the β control
    // channel; once it closes, degrade to data-only so the closed arm can't spin.
    let mut control_open = true;
    loop {
        // M78 flow gate: below `Playing` the sink parks here, so it stops
        // draining its edge and backpressure stalls the DAG upstream. Non-live
        // `Paused` admits this sink's one preroll buffer; `Null` ends the arm.
        if let Some(sc) = &state {
            if sc.flow_gate(prerolled_self, preroll_gen).await == Flow::Stop {
                return Ok(consumed);
            }
            // M360: a `request_repreroll` (paused flushing seek) bumped the
            // generation; re-arm preroll and drain stale pre-seek frames until
            // the `Flush`, so the post-flush target is the new visible preroll.
            let gen = sc.preroll_generation();
            if gen != preroll_gen {
                preroll_gen = gen;
                prerolled_self = false;
                flushing = true;
            }
        }
        // M87 buffering: sample the input link's fill and post a `Buffering`
        // report when it crosses a quartile band. The first iteration samples
        // an as-yet-unfilled link, so a `bus` always sees at least one report.
        report_buffering(bus.as_ref(), &probe, &in_rx, &mut last_buffer_bucket);
        let packet = if control_open {
            match select2(arm_rx.recv(), in_rx.recv()).await {
                // M839: a producer upstream (a muxer) settled a new pool on this
                // sink's input link. Absorb it; a sink is the end of the walk, so
                // it has nothing to report onward.
                Either::Left(Some(directive)) => {
                    elem.configure_allocation(directive.params());
                    continue;
                }
                Either::Left(None) => {
                    control_open = false;
                    continue;
                }
                Either::Right(packet) => packet,
            }
        } else {
            in_rx.recv().await
        };
        match packet {
            // M360: discard stale pre-seek buffers while draining toward the
            // `Flush`; control packets fall through (the `Flush` ends drain).
            Some(PipelinePacket::DataFrame(_)) if flushing => continue,
            Some(PipelinePacket::Eos) => {
                elem.process(PipelinePacket::Eos, &mut null).await?;
                // Count this sink toward preroll only if it never took a real
                // preroll buffer (an empty stream). Without the guard a sink
                // that already prerolled double-decrements the shared counter
                // and completes the pipeline preroll prematurely.
                if !prerolled_self {
                    if let Some(sc) = &state {
                        sc.notify_prerolled();
                    }
                }
                return Ok(consumed);
            }
            Some(PipelinePacket::CapsChanged(new_caps)) => {
                // A sink that rejects the refined caps has no solution: no
                // runtime producer renegotiates its output caps, so a tee branch
                // (`FailLoud`) and a single-producer chain (`Reconfigure`) both
                // fail the run loud rather than flowing stale caps; `Drop` (a
                // tee) ends this branch while its siblings continue.
                let sink_caps =
                    match re_solve_downstream_dyn_sink(&new_caps, &*elem as &dyn DynAsyncElement) {
                        Ok(caps) => caps,
                        Err(failure) => match mode {
                            BranchMode::FailLoud | BranchMode::Reconfigure => {
                                report_runtime_caps_conflict(&new_caps);
                                report_nego_failure(bus.as_ref(), failure);
                                return Err(G2gError::CapsMismatch);
                            }
                            BranchMode::Drop => {
                                report_nego_failure(bus.as_ref(), failure);
                                return Ok(consumed);
                            }
                        },
                    };
                let instance = probe.as_deref().map(|p| p.name());
                log_caps_forward(instance, &new_caps, &sink_caps, true);
                match log_caps_rejected(instance, &sink_caps, elem.configure_pipeline(&sink_caps))?
                {
                    ConfigureOutcome::Accepted => {
                        let proposal = with_meta_demand(
                            elem.propose_allocation(&sink_caps),
                            elem.meta_requests(),
                        );
                        if let Some(p) = &proposal {
                            elem.configure_allocation(p);
                        }
                        coord
                            .report(Recascade {
                                node,
                                route: RecascadeRoute::Upstream,
                                proposal,
                            })
                            .await;
                        elem.process(PipelinePacket::CapsChanged(sink_caps), &mut null)
                            .await?;
                    }
                    ConfigureOutcome::ReFixate(counter) => {
                        in_rx.request_reconfigure(Reconfigure::Propose(counter));
                    }
                }
            }
            // M360: the `Flush` ends the re-preroll drain; the next (post-flush)
            // DataFrame becomes the new visible preroll.
            Some(PipelinePacket::Flush) => {
                flushing = false;
                elem.process(PipelinePacket::Flush, &mut null).await?;
            }
            Some(packet) => {
                // M203: follow the segment and publish each buffer's stream-time
                // position, so an application POSITION poll is answerable. The
                // sink is the position authority, as in GStreamer (segment + last
                // buffer). Inspect before `process` moves the packet.
                match &packet {
                    PipelinePacket::Segment(seg) => current_segment = Some(*seg),
                    PipelinePacket::DataFrame(frame) => {
                        if let Some(p) = &progress {
                            let pts = frame.timing.pts_ns;
                            let pos = current_segment
                                .as_ref()
                                .and_then(|s| s.to_stream_time(pts))
                                .unwrap_or(pts);
                            p.set_position(pos);
                        }
                    }
                    _ => {}
                }
                let seq = match &packet {
                    PipelinePacket::DataFrame(f) => Some(f.sequence),
                    _ => None,
                };
                let is_buffer = seq.is_some();
                if is_buffer {
                    consumed += 1;
                }
                // M399: time the data-frame `process()` and sample input fill.
                let timed = probe.as_deref().filter(|_| is_buffer);
                let mut wait_ns = 0;
                if let Some(p) = timed {
                    p.record_fill(in_rx.fill_percent());
                    // Queue-residency of this frame on the input link (M684).
                    if let Some(t) = in_rx.pop_transit_ns() {
                        p.record_transit(t);
                        wait_ns = t;
                    }
                }
                // M882: sample the animated properties at this frame's PTS.
                apply_control(
                    control.as_ref(),
                    &mut *elem as &mut dyn DynAsyncElement,
                    &packet,
                    &probe,
                )?;
                let t0 = ElementProbe::mark();
                elem.process(packet, &mut null).await?;
                if let (Some(p), Some(seq)) = (timed, seq) {
                    let push_wait_ns = p.record_proc_since(t0);
                    // M851: the last hop of a frame's journey. A sink pushes
                    // nowhere, so the banked wait is always 0 here.
                    p.record_visit(seq, wait_ns, t0, push_wait_ns);
                }
                // M175 upstream QoS: a sink that dropped a late frame asks to
                // shed load; store its report on this sink's input link, where
                // the upstream transform relays it one hop further (or the source
                // observes it directly as `PushOutcome::Qos`).
                if let Some(qos) = elem.take_qos() {
                    in_rx.request_qos(qos);
                }
                // Keyframe-request / renegotiation a sink originates (WebRTC PLI);
                // store it on the input link, where the upstream encoder/transform
                // observes it as `PushOutcome::Reconfigure`.
                if let Some(reconf) = elem.take_reconfigure() {
                    in_rx.request_reconfigure(reconf);
                }
                // Target bitrate (WebRTC BWE) up the reverse channel to the encoder.
                if let Some(bps) = elem.take_bitrate() {
                    in_rx.request_bitrate(bps);
                }
                // M78: the first buffer in non-live `Paused` is this sink's
                // preroll frame; mark this arm prerolled so the gate flips to a
                // hold, and report it so the pipeline preroll aggregates toward
                // a single `AsyncDone`.
                if is_buffer && !prerolled_self {
                    prerolled_self = true;
                    if let Some(sc) = &state {
                        sc.notify_prerolled();
                    }
                }
            }
            None => return Ok(consumed),
        }
    }
}

async fn tee_arm(
    in_rx: LinkReceiver,
    out_txs: Vec<LinkSender>,
    branch_drop: bool,
) -> Result<u64, G2gError> {
    let mut senders: Vec<SenderSink> = out_txs.into_iter().map(SenderSink::new).collect();
    loop {
        match in_rx.recv().await {
            Some(PipelinePacket::Eos) => {
                for s in senders.iter_mut() {
                    match s.push(PipelinePacket::Eos).await {
                        Ok(_) => {}
                        // A dropped branch (`AllowBranchDrop`) has closed its
                        // channel; skip it. Under `FailLoud` a closed branch is a
                        // genuine error and propagates.
                        Err(G2gError::Shutdown) if branch_drop => {}
                        Err(e) => return Err(e),
                    }
                }
                return Ok(0);
            }
            Some(packet) => {
                if branch_drop {
                    broadcast_drop_closed(&mut senders, packet).await?;
                    // Every branch has dropped: the fan-out has no consumers left,
                    // so this tee is done.
                    if senders.is_empty() {
                        return Ok(0);
                    }
                } else {
                    broadcast(&mut senders, packet).await?;
                }
            }
            None => return Ok(0),
        }
    }
}

/// Broadcast like [`broadcast`], but a branch whose receiver has closed
/// (a dropped `AllowBranchDrop` branch) is removed from `senders` instead of
/// failing the fan-out. A genuine downstream error still surfaces through that
/// branch arm's own result, so swallowing the closed channel here is safe.
async fn broadcast_drop_closed(
    senders: &mut Vec<SenderSink>,
    mut packet: PipelinePacket,
) -> Result<(), G2gError> {
    if let PipelinePacket::DataFrame(frame) = &mut packet {
        frame.domain.make_shareable();
    }
    let mut dead: Vec<usize> = Vec::new();
    for (i, s) in senders.iter_mut().enumerate() {
        match s.push(try_clone_packet(&packet)?).await {
            Ok(_) => {}
            Err(G2gError::Shutdown) => dead.push(i),
            Err(e) => return Err(e),
        }
    }
    // Remove dead senders high-index-first so earlier indices stay valid.
    for &i in dead.iter().rev() {
        senders.remove(i);
    }
    Ok(())
}

/// Everything a demux arm needs besides its element (M1009), the fan-out mirror
/// of [`TransformArmIo`]. Opaque on purpose: only the runner can build one, so
/// the `drive_demux_arm` hook stays implementable only through the blanket impl.
#[doc(hidden)]
#[allow(missing_debug_implementations)]
pub struct DemuxArmIo {
    pub(crate) in_rx: LinkReceiver,
    pub(crate) out_txs: Vec<LinkSender>,
    pub(crate) probe: Probe,
}

/// The demux arm: drain the single input edge and let the routing element
/// dispatch each packet to a chosen output port (the transpose of `muxer_arm`).
/// Mirrors the `run_source_fanout` router loop: a packet goes to
/// `MultiOutputElement::process`, which calls `push_to(port, ..)`; on `Eos` the
/// element flushes first, then the arm closes every branch with its own `Eos`
/// (the runner owns the per-branch end, like the tee arm).
///
/// Monomorphized over the element type by the `drive_demux_arm` blanket hook
/// (M1009), so the per-packet `process` future is the element's own unboxed
/// state machine; see [`transform_arm`].
pub(crate) async fn demux_arm<E: MultiOutputElement>(
    mut demux: E,
    io: DemuxArmIo,
) -> Result<u64, G2gError> {
    let DemuxArmIo {
        in_rx,
        out_txs,
        probe,
    } = io;
    let branch_count = out_txs.len();
    let senders: Vec<SenderSink> = out_txs.into_iter().map(SenderSink::new).collect();
    let mut multi = MultiSenderSink::new(senders);
    multi.set_push_wait_probe(probe.clone());
    loop {
        match in_rx.recv().await {
            Some(PipelinePacket::Eos) => {
                demux.process(PipelinePacket::Eos, &mut multi).await?;
                for port in 0..branch_count {
                    multi.push_to(port, PipelinePacket::Eos).await?;
                }
                return Ok(0);
            }
            Some(packet) => {
                // M694: time the data-frame `process()` and sample input fill;
                // control packets are excluded so the histogram reflects real work.
                let timed = probe
                    .as_deref()
                    .filter(|_| matches!(&packet, PipelinePacket::DataFrame(_)));
                if let Some(p) = timed {
                    p.record_fill(in_rx.fill_percent());
                    if let Some(t) = in_rx.pop_transit_ns() {
                        p.record_transit(t);
                    }
                }
                let t0 = ElementProbe::mark();
                demux.process(packet, &mut multi).await?;
                if let Some(p) = timed {
                    p.record_proc_since(t0);
                }
            }
            None => return Ok(0),
        }
    }
}

/// Send `packet` to every tee branch. The frame's memory is made shareable once
/// (a zero-copy refcount handle, M250), so the per-branch clones below are
/// refcount bumps, not deep copies; the original is then moved into the last
/// branch. A fan-out of `n` makes zero byte copies of a `System` / GPU frame.
pub(crate) async fn broadcast(
    senders: &mut [SenderSink],
    mut packet: PipelinePacket,
) -> Result<(), G2gError> {
    if let PipelinePacket::DataFrame(frame) = &mut packet {
        frame.domain.make_shareable();
    }
    let last = senders.len() - 1;
    for s in senders[..last].iter_mut() {
        s.push(try_clone_packet(&packet)?).await?;
    }
    senders[last].push(packet).await?;
    Ok(())
}

/// One muxer input: drain its edge and tag every packet with its pad index for
/// the muxer arm. A per-input `Eos` (or a closed edge) is tagged so the muxer
/// arm can aggregate the single merged `Eos`.
async fn muxer_forwarder(
    in_rx: LinkReceiver,
    tagged: Sender<PipelinePacket>,
) -> Result<u64, G2gError> {
    loop {
        match in_rx.recv().await {
            Some(PipelinePacket::Eos) | None => {
                tagged
                    .send(PipelinePacket::Eos)
                    .await
                    .map_err(|_| G2gError::Shutdown)?;
                return Ok(0);
            }
            Some(packet) => {
                tagged.send(packet).await.map_err(|_| G2gError::Shutdown)?;
            }
        }
    }
}

/// A pad receiver a fan-in arm can round-robin over: either a muxer's per-pad
/// bounded [`Receiver`] or a terminal fan-in's input-edge [`LinkReceiver`]. Both
/// expose the same `RecvFuture`, so [`muxer_recv_any`] blocks on either kind.
trait PadReceiver {
    fn recv_packet(&self) -> RecvFuture<'_, PipelinePacket>;
}

impl PadReceiver for Receiver<PipelinePacket> {
    fn recv_packet(&self) -> RecvFuture<'_, PipelinePacket> {
        self.recv()
    }
}

impl PadReceiver for LinkReceiver {
    fn recv_packet(&self) -> RecvFuture<'_, PipelinePacket> {
        self.recv()
    }
}

/// Block until some open input pad delivers a packet (or closes), scanning
/// round-robin from `start` so the wake-up path stays fair too. Returns the
/// pad's slot index (into `pad_rxs`) and its packet; `None` means that pad's
/// channel closed without an `Eos` (an upstream error), treated as an end. All
/// receivers register the same task waker, so a push on any one wakes us.
async fn muxer_recv_any<R: PadReceiver>(
    pad_rxs: &[(usize, R)],
    open: &[bool],
    start: usize,
) -> (usize, Option<PipelinePacket>) {
    let n = pad_rxs.len();
    core::future::poll_fn(|cx| {
        for k in 0..n {
            let slot = (start + k) % n;
            if !open[slot] {
                continue;
            }
            // `RecvFuture` holds only a `&Receiver`, so it is `Unpin`; polling it
            // parks our waker on that channel when pending.
            let mut f = pad_rxs[slot].1.recv_packet();
            if let core::task::Poll::Ready(v) = core::future::Future::poll(Pin::new(&mut f), cx) {
                return core::task::Poll::Ready((slot, v));
            }
        }
        core::task::Poll::Pending
    })
    .await
}

/// Pair each of a terminal fan-in's input-edge receivers with its pad index, and
/// clone each pad's reverse-signal channel from the session before it moves into
/// the arm. `in_rxs` is in the node's `in_edges` order, so it aligns with the pad
/// indices read from those same edges.
fn fanin_sink_pads<'a>(
    vg: &ValidatedGraph<GraphNodeRef<'a>>,
    node: NodeId,
    in_rxs: Vec<LinkReceiver>,
    session: &dyn DynMultiInputElement,
) -> (Vec<(usize, LinkReceiver)>, Vec<Option<ReverseChannel>>) {
    let pads: Vec<usize> = vg
        .in_edges(node)
        .iter()
        .map(|&eid| vg.edge(eid).dst.index as usize)
        .collect();
    let reverse: Vec<Option<ReverseChannel>> = pads
        .iter()
        .map(|&pad| session.reverse_channel(pad))
        .collect();
    let pad_rxs: Vec<(usize, LinkReceiver)> = in_rxs
        .into_iter()
        .zip(pads)
        .map(|(rx, pad)| (pad, rx))
        .collect();
    (pad_rxs, reverse)
}

/// Relay each terminal fan-in input's pending reverse signal (WebRTC PLI / BWE)
/// onto that pad's edge, so it reaches the upstream encoder as a [`PushOutcome`]
/// on its next push, the same one-hop route a linked sink's reverse channel takes.
/// `reverse` is aligned with `pad_rxs` by slot.
fn relay_reverse(pad_rxs: &[(usize, LinkReceiver)], reverse: &[Option<ReverseChannel>]) {
    for (slot, (_pad, in_rx)) in pad_rxs.iter().enumerate() {
        let Some(rc) = &reverse[slot] else { continue };
        match rc.take() {
            Some(PushOutcome::Reconfigure(r)) => in_rx.request_reconfigure(r),
            Some(PushOutcome::Bitrate(bps)) => in_rx.request_bitrate(bps),
            _ => {}
        }
    }
}

/// Order a terminal fan-out source's output senders by PORT (out-edges arrive
/// in edge order; the port is each edge's `src.index`), wrapped as the
/// [`MultiOutputSink`] the source pushes into.
/// Order a demux node's output senders by each out-edge's source pad index, so
/// `push_to(port)` reaches the branch linked to that port. Out-edges arrive in
/// insertion order, which only matches the port order when the graph happened
/// to be linked port-first (launch lines do, builders need not).
fn demux_out_txs_by_port<'a>(
    vg: &ValidatedGraph<GraphNodeRef<'a>>,
    node: NodeId,
    out_txs: Vec<LinkSender>,
) -> Vec<LinkSender> {
    let mut indexed: Vec<(usize, LinkSender)> = vg
        .out_edges(node)
        .iter()
        .map(|&oe| vg.edge(oe).src.index as usize)
        .zip(out_txs)
        .collect();
    indexed.sort_by_key(|(port, _)| *port);
    indexed.into_iter().map(|(_, tx)| tx).collect()
}

fn fanout_src_ports<'a>(
    vg: &ValidatedGraph<GraphNodeRef<'a>>,
    node: NodeId,
    out_txs: Vec<LinkSender>,
) -> MultiSenderSink {
    let mut by_port: Vec<Option<SenderSink>> = (0..out_txs.len()).map(|_| None).collect();
    for (tx, &oe) in out_txs.into_iter().zip(vg.out_edges(node)) {
        let port = vg.edge(oe).src.index as usize;
        if let Some(slot) = by_port.get_mut(port) {
            *slot = Some(SenderSink::new(tx));
        }
    }
    // D1 validation guarantees each declared port is linked exactly once.
    MultiSenderSink::new(
        by_port
            .into_iter()
            .map(|s| s.expect("validated: every fan-out port linked"))
            .collect(),
    )
}

/// Drive a terminal fan-out source (a 0-in / N-out [`MultiOutputSource`], e.g.
/// a WebRTC session receiving several tracks) over the DAG's per-edge channels:
/// the element pushes to each port itself and owes every port an `Eos`
/// (its `run` contract), so the arm just runs it to completion.
async fn fanout_src_arm<'a>(
    mut source: Box<dyn DynMultiOutputSource + 'a>,
    mut sinks: MultiSenderSink,
) -> Result<u64, G2gError> {
    source.run(&mut sinks).await
}

/// Everything a terminal fan-in arm needs besides its element (M1009).
#[doc(hidden)]
#[allow(missing_debug_implementations)]
pub struct FaninSinkArmIo {
    pub(crate) pad_rxs: Vec<(usize, LinkReceiver)>,
    pub(crate) reverse: Vec<Option<ReverseChannel>>,
    pub(crate) probe: Probe,
}

/// Drive a terminal fan-in element (an N-input [`MultiInputElement`] with no
/// downstream output, e.g. a WebRTC session) the `run_fanin_session` way but over
/// the DAG's per-edge channels. Drains the input edges round-robin (fairness, so a
/// fast layer cannot starve a slow one), calls `session.process(pad, ..)` serially,
/// delivers a per-input `Eos` so the session can flush that track, and ends once
/// every input has ended. Per-input reverse signals (`session.reverse_channel(pad)`)
/// are relayed onto each pad's edge, so a per-layer PLI reaches the encoder feeding
/// it (`Reconfigure::ForceKeyframe` through that arm), the analog of the
/// `TaggingSink` reverse routing in the standalone fan-in session runner.
///
/// Monomorphized over the element type by the `drive_fanin_sink_arm` hook
/// (M1009), so the per-packet `process` future is unboxed; see [`transform_arm`].
pub(crate) async fn fanin_sink_arm<E: MultiInputElement>(
    mut session: E,
    io: FaninSinkArmIo,
) -> Result<u64, G2gError> {
    let FaninSinkArmIo {
        pad_rxs,
        reverse,
        probe,
    } = io;
    let mut null = NullSink;
    let input_count = pad_rxs.len();
    let mut open = alloc::vec![true; input_count];
    let mut ended = 0usize;
    let mut consumed = 0u64;
    let mut next = 0usize;
    loop {
        relay_reverse(&pad_rxs, &reverse);
        // Round-robin try-drain across the input edges, then block on any.
        let mut picked: Option<(usize, PipelinePacket)> = None;
        for k in 0..input_count {
            let slot = (next + k) % input_count;
            if !open[slot] {
                continue;
            }
            if let Some(pkt) = pad_rxs[slot].1.try_recv() {
                picked = Some((slot, pkt));
                next = (slot + 1) % input_count;
                break;
            }
        }
        let (slot, packet) = match picked {
            Some(p) => p,
            None => {
                if !open.iter().any(|&o| o) {
                    return Ok(consumed);
                }
                let (slot, maybe) = muxer_recv_any(&pad_rxs, &open, next).await;
                next = (slot + 1) % input_count;
                // A closed channel with no `Eos` is an upstream end.
                (slot, maybe.unwrap_or(PipelinePacket::Eos))
            }
        };
        let pad = pad_rxs[slot].0;
        match packet {
            PipelinePacket::Eos => {
                session.process(pad, PipelinePacket::Eos, &mut null).await?;
                open[slot] = false;
                ended += 1;
                if ended == input_count {
                    return Ok(consumed);
                }
            }
            PipelinePacket::CapsChanged(new_caps) => {
                // Mid-stream re-solve (M724): re-configure the changed pad
                // before the session sees the new caps. A counter-fixation
                // travels back up this pad's edge like a sink's would.
                let instance = probe.as_deref().map(|p| p.name());
                match log_caps_rejected(
                    instance,
                    &new_caps,
                    session.configure_pipeline(pad, &new_caps),
                )? {
                    ConfigureOutcome::Accepted => {
                        session
                            .process(pad, PipelinePacket::CapsChanged(new_caps), &mut null)
                            .await?;
                    }
                    ConfigureOutcome::ReFixate(counter) => {
                        pad_rxs[slot]
                            .1
                            .request_reconfigure(Reconfigure::Propose(counter));
                    }
                }
            }
            packet => {
                let is_frame = matches!(packet, PipelinePacket::DataFrame(_));
                let timed = probe.as_deref().filter(|_| is_frame);
                if let Some(p) = timed {
                    p.record_fill(pad_rxs[slot].1.fill_percent());
                }
                if is_frame {
                    consumed += 1;
                }
                let t0 = ElementProbe::mark();
                session.process(pad, packet, &mut null).await?;
                if let Some(p) = timed {
                    p.record_proc_since(t0);
                }
                // The session may have raised a reverse signal for this track
                // during `process`; relay it up its pad now.
                relay_reverse(&pad_rxs, &reverse);
            }
        }
    }
}

/// Maximum times a muxer may re-derive its merged-output allocation while
/// settling one mid-stream change (M839). Each round re-derives the output pool,
/// absorbs it, and re-cascades the input pads whose demand moved. Well-behaved
/// elements settle in a couple of rounds (the last confirms nothing moved); a
/// mutually-constraining pair of pads that keeps flipping the output never
/// settles, so this stops it with `AllocationConflict` instead of looping
/// forever. Reset on each new stimulus (a pad `CapsChanged`, a consumer demand).
const MAX_RECASCADE_ROUNDS: u32 = 8;

/// One input pad's β state inside a muxer arm.
#[derive(Debug)]
struct MuxPad {
    /// This slot's pad index on the element (`pad_rxs` order).
    pad: usize,
    /// The arm feeding this pad's branch, `None` when a source feeds it directly
    /// (nothing interruptible to re-cascade into).
    upstream: Option<NodeId>,
    /// The pad's negotiated caps, refreshed by the MX-1 re-solve.
    caps: Caps,
    /// The demand last cascaded up this branch, so an unchanged one is not
    /// re-sent.
    alloc: Option<AllocationParams>,
}

/// M839: walks a β allocation change *through* a multi-input boundary.
///
/// A muxer is where the node-keyed walk used to stop: a change arriving on one
/// input pad has no single continuation, since the merged output is derived from
/// every pad and any pool it settles on can constrain the others. This carries the
/// per-pad state that gives it one. A change on a pad (a mid-stream `CapsChanged`,
/// or a consumer's demand re-cascading into the output) moves that pad's demand;
/// the merged-output pool is then re-derived from all pads, absorbed by the
/// element, and pushed to the consumer arms; the pads are re-queried and only
/// those whose demand actually moved re-cascade up their own branch.
///
/// Convergence: the output derivation is walked to a fixed point, and `rounds`
/// bounds it at [`MAX_RECASCADE_ROUNDS`] per stimulus. A well-behaved element
/// settles in a couple of rounds (the last confirms nothing moved); a mutually-
/// constraining pair of pads that keeps flipping the output fails loud with
/// `AllocationConflict` instead of looping forever.
#[derive(Debug)]
pub(crate) struct MuxBeta {
    node: NodeId,
    coord: GraphCoordHandle,
    pads: Vec<MuxPad>,
    /// Arms reading the merged output, resolved through structural tees.
    consumers: Vec<NodeId>,
    /// The pool last derived for the merged output.
    out_alloc: Option<AllocationParams>,
    /// Output re-derivations since the last new stimulus.
    rounds: u32,
}

impl MuxBeta {
    /// Baseline from the startup solve: what the muxer proposes per pad (exactly
    /// what the reverse-topo cascade stored on each input edge) and for its
    /// output, so the first mid-stream round re-cascades only what moved.
    fn seed(&mut self, mux: &(dyn DynMultiInputElement + '_), out_caps: &Caps) {
        for p in &mut self.pads {
            p.alloc = mux.propose_allocation_for_input(p.pad, &p.caps);
        }
        self.out_alloc = mux.propose_allocation_for_output(out_caps);
    }

    /// A mid-stream `CapsChanged` fixated new caps on one pad.
    async fn pad_changed<'m>(
        &mut self,
        mux: &'m mut (dyn DynMultiInputElement + 'm),
        slot: usize,
        caps: Caps,
        out_caps: &Caps,
    ) -> Result<(), G2gError> {
        self.pads[slot].caps = caps;
        self.rounds = 0;
        self.refresh_pads(mux).await;
        self.settle_output(mux, out_caps).await
    }

    /// A consumer's demand re-cascaded into the merged output: absorb it, then
    /// carry it across the boundary onto the input pads.
    async fn output_demand<'m>(
        &mut self,
        mux: &'m mut (dyn DynMultiInputElement + 'm),
        params: AllocationParams,
        out_caps: &Caps,
    ) -> Result<(), G2gError> {
        self.rounds = 0;
        mux.configure_allocation_for_output(&params);
        self.refresh_pads(mux).await;
        self.settle_output(mux, out_caps).await
    }

    /// Re-query every pad's demand and re-cascade the ones that moved, so an
    /// input the change does not reach keeps the allocation it already has.
    async fn refresh_pads(&mut self, mux: &(dyn DynMultiInputElement + '_)) {
        for i in 0..self.pads.len() {
            let want = mux.propose_allocation_for_input(self.pads[i].pad, &self.pads[i].caps);
            if want == self.pads[i].alloc {
                continue;
            }
            self.pads[i].alloc = want;
            if let (Some(target), Some(p)) = (self.pads[i].upstream, want) {
                self.coord
                    .report(Recascade {
                        node: self.node,
                        route: RecascadeRoute::Pad(target),
                        proposal: Some(p),
                    })
                    .await;
            }
        }
    }

    /// Re-derive the merged-output pool until it stops moving, absorbing each
    /// answer (so the pads can react to it) and telling the consumer arms.
    async fn settle_output<'m>(
        &mut self,
        mux: &'m mut (dyn DynMultiInputElement + 'm),
        out_caps: &Caps,
    ) -> Result<(), G2gError> {
        loop {
            let derived = mux.propose_allocation_for_output(out_caps);
            if derived == self.out_alloc {
                return Ok(());
            }
            self.rounds += 1;
            if self.rounds > MAX_RECASCADE_ROUNDS {
                return Err(G2gError::AllocationConflict);
            }
            self.out_alloc = derived;
            if let Some(p) = derived {
                mux.configure_allocation_for_output(&p);
                for &c in &self.consumers {
                    self.coord
                        .report(Recascade {
                            node: self.node,
                            route: RecascadeRoute::Consumer(c),
                            proposal: Some(p),
                        })
                        .await;
                }
            }
            self.refresh_pads(mux).await;
        }
    }
}

/// Apply a β directive that reached a muxer arm (M839): a consumer's demand on
/// the merged output, which crosses the boundary onto the input pads. A muxer is
/// never a `ProducerAllocation` target, since that notice names one input pad and
/// the downstream-facing route cannot express which.
async fn apply_mux_directive<'m>(
    mux: &'m mut (dyn DynMultiInputElement + 'm),
    beta: &mut MuxBeta,
    directive: ArmDirective,
    out_caps: &Caps,
) -> Result<(), G2gError> {
    match directive {
        ArmDirective::Recascade(p) => beta.output_demand(mux, p, out_caps).await,
        ArmDirective::ProducerAllocation(_) => Ok(()),
    }
}

/// A fan-in arm's deadline tick (M875): the clock to sleep on, the period the
/// element declared via
/// [`tick_interval_ns`](DynMultiInputElement::tick_interval_ns), and the next
/// deadline. Both [`muxer_arm`] and [`muxer_arm_pts`] drive it the same way, so
/// they share it: [`fire_if_due`](Self::fire_if_due) once per loop iteration for
/// the busy case, [`park_or_tick`] for the parked one.
struct ArmTick<'a> {
    clock: &'a dyn DynAsyncClock,
    period_ns: u64,
    next_ns: u64,
}

impl<'a> ArmTick<'a> {
    /// Ticking needs both halves: a clock to sleep on and an element that asked.
    fn new(
        ticker: Option<&'a dyn DynAsyncClock>,
        mux: &(dyn DynMultiInputElement + '_),
    ) -> Option<Self> {
        let (clock, period_ns) = ticker.zip(mux.tick_interval_ns())?;
        Some(Self {
            clock,
            period_ns,
            next_ns: clock.now_ns().saturating_add(period_ns),
        })
    }

    /// Deliver one tick, then snap the next deadline forward from now rather than
    /// adding a period to a deadline already in the past, so a slow element does
    /// not build up a backlog of ticks to fire.
    async fn fire<E: MultiInputElement>(
        &mut self,
        mux: &mut E,
        out: &mut dyn OutputSink,
    ) -> Result<(), G2gError> {
        mux.process(0, PipelinePacket::Tick, out).await?;
        self.next_ns = self.clock.now_ns().saturating_add(self.period_ns);
        Ok(())
    }

    /// The busy path: a live pad can spin an arm's drain loop forever without
    /// ever parking below, so the deadline is checked per iteration too.
    async fn fire_if_due<E: MultiInputElement>(
        &mut self,
        mux: &mut E,
        out: &mut dyn OutputSink,
    ) -> Result<(), G2gError> {
        if self.clock.now_ns() >= self.next_ns {
            self.fire(mux, out).await?;
        }
        Ok(())
    }
}

/// Race an arm's parked wait (data + β control) against its tick deadline, so a
/// fan-in whose inputs have all gone quiet still gets its tick. `None` means the
/// deadline won and a tick fired, so the caller loops. Data and control stay
/// biased ahead of the tick: a packet already waiting is handled first.
async fn park_or_tick<T, E: MultiInputElement>(
    park: impl core::future::Future<Output = T>,
    tick: Option<&mut ArmTick<'_>>,
    mux: &mut E,
    out: &mut dyn OutputSink,
) -> Result<Option<T>, G2gError> {
    let Some(tick) = tick else {
        return Ok(Some(park.await));
    };
    // The sleep rides the copied clock reference, not a borrow of `tick`, so
    // `fire` can still take it mutably below.
    let sleep = tick.clock.sleep_until_ns(tick.next_ns);
    match select2(park, sleep).await {
        Either::Left(v) => Ok(Some(v)),
        Either::Right(()) => {
            tick.fire(mux, out).await?;
            Ok(None)
        }
    }
}

/// Everything a muxer arm needs besides its element and its ticker (M1009).
/// Split from the ticker so the cooperative arm (which borrows the graph's
/// clock) and the thread-per-arm one (which owns a shared handle) carry the
/// same payload.
#[allow(missing_debug_implementations)]
pub(crate) struct MuxerArmParts {
    pub(crate) pad_rxs: Vec<(usize, Receiver<PipelinePacket>)>,
    pub(crate) out_tx: LinkSender,
    pub(crate) input_count: usize,
    pub(crate) current_output: Caps,
    pub(crate) beta: MuxBeta,
    pub(crate) arm_rx: Receiver<ArmDirective>,
    pub(crate) probe: Probe,
    pub(crate) control: Option<ArmController>,
}

/// A muxer arm's input, as the cooperative runner builds it. Opaque on purpose,
/// like [`TransformArmIo`]: only the runner can build one, so `drive_muxer_arm`
/// stays implementable only through the blanket impl.
#[doc(hidden)]
#[allow(missing_debug_implementations)]
pub struct MuxerArmIo<'a> {
    pub(crate) parts: MuxerArmParts,
    pub(crate) ticker: Option<&'a dyn DynAsyncClock>,
}

/// As [`MuxerArmIo`], for the thread-per-arm runner: a builder closure that
/// crosses onto a worker thread must own everything it carries, so the ticker
/// is the shared handle rather than a borrow (M879).
#[cfg(all(feature = "std", feature = "multi-thread"))]
#[doc(hidden)]
#[allow(missing_debug_implementations)]
pub struct MuxerArmOwnedTickIo {
    pub(crate) parts: MuxerArmParts,
    pub(crate) ticker: Option<alloc::sync::Arc<dyn DynAsyncClock + Send + Sync>>,
}

/// The muxer arm: drain the per-input channels round-robin, combine each input's
/// packets via `process(pad, ..)`, and emit a single `Eos` once every input has
/// ended. Round-robin draining keeps a fast input from starving a slow one (a
/// frozen overlay and a hung EOS aggregation). The per-input `Eos` is delivered
/// to the element first (so a stateful muxer can flush) but the element must not
/// forward it; the runner owns the merged one.
///
/// M875: when `ticker` is set and the element declares a
/// [`tick_interval_ns`](DynMultiInputElement::tick_interval_ns), the arm also
/// delivers [`PipelinePacket::Tick`] on that period, so a fan-in element whose
/// output cadence is its own (zero-order-hold over a stalled pad) emits without a
/// packet arriving. The deadline is checked on both paths: once per loop iteration
/// for the busy case (packets flowing on one pad while another is stalled never
/// park the arm), and raced against the parked `recv` otherwise.
///
/// Monomorphized over the element type by the `drive_muxer_arm` blanket hook
/// (M1009), so the per-packet `process` future is the element's own unboxed
/// state machine; see [`transform_arm`].
pub(crate) async fn muxer_arm<E: MultiInputElement>(
    mut mux: E,
    io: MuxerArmIo<'_>,
) -> Result<u64, G2gError> {
    let MuxerArmIo {
        parts:
            MuxerArmParts {
                pad_rxs,
                out_tx,
                input_count,
                mut current_output,
                mut beta,
                arm_rx,
                probe,
                control,
            },
        ticker,
    } = io;
    let mut adapter = SenderSink::new(out_tx);
    adapter.set_push_wait_probe(probe.clone());
    let mut open = alloc::vec![true; input_count];
    let mut ended = 0usize;
    // Cursor for round-robin fairness across both the try-drain and block paths.
    let mut next = 0usize;
    let mut control_open = true;
    let mut tick = ArmTick::new(ticker, &mux as &dyn DynMultiInputElement);
    beta.seed(&mux as &dyn DynMultiInputElement, &current_output);
    loop {
        // M839: a consumer's demand on the merged output arrives here. Checked
        // before the data drain so a busy muxer still applies it promptly.
        while let Some(directive) = arm_rx.try_recv() {
            apply_mux_directive(
                &mut mux as &mut dyn DynMultiInputElement,
                &mut beta,
                directive,
                &current_output,
            )
            .await?;
        }
        // Deadline reached while packets keep this arm busy (a live overlay pad
        // spinning the drain loop below without ever parking).
        if let Some(t) = tick.as_mut() {
            t.fire_if_due(&mut mux, &mut adapter).await?;
        }
        // Take one buffered packet, scanning round-robin from `next`, so no
        // single input can monopolize the muxer while others have data waiting.
        let mut picked: Option<(usize, PipelinePacket)> = None;
        for k in 0..input_count {
            let slot = (next + k) % input_count;
            if !open[slot] {
                continue;
            }
            if let Some(pkt) = pad_rxs[slot].1.try_recv() {
                picked = Some((slot, pkt));
                next = (slot + 1) % input_count;
                break;
            }
        }
        let (slot, packet) = match picked {
            Some(p) => p,
            None => {
                if !open.iter().any(|&o| o) {
                    return Ok(0);
                }
                // Race the β control channel against the pads, so a directive
                // reaches this arm while it is parked waiting for data.
                let park = async {
                    if control_open {
                        select2(arm_rx.recv(), muxer_recv_any(&pad_rxs, &open, next)).await
                    } else {
                        Either::Right(muxer_recv_any(&pad_rxs, &open, next).await)
                    }
                };
                // M875: the tick deadline joins the race; a fired tick loops.
                let Some(parked) =
                    park_or_tick(park, tick.as_mut(), &mut mux, &mut adapter).await?
                else {
                    continue;
                };
                let (slot, maybe) = match parked {
                    Either::Left(Some(directive)) => {
                        apply_mux_directive(
                            &mut mux as &mut dyn DynMultiInputElement,
                            &mut beta,
                            directive,
                            &current_output,
                        )
                        .await?;
                        continue;
                    }
                    Either::Left(None) => {
                        control_open = false;
                        continue;
                    }
                    Either::Right(v) => v,
                };
                next = (slot + 1) % input_count;
                // A closed channel with no `Eos` is an upstream end; fold it into
                // the same end-of-input path so aggregation still completes.
                (slot, maybe.unwrap_or(PipelinePacket::Eos))
            }
        };
        let pad = pad_rxs[slot].0;
        match packet {
            PipelinePacket::Eos => {
                mux.process(pad, PipelinePacket::Eos, &mut adapter).await?;
                open[slot] = false;
                ended += 1;
                if ended == input_count {
                    adapter.push(PipelinePacket::Eos).await?;
                    return Ok(0);
                }
            }
            PipelinePacket::CapsChanged(new_caps) => {
                // MX-1: re-solve this input against its pad constraint and
                // reconfigure the pad; the input-side `CapsChanged` is consumed,
                // not forwarded as if it were the merged output.
                let input_caps =
                    solve_mux_input_dyn(&new_caps, &mux as &dyn DynMultiInputElement, pad)?;
                log_caps_rejected(
                    probe.as_deref().map(|p| p.name()),
                    &input_caps,
                    mux.configure_pipeline(pad, &input_caps),
                )?
                .reject_refixate()?;
                // MX-2: the per-input change may shift the merged output. Emit one
                // downstream `CapsChanged` only when it actually changed.
                let new_output = solve_mux_output_dyn(&mux as &dyn DynMultiInputElement)?;
                if new_output != current_output {
                    current_output = new_output.clone();
                    adapter
                        .push(PipelinePacket::CapsChanged(new_output))
                        .await?;
                }
                // MX-1β / M839: walk the allocation change through this boundary.
                beta.pad_changed(
                    &mut mux as &mut dyn DynMultiInputElement,
                    slot,
                    input_caps,
                    &current_output,
                )
                .await?;
            }
            packet => {
                // M694: time the data-frame `process()` and sample this pad's
                // input fill; the per-pad channel is a plain `Receiver`, so it has
                // no transit ring (transit stays empty for muxer pads).
                let timed = probe
                    .as_deref()
                    .filter(|_| matches!(&packet, PipelinePacket::DataFrame(_)));
                if let Some(p) = timed {
                    p.record_fill(pad_rxs[slot].1.fill_percent());
                }
                // M882: sample the animated properties at this frame's PTS.
                apply_control(
                    control.as_ref(),
                    &mut mux as &mut dyn DynMultiInputElement,
                    &packet,
                    &probe,
                )?;
                let t0 = ElementProbe::mark();
                mux.process(pad, packet, &mut adapter).await?;
                if let Some(p) = timed {
                    p.record_proc_since(t0);
                }
            }
        }
    }
}

/// The PTS-ordered muxer arm (the opt-in alternative to [`muxer_arm`], selected
/// by [`DynMultiInputElement::input_pts_ordered`]): buffer each input's
/// `DataFrame`s in an [`InputAggregator`] and release the globally-earliest-PTS
/// one only once every still-open input has a head queued, so `process(pad, ..)`
/// sees frames in non-decreasing PTS across all pads. The runner does the
/// time-ordered interleave a multi-camera grid / PTS-synchronized compositor
/// would otherwise hand-roll. `Eos` (per-input flush + aggregation) and
/// `CapsChanged` (MX-1 / MX-2) are handled as in [`muxer_arm`]; only `DataFrame`s
/// are reordered.
///
/// The deadline tick works as in [`muxer_arm`] (busy check per iteration plus the
/// parked race, via [`ArmTick`]), except that it comes after the merged-`Eos`
/// exit and after the aggregator's release round: an arm on its way out ticks no
/// more, and a tick never jumps ahead of a frame already safe to emit in PTS
/// order.
///
/// Monomorphized over the element type like [`muxer_arm`] (M1009).
pub(crate) async fn muxer_arm_pts<E: MultiInputElement>(
    mut mux: E,
    io: MuxerArmIo<'_>,
) -> Result<u64, G2gError> {
    let MuxerArmIo {
        parts:
            MuxerArmParts {
                pad_rxs,
                out_tx,
                input_count,
                mut current_output,
                mut beta,
                arm_rx,
                probe,
                control,
            },
        ticker,
    } = io;
    let mut adapter = SenderSink::new(out_tx);
    adapter.set_push_wait_probe(probe.clone());
    let mut open = alloc::vec![true; input_count];
    let mut agg: InputAggregator<Frame> = InputAggregator::new(input_count);
    // Round-robin wake cursor, so a fast input does not bias the block path.
    let mut next = 0usize;
    let mut control_open = true;
    let mut tick = ArmTick::new(ticker, &mux as &dyn DynMultiInputElement);
    beta.seed(&mux as &dyn DynMultiInputElement, &current_output);
    loop {
        // Release every frame now safe to emit, in global PTS order: the
        // aggregator yields the earliest only once every still-contributing input
        // has a head, so no later input can still deliver something earlier.
        while let Some((slot, frame)) = agg.take_earliest_by(|f| f.timing.pts_ns) {
            let pad = pad_rxs[slot].0;
            // M694: released frames are all DataFrames; time each `process()` and
            // sample this pad's input fill (plain `Receiver`, so no transit).
            if let Some(p) = probe.as_deref() {
                p.record_fill(pad_rxs[slot].1.fill_percent());
            }
            // M882: sampled in release (PTS) order, so the animation follows the
            // ordered stream rather than pad arrival.
            apply_control_at(
                control.as_ref(),
                &mut mux as &mut dyn DynMultiInputElement,
                frame.timing.pts_ns,
                &probe,
            )?;
            let t0 = ElementProbe::mark();
            mux.process(pad, PipelinePacket::DataFrame(frame), &mut adapter)
                .await?;
            if let Some(p) = probe.as_deref() {
                p.record_proc_since(t0);
            }
        }
        // Once every input has ended, the loop above has drained the aggregator
        // (ended+empty inputs drop out of the round); emit the single merged Eos.
        if !open.iter().any(|&o| o) {
            adapter.push(PipelinePacket::Eos).await?;
            return Ok(0);
        }
        // Deadline reached while frames keep arriving on one pad (the arm never
        // parks below).
        if let Some(t) = tick.as_mut() {
            t.fire_if_due(&mut mux, &mut adapter).await?;
        }
        // Make progress: block for the next packet from any still-open input,
        // racing the β control channel (M839) so a consumer's demand on the
        // merged output reaches this arm while it waits, and the tick deadline so
        // a quiet fan-in still gets its tick.
        let park = async {
            if control_open {
                select2(arm_rx.recv(), muxer_recv_any(&pad_rxs, &open, next)).await
            } else {
                Either::Right(muxer_recv_any(&pad_rxs, &open, next).await)
            }
        };
        let Some(parked) = park_or_tick(park, tick.as_mut(), &mut mux, &mut adapter).await? else {
            continue;
        };
        let (slot, maybe) = match parked {
            Either::Left(Some(directive)) => {
                apply_mux_directive(
                    &mut mux as &mut dyn DynMultiInputElement,
                    &mut beta,
                    directive,
                    &current_output,
                )
                .await?;
                continue;
            }
            Either::Left(None) => {
                control_open = false;
                continue;
            }
            Either::Right(v) => v,
        };
        next = (slot + 1) % input_count;
        let pad = pad_rxs[slot].0;
        // A closed channel with no `Eos` is an upstream end (as in `muxer_arm`).
        match maybe.unwrap_or(PipelinePacket::Eos) {
            PipelinePacket::DataFrame(frame) => agg.push(slot, frame),
            PipelinePacket::Eos => {
                mux.process(pad, PipelinePacket::Eos, &mut adapter).await?;
                open[slot] = false;
                agg.mark_ended(slot);
            }
            PipelinePacket::CapsChanged(new_caps) => {
                // MX-1 / MX-2 / MX-1β, identical to `muxer_arm`: re-solve this
                // input's pad, emit one downstream `CapsChanged` only when the
                // merged output actually shifts, and walk the allocation change
                // through the boundary.
                let input_caps =
                    solve_mux_input_dyn(&new_caps, &mux as &dyn DynMultiInputElement, pad)?;
                log_caps_rejected(
                    probe.as_deref().map(|p| p.name()),
                    &input_caps,
                    mux.configure_pipeline(pad, &input_caps),
                )?
                .reject_refixate()?;
                let new_output = solve_mux_output_dyn(&mux as &dyn DynMultiInputElement)?;
                if new_output != current_output {
                    current_output = new_output.clone();
                    adapter
                        .push(PipelinePacket::CapsChanged(new_output))
                        .await?;
                }
                beta.pad_changed(
                    &mut mux as &mut dyn DynMultiInputElement,
                    slot,
                    input_caps,
                    &current_output,
                )
                .await?;
            }
            // Every other packet reaches the element, as on the single-input
            // path and in `muxer_arm`: `Segment` and `Flush` are per-input
            // control a fan-in has to see. A fan-in that swallows the segment
            // leaves a paced sink downstream with no running-time mapping (the
            // compositor forwards its timing input's; a muxer whose timestamps
            // are already container-mapped ignores it).
            packet => {
                mux.process(pad, packet, &mut adapter).await?;
            }
        }
    }
}

/// Owning-ticker shim for the thread-per-arm runner (M879): both fan-in arms take
/// their ticker as a borrow, but a builder closure that crosses onto a worker
/// thread must own everything it carries. This future owns the shared clock for its
/// whole life and lends it to the arm it drives, so the arms' signatures stay as
/// the cooperative runner needs them. [`MultiInputElement::input_pts_ordered`]
/// picks the arm, the same choice `drive_muxer_arm` makes on the cooperative path.
#[cfg(all(feature = "std", feature = "multi-thread"))]
pub(crate) async fn muxer_arm_owned_tick<E: MultiInputElement>(
    mux: E,
    io: MuxerArmOwnedTickIo,
) -> Result<u64, G2gError> {
    let MuxerArmOwnedTickIo { parts, ticker } = io;
    let tick: Option<&dyn DynAsyncClock> = ticker.as_deref().map(|c| c as &dyn DynAsyncClock);
    let io = MuxerArmIo {
        parts,
        ticker: tick,
    };
    if MultiInputElement::input_pts_ordered(&mux) {
        muxer_arm_pts(mux, io).await
    } else {
        muxer_arm(mux, io).await
    }
}

/// Clone a packet for a tee branch (M213, M250). Control packets clone trivially;
/// a data frame's memory is shared via [`MemoryDomain::share`]: a zero-copy
/// refcount bump for the GPU domains, the shared-CPU `SystemView`, and (once
/// `broadcast` has called `make_shareable`) owned-CPU `System` bytes too. So a
/// GPU-decoded or CPU frame fans out to several consumers (eg inference +
/// display) with no copy, where `System` previously deep-copied per branch and a
/// GPU frame failed loud.
pub(crate) fn try_clone_packet(packet: &PipelinePacket) -> Result<PipelinePacket, G2gError> {
    Ok(match packet {
        PipelinePacket::CapsChanged(caps) => PipelinePacket::CapsChanged(caps.clone()),
        PipelinePacket::Eos => PipelinePacket::Eos,
        PipelinePacket::Flush => PipelinePacket::Flush,
        PipelinePacket::Segment(seg) => PipelinePacket::Segment(*seg),
        // Tee clone: shares the buffer where the domain allows (GPU handles /
        // pre-shared System bytes refcount, owned CPU bytes deep-copy) and shares
        // per-frame metadata by Arc refcount with copy-on-write on mutation, so a
        // detector branch and a video branch carry the same AnalyticsMeta without
        // aliasing. The frame-level fan-out primitive.
        PipelinePacket::DataFrame(frame) => PipelinePacket::DataFrame(frame.share()),
        // Runner-internal (a fan-in arm's deadline), so it is never on a link to
        // begin with; trivially cloneable all the same.
        PipelinePacket::Tick => PipelinePacket::Tick,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::frame::FrameTiming;
    use crate::memory::{MemoryDomain, SystemSlice};

    fn system_frame(bytes: &[u8], seq: u64) -> PipelinePacket {
        PipelinePacket::DataFrame(Frame {
            domain: MemoryDomain::System(SystemSlice::from_boxed(
                bytes.to_vec().into_boxed_slice(),
            )),
            timing: FrameTiming {
                pts_ns: 7,
                ..FrameTiming::default()
            },
            sequence: seq,
            meta: Default::default(),
        })
    }

    #[test]
    fn buffering_bucket_bands_fill_into_quartiles() {
        // Empty, quarter steps, and full map to distinct bands; values within a
        // band collapse so the sink posts only on a level transition.
        assert_eq!(buffering_bucket(0), 0);
        assert_eq!(buffering_bucket(24), 0);
        assert_eq!(buffering_bucket(25), 1);
        assert_eq!(buffering_bucket(50), 2);
        assert_eq!(buffering_bucket(75), 3);
        assert_eq!(buffering_bucket(100), 4);
    }

    #[test]
    fn clones_system_frame_bytes_and_timing() {
        let original = system_frame(&[1, 2, 3, 4], 9);
        let cloned = try_clone_packet(&original).expect("system frame clones");
        let (PipelinePacket::DataFrame(a), PipelinePacket::DataFrame(b)) = (&original, &cloned)
        else {
            panic!("expected data frames");
        };
        let Some(sa) = a.domain.as_system_slice() else {
            panic!()
        };
        let Some(sb) = b.domain.as_system_slice() else {
            panic!()
        };
        assert_eq!(sa, sb, "bytes copied");
        assert_ne!(sb.as_ptr(), sa.as_ptr(), "distinct allocation");
        assert_eq!(b.timing, a.timing);
        assert_eq!(b.sequence, 9);
    }

    #[test]
    fn clones_control_packets() {
        assert!(matches!(
            try_clone_packet(&PipelinePacket::Eos),
            Ok(PipelinePacket::Eos)
        ));
        assert!(matches!(
            try_clone_packet(&PipelinePacket::Flush),
            Ok(PipelinePacket::Flush)
        ));
    }

    #[cfg(feature = "metadata")]
    #[test]
    fn tee_clone_carries_analytics_meta() {
        use crate::meta::{AnalyticsMeta, BBox, ObjectDetection};
        // A detector attaches analytics; the tee clone must carry it onto the
        // sibling (video) branch so a downstream overlay can read it.
        let PipelinePacket::DataFrame(mut original) = system_frame(&[0, 0, 0, 0], 1) else {
            panic!("data frame");
        };
        let mut a = AnalyticsMeta::new();
        a.add_detection(ObjectDetection {
            bbox: BBox {
                x: 0.1,
                y: 0.1,
                w: 0.2,
                h: 0.2,
            },
            label: 5,
            confidence: 0.9,
        });
        original.meta.attach(a);

        let cloned = try_clone_packet(&PipelinePacket::DataFrame(original)).expect("clone");
        let PipelinePacket::DataFrame(b) = cloned else {
            panic!("data frame")
        };
        let meta = b
            .meta
            .get::<AnalyticsMeta>()
            .expect("meta carried to tee branch");
        assert_eq!(meta.detections().count(), 1);
        assert_eq!(meta.detections().next().unwrap().label, 5);
    }
}