g2g_core/runtime/

launch.rs

//! `gst-launch`-style text pipeline parser (M106, M117, M118): turn
//! `"videotestsrc num-buffers=3 ! videoflip method=rotate-180 ! fakesink"` into a
//! runnable [`Graph`], the front door that makes g2g usable without hand-writing
//! Rust for every pipeline.
//!
//! Built on the M104 property system and the M105 by-name registry: each `!`
//! separated node is `element-name key=value ...`; the parser constructs the
//! element by name from the [`Registry`], looks up each property's
//! [`PropKind`](crate::PropKind) to parse its textual value, and applies it.
//! Roles follow connectivity: an element with no incoming link is a source, one
//! with no outgoing link a sink, the rest transforms (so a linear chain is still
//! source -> transforms -> sink).
//!
//! Branching (M118): `tee name=t` fans one output to many. A `tee` is the
//! structural fan-out node (no element), its output width derived from how many
//! branches reference it; a branch is a `t.` pad reference that starts a chain (a
//! head ref, linking *from* the named element) or, right after a `!`, ends one (a
//! tail ref, linking *into* it). So
//! `videotestsrc ! tee name=t ! fakesink   t. ! fakesink` broadcasts each frame
//! to two sinks. The caps shorthand (`! video/x-raw,format=nv12,... !`, M117) is
//! a bare media-type node rewritten to a `capsfilter`.
//!
//! Fan-in (M122): an element with several inbound links is a muxer, built from
//! the registry's [`MuxerFactory`](crate::runtime::MuxerFactory) with that input
//! count (so its `input_count` matches the node's pads). Each feeding chain ends
//! with a `m.` tail ref, so
//! `src1 ! m.   src2 ! m.   funnel name=m ! fakesink` joins two streams. Feeding
//! chains come first (a new chain can only begin after a `!` / ref / caps
//! boundary, so a chain starting with a bare element name would be read as the
//! previous element's property); the muxer chain is last. A muxer has one output
//! pad, so it must feed a downstream consumer.
//!
//! Value grammar (M840): a value may carry spaces by quoting (`"..."` or
//! `'...'`, either anywhere in the fragment: `location=/tmp/"my dir"/a.ts`) or by
//! escaping (`\ `). A `\` escapes a quote, another `\`, a `!`, a `#`, or
//! whitespace; before anything else it stays literal, so a Windows path
//! (`C:\videos\a.ts`) survives unescaped. An enum property's value is checked
//! against its [`PropertySpec::enum_values`](crate::PropertySpec) before the
//! element sees it, and a [`Flags`](crate::PropKind::Flags) property takes a
//! `+`-joined set (`protocols=udp+tcp`), so a bad name names the valid ones.
//! A pad-name suffix on a reference (`t.src_0`) is accepted but ignored (pads are
//! positional).
//!
//! Encoding profiles (M1089): `encodebin profile="<container>:<stream>[:<stream>]"`
//! is a macro too, expanding into the encoder each stream names plus the muxer
//! for the container (`encodebin2` and `transcodebin` are gst's other names for
//! it, the latter being `decodebin ! encodebin`). The container picks the muxer
//! and a stream picks its encoder; anything the profile does not name
//! negotiates. A stream part may also pin what it does care about (M1097): a
//! `width` / `height`, a `framerate`, a sample `rate` or a `channels` each
//! splice the converter that applies it, and a `bitrate` is set on the encoder.
//! Any other field is refused. An uncompressed stream
//! (`audio/x-raw`) inserts no encoder, and a profile with no container at all is
//! just the encoder. A converter goes in ahead of an encoder that does not take
//! what the branch produces (`videoconvert`, or `audioconvert` +
//! `audioresample`), so a profile works off whatever raw form the source has. With several streams the bin's `name=` moves to the muxer,
//! so a second branch reaches it as `e.` the way any fan-in does, and that
//! branch gets the encoder for its own kind of input.
//!
//! Two `key=value` pairs are launch keywords rather than properties: `name=` is
//! the instance name (and the handle pad references resolve against), and
//! `log-category=` (M847) replaces that instance's `G2G_DEBUG` category, leaving
//! the auto `<type>N` naming alone.

use alloc::boxed::Box;
use alloc::string::{String, ToString};
use alloc::vec::Vec;

use crate::caps::{Caps, CapsSet};
use crate::element::DynAsyncElement;
use crate::graph::{Graph, GraphError, NodeId, PadId};
use crate::link::LinkPolicy;
use crate::memory::MemoryDomainKind;
use crate::property::{PropError, PropValue, PropertySpec, ValueError};
use crate::runtime::autoplug::{
    is_raw_audio, is_raw_video, PadKind, PadRequest, Registry, UriError,
};
use crate::runtime::{DynSourceLoop, GraphNode, GraphNodeRef};

/// Why [`parse_launch`] could not build a graph.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ParseError {
    /// The pipeline string was empty or all whitespace.
    Empty,
    /// A node between `!` separators had no element name.
    EmptyStage,
    /// Fewer than two elements: a runnable pipeline needs at least a source and a
    /// sink.
    TooFewStages,
    /// A source-position element names no registered source.
    UnknownSource(String),
    /// A transform / sink-position element names no registered element.
    UnknownElement(String),
    /// A property token had no `=` (expected `key=value`).
    MalformedProperty { element: String, token: String },
    /// The element has no property of that name.
    UnknownProperty { element: String, key: String },
    /// The value did not parse for the property's kind, or was rejected.
    BadValue {
        element: String,
        key: String,
        value: String,
    },
    /// The value names no declared choice of an enum / flag property. `value` is
    /// the offending nick (one entry of a `+`-joined set), or the whole value when
    /// the set was malformed; `values` is the property's declared list.
    BadEnumValue {
        element: String,
        key: String,
        value: String,
        values: &'static str,
    },
    /// A `name.` reference names no element declared with that `name=`.
    UnknownReference(String),
    /// Two elements share the same `name=` handle.
    DuplicateName(String),
    /// More than one link feeds an element's input, but it names no registered
    /// muxer: fan-in needs a [`MuxerFactory`](crate::runtime::MuxerFactory).
    NotAMuxer(String),
    /// A muxer (an element with several inputs) has no outgoing link; its single
    /// output pad must feed a downstream consumer.
    MuxerWithoutOutput(String),
    /// A named input-pad reference (`mux.foo_0`) names a request pad this muxer
    /// does not define (M481): the element's `input_pad_index` scheme declined it.
    UnknownInputPad(String),
    /// Two input-pad references resolve to the same muxer input index (M481), e.g.
    /// `mux.video_0` named twice, or a named pad colliding with a positional one.
    DuplicateInputPad(String),
    /// A `queue` / `queue2` sits anywhere but a 1-in/1-out position. It is not an
    /// element in g2g (it collapses into the edge's backpressure policy), so it
    /// cannot be a source, a sink, or a fan-out / fan-in node.
    QueueRole(String),
    /// A `decodebin` has no upstream element to take its input caps from (it was
    /// the first element, or followed a bare `name.` reference). decodebin
    /// auto-plugs from its predecessor's declared caps, so it needs one.
    DecodebinNoUpstream,
    /// `decodebin` found no chain of registered decoders / parsers from its input
    /// caps to raw video or audio (the input caps are quoted). Either no decoder
    /// feature is compiled in, or the input is a container that needs a demuxer
    /// (auto-plugging through fan-out demuxers is not yet supported).
    NoDecodeChain(String),
    /// An `encodebin` had no `profile=`, or one that is not
    /// `container-caps:stream-caps[:stream-caps...]` (the message says which).
    EncodingProfile(String),
    /// An `encodebin` profile named a container no registered muxer writes, or a
    /// stream codec no compiled-in encoder produces (the message quotes it).
    NoEncoder(String),
    /// An `encodebin` chain's input is not raw video or audio, or its profile has
    /// no stream of that kind, so nothing in the profile can encode it.
    EncodebinInput(String),
    /// A `uridecodebin` / `playbin` was not at the head of its chain. It provides
    /// the source, so it must start the pipeline.
    UriSourceNotAtHead(String),
    /// A `uridecodebin` / `playbin` had no `uri=` property.
    MissingUri(String),
    /// The `uri=` could not be turned into a source (bad URI, or no handler
    /// registered for its scheme). The message quotes the URI and reason.
    Uri(String),
    /// Linking two nodes into the graph failed.
    Graph(GraphError),
}

impl From<GraphError> for ParseError {
    fn from(e: GraphError) -> Self {
        ParseError::Graph(e)
    }
}

impl core::fmt::Display for ParseError {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        match self {
            ParseError::Empty => f.write_str("empty pipeline"),
            ParseError::EmptyStage => f.write_str("empty node between '!' separators"),
            ParseError::TooFewStages => f.write_str("pipeline needs at least a source and a sink"),
            ParseError::UnknownSource(n) => write!(f, "unknown source element: {n}"),
            ParseError::UnknownElement(n) => write!(f, "unknown element: {n}"),
            ParseError::MalformedProperty { element, token } => {
                write!(
                    f,
                    "{element}: malformed property '{token}' (expected key=value)"
                )
            }
            ParseError::UnknownProperty { element, key } => {
                write!(f, "{element}: no property named '{key}'")
            }
            ParseError::BadValue {
                element,
                key,
                value,
            } => {
                write!(f, "{element}: invalid value '{value}' for property '{key}'")?;
                // Inline caps become a `capsfilter` (key `caps`); a gst dev often
                // reaches for range / list / feature syntax g2g's launch parser
                // does not accept, and the bare "invalid value" hides why.
                if element == "capsfilter" && key == "caps" && value.contains(['[', '{', '(']) {
                    write!(
                        f,
                        " (a launch caps filter takes fixed fields, ranges [min,max], \
                         and lists {{a,b}} with numeric / known values; caps features \
                         like (memory:...) are not supported)"
                    )?;
                }
                Ok(())
            }
            ParseError::BadEnumValue {
                element,
                key,
                value,
                values,
            } => {
                write!(f, "{element}: invalid value '{value}' for property '{key}'")?;
                if !values.is_empty() {
                    write!(f, " (valid: {values})")?;
                }
                if value.contains('+') {
                    write!(f, " (a flag set joins nicks with '+', e.g. video+audio)")?;
                }
                Ok(())
            }
            ParseError::UnknownReference(n) => {
                write!(f, "reference to undeclared element name: {n}")
            }
            ParseError::DuplicateName(n) => write!(f, "duplicate element name: {n}"),
            ParseError::NotAMuxer(n) => {
                write!(
                    f,
                    "{n}: more than one input links here, but it is not a registered muxer"
                )
            }
            ParseError::MuxerWithoutOutput(n) => {
                write!(
                    f,
                    "{n}: muxer has no outgoing link; its output must feed a consumer"
                )
            }
            ParseError::UnknownInputPad(n) => {
                write!(
                    f,
                    "{n}: no such input request pad (this element defines no pad by that name)"
                )
            }
            ParseError::DuplicateInputPad(n) => {
                write!(f, "{n}: two inputs resolve to the same request pad index")
            }
            ParseError::QueueRole(n) => {
                write!(f, "{n}: a queue must sit between two elements (1-in/1-out); it maps to an edge policy, not a source/sink/branch")
            }
            ParseError::DecodebinNoUpstream => {
                write!(f, "decodebin has no upstream element to decode; it must follow a source or element with declared caps")
            }
            ParseError::NoDecodeChain(caps) => {
                write!(f, "decodebin: no decoder chain from {caps} to raw (no decoder feature compiled in, or a container that needs a demuxer)")
            }
            ParseError::EncodingProfile(msg) => write!(
                f,
                "encodebin profile: {msg} (expected profile=\"container-caps:stream-caps[:stream-caps]\", e.g. profile=\"video/x-matroska:video/x-vp8:audio/x-opus\")"
            ),
            ParseError::NoEncoder(msg) => write!(f, "encodebin: {msg}"),
            ParseError::EncodebinInput(msg) => write!(f, "encodebin: {msg}"),
            ParseError::UriSourceNotAtHead(n) => {
                write!(f, "{n}: provides the source, so it must start the pipeline (be the first element)")
            }
            ParseError::MissingUri(n) => write!(f, "{n}: missing required 'uri=' property"),
            ParseError::Uri(msg) => write!(f, "uri error: {msg}"),
            ParseError::Graph(e) => write!(f, "graph link error: {e:?}"),
        }
    }
}

/// One parsed element: factory name plus its `key=value` properties (all owned so
/// errors can name them), and the optional `name=` handle that pad references
/// resolve against. `name` and `log-category` are special-cased here (launch
/// keywords, not properties), never applied as properties.
struct ElementSpec {
    name: String,
    props: Vec<(String, String)>,
    instance: Option<String>,
    log_category: Option<String>,
}

/// An item in a chain: an element to build, a `t.` reference to a named element
/// declared elsewhere (the branching / link-by-name syntax), or a node already
/// constructed by a macro expansion (`uridecodebin` / `playbin`, M196), spliced
/// in directly rather than built by name.
enum Item {
    Element(ElementSpec),
    /// A `t.` / `d.video_0` reference to a named element. `pad` is the suffix after
    /// the dot (`""` for a bare `t.`, `"video_0"` for `d.video_0`), used by the
    /// explicit-demux fan-out (M476) to select which stream a branch reads; ignored
    /// for a tee (positional).
    Ref {
        name: String,
        pad: String,
    },
    Prebuilt(PrebuiltNode),
}

/// A node a macro expansion built ahead of the structural pass: a source
/// constructed from a `uri=` scheme handler, or a decoder the auto-plug search
/// instantiated. Spliced into the graph as-is (it has no name to build by).
enum PrebuiltNode {
    Source(Box<dyn DynSourceLoop>),
    Element(Box<dyn DynAsyncElement>),
}

/// A run of items linked left-to-right by `!`. Branches are separate chains
/// joined through named references.
type Chain = Vec<Item>;

/// A caps description node (`video/x-raw,format=nv12,...`): a media type whose
/// `/` precedes any `=` field. A property value's `/` (a path or a fraction)
/// comes after its `=`, so it is not mistaken for caps.
fn is_caps_token(tok: &str) -> bool {
    match (tok.find('/'), tok.find('=')) {
        (Some(slash), Some(eq)) => slash < eq,
        (Some(_), None) => true,
        _ => false,
    }
}

/// A pad reference (`t.`, `t.src_0`, `d.video_0`): a name, a `.`, and no `=` / `/`.
/// Returns the referenced element name and the pad suffix after the first dot
/// (`""` for a bare `t.`). The suffix drives the explicit-demux fan-out (M476);
/// a tee ignores it (positional).
fn split_pad_ref(tok: &str) -> Option<(&str, &str)> {
    if tok.contains('=') || tok.contains('/') || !tok.contains('.') {
        return None;
    }
    let (name, pad) = tok.split_once('.').unwrap_or((tok, ""));
    (!name.is_empty()).then_some((name, pad))
}

/// The referenced element name of a pad reference (the suffix dropped); the
/// token-boundary test in [`consume_element`].
fn as_ref_name(tok: &str) -> Option<&str> {
    split_pad_ref(tok).map(|(name, _)| name)
}

/// Parse a demux output-pad suffix into a [`PadRequest`] (M476): `"video_0"` ->
/// `{ Video, 0 }`, `"audio_1"` -> `{ Audio, 1 }`, `"text_0"` / `"subtitle_0"` ->
/// `{ Text, 0 }`, `"caption"` -> `{ Text, 0 }`, `"src_2"` -> `{ Any, 2 }`. A bare `d.` (empty suffix) or an
/// unrecognized prefix is `{ Any, ordinal }`, i.e. positional by reference order.
fn parse_pad_request(pad: &str, ordinal: usize) -> PadRequest {
    let (prefix, index) = match pad.rsplit_once('_') {
        Some((p, n)) => (p, n.parse::<usize>().ok()),
        None => (pad, None),
    };
    let kind = match prefix {
        "video" => PadKind::Video,
        "audio" => PadKind::Audio,
        // `caption` is the name a closed-caption fan-in gives its second pad; it
        // rides the text kind, which is the one every sidecar-cue pad uses.
        "text" | "subtitle" | "caption" => PadKind::Text,
        _ => PadKind::Any,
    };
    // `src_N` (output) / `sink_N` (input) and unrecognized prefixes select the Nth
    // stream / pad; a bare `d.` has no index, so it takes the positional ordinal.
    let index = match kind {
        PadKind::Any if prefix == "src" || prefix == "sink" || prefix.is_empty() => {
            index.unwrap_or(ordinal)
        }
        PadKind::Any => ordinal,
        _ => index.unwrap_or(0),
    };
    PadRequest { kind, index }
}

/// Consume an element's `key=value` properties from the token stream, stopping at
/// a `!`, a caps node, or a pad reference (the next node begins). A bare token
/// with no `=` is a malformed property (the gst typo case), reported by name.
fn consume_element<'a, I: Iterator<Item = &'a str>>(
    name: &str,
    tokens: &mut core::iter::Peekable<I>,
    knows_element: &dyn Fn(&str) -> bool,
) -> Result<ElementSpec, ParseError> {
    let mut spec = ElementSpec {
        name: name.to_string(),
        props: Vec::new(),
        instance: None,
        log_category: None,
    };
    while let Some(&tok) = tokens.peek() {
        if tok == "!" || is_caps_token(tok) || as_ref_name(tok).is_some() {
            break;
        }
        // A bare token naming a registered element opens a new top-level chain
        // rather than being this element's property, the way gst-launch parses
        // `videotestsrc ! xvimagesink audiotestsrc ! pulsesink`. One that names
        // nothing registered is the typo it looks like.
        if !tok.contains('=') && knows_element(tok) {
            break;
        }
        let (key, value) = tok
            .split_once('=')
            .ok_or_else(|| ParseError::MalformedProperty {
                element: name.to_string(),
                token: tok.to_string(),
            })?;
        tokens.next();
        let value = unquote_value(value);
        if key == "name" {
            spec.instance = Some(value);
        } else if key == "log-category" {
            // M847: a launch keyword like `name=`, not a property: it renames this
            // instance's `G2G_DEBUG` filter key rather than configuring it.
            spec.log_category = Some(value);
        } else {
            spec.props.push((key.to_string(), value));
        }
    }
    Ok(spec)
}

/// The characters a `\` may escape in a launch line: the quote characters, a `\`
/// itself, the `!` separator, the `#` comment marker, and whitespace. A `\`
/// before anything else is literal, so an unescaped Windows path
/// (`location=C:\videos\a.ts`) survives.
fn is_escapable(c: char) -> bool {
    matches!(c, '"' | '\'' | '\\' | '!' | '#') || c.is_whitespace()
}

/// Resolve a token's quoting into the literal property value: drop the quote
/// characters that open / close a quoted region and resolve `\x` escapes. A value
/// may open and close a quoted region more than once
/// (`location=/tmp/"my dir"/a.ts`), so this walks the whole token rather than
/// stripping one surrounding pair. Both quote characters work; gst-launch quotes
/// only with `"` (a `'` is literal there), so this accepts a superset.
fn unquote_value(v: &str) -> String {
    let mut out = String::with_capacity(v.len());
    let mut quote: Option<char> = None;
    let mut chars = v.chars();
    while let Some(c) = chars.next() {
        match c {
            '\\' => match chars.next() {
                Some(n) if is_escapable(n) => out.push(n),
                Some(n) => {
                    out.push('\\');
                    out.push(n);
                }
                None => out.push('\\'),
            },
            '"' | '\'' if quote.is_none() => quote = Some(c),
            _ if Some(c) == quote => quote = None,
            _ => out.push(c),
        }
    }
    out
}

/// Split a pipeline string into tokens, honoring quoted property values and
/// `#` comments. Outside quotes, whitespace separates tokens and `!` is a
/// standalone token; inside a `"..."` or `'...'` region both are literal, so a
/// value may contain spaces (and even `!`), e.g. `element="x264enc bitrate=4000"`
/// or `location='/my file.ts'`. A `\` escapes the next character when it is one
/// of the launch-special ones ([`is_escapable`]), so `location=/my\ file.ts` and
/// a `\"` inside a quoted region are literal. A `#` outside quotes starts a
/// comment that runs to end of line (a pasted multi-line pipeline may carry
/// them). Quotes and escapes are kept on the token; [`unquote_value`] resolves
/// them once the `key=` split is known. An unterminated quote runs to end of
/// input (best-effort; the property parse then reports any resulting malformed
/// token).
fn tokenize(s: &str) -> Vec<String> {
    let mut tokens = Vec::new();
    let mut cur = String::new();
    // The open quote char (`"` or `'`), or `None` outside a quoted region.
    let mut quote: Option<char> = None;
    let mut in_comment = false;
    let mut chars = s.chars().peekable();
    while let Some(c) = chars.next() {
        if in_comment {
            // A comment runs to end of line; the newline (whitespace) ends it.
            if c == '\n' {
                in_comment = false;
            }
            continue;
        }
        // An escape pair stays whole (and inert) through tokenization: neither
        // char can close a quote, split a token, or start a comment.
        if c == '\\' {
            if let Some(&next) = chars.peek() {
                if is_escapable(next) {
                    chars.next();
                    cur.push('\\');
                    cur.push(next);
                    continue;
                }
            }
            cur.push('\\');
            continue;
        }
        match c {
            '"' | '\'' if quote.is_none() => {
                quote = Some(c);
                cur.push(c);
            }
            _ if Some(c) == quote => {
                quote = None;
                cur.push(c);
            }
            // A `#` only starts a comment at a token boundary (nothing buffered);
            // mid-token it is literal, so a URI fragment (`uri=...#closed-captions=cc1`,
            // `#t=10`) is preserved.
            '#' if quote.is_none() && cur.is_empty() => {
                in_comment = true;
            }
            '!' if quote.is_none() => {
                if !cur.is_empty() {
                    tokens.push(core::mem::take(&mut cur));
                }
                tokens.push("!".to_string());
            }
            c if c.is_whitespace() && quote.is_none() => {
                if !cur.is_empty() {
                    tokens.push(core::mem::take(&mut cur));
                }
            }
            _ => cur.push(c),
        }
    }
    if !cur.is_empty() {
        tokens.push(cur);
    }
    tokens
}

/// Split a `gst-launch` pipeline string into chains: runs of nodes linked by `!`,
/// with branches expressed as separate chains joined through `name=` / `t.`.
#[cfg(test)]
fn parse_chains(pipeline: &str) -> Result<Vec<Chain>, ParseError> {
    // The unit tests exercise the tokenizer alone, with nothing registered: every
    // bare token is then a property, which is the pre-registry behaviour.
    parse_chains_with(pipeline, &|_| false)
}

/// Split a `gst-launch` pipeline string into chains, asking `knows_element`
/// whether a bare token opens a new chain (see [`consume_element`]).
fn parse_chains_with(
    pipeline: &str,
    knows_element: &dyn Fn(&str) -> bool,
) -> Result<Vec<Chain>, ParseError> {
    let trimmed = pipeline.trim();
    if trimmed.is_empty() {
        return Err(ParseError::Empty);
    }
    // Tokenize quote-aware so a `!` (a standalone token) and whitespace inside a
    // quoted value are literal, letting a property value carry spaces.
    let toks = tokenize(trimmed);
    let mut tokens = toks.iter().map(String::as_str).peekable();

    #[derive(Clone, Copy)]
    enum St {
        Start,
        AfterBang,
        AfterNode,
    }

    let mut chains: Vec<Chain> = Vec::new();
    let mut cur: Chain = Vec::new();
    let mut st = St::Start;

    loop {
        match st {
            St::Start | St::AfterBang => {
                let after_bang = matches!(st, St::AfterBang);
                let Some(tok) = tokens.next() else {
                    if after_bang {
                        return Err(ParseError::EmptyStage); // trailing `!`
                    }
                    break;
                };
                if tok == "!" {
                    return Err(ParseError::EmptyStage); // leading or doubled `!`
                }
                if is_caps_token(tok) {
                    cur.push(Item::Element(ElementSpec {
                        name: "capsfilter".to_string(),
                        props: alloc::vec![("caps".to_string(), tok.to_string())],
                        instance: None,
                        log_category: None,
                    }));
                    st = St::AfterNode;
                } else if let Some((name, pad)) = split_pad_ref(tok) {
                    cur.push(Item::Ref {
                        name: name.to_string(),
                        pad: pad.to_string(),
                    });
                    if after_bang {
                        // Tail ref (`! t.`): links the upstream node into the
                        // named element and ends the chain.
                        chains.push(core::mem::take(&mut cur));
                        st = St::Start;
                    } else {
                        // Head ref (`t. ! ...`): feeds the chain from it.
                        st = St::AfterNode;
                    }
                } else {
                    cur.push(Item::Element(consume_element(
                        tok,
                        &mut tokens,
                        knows_element,
                    )?));
                    st = St::AfterNode;
                }
            }
            St::AfterNode => match tokens.peek() {
                Some(&"!") => {
                    tokens.next();
                    st = St::AfterBang;
                }
                Some(_) => {
                    // A node not joined by `!`: the current chain ends here and a
                    // new one starts at this token (reprocessed as a head).
                    chains.push(core::mem::take(&mut cur));
                    st = St::Start;
                }
                None => break,
            },
        }
    }

    if !cur.is_empty() {
        chains.push(cur);
    }
    Ok(chains)
}

/// The property surface a launch node exposes, so one applier serves every node
/// shape (source, transform / sink, muxer, demuxer, fan-out source) instead of a
/// copy of the parse-and-set loop per shape.
trait PropTarget {
    fn specs(&self) -> &'static [PropertySpec];
    fn set(&mut self, name: &str, value: PropValue) -> Result<(), PropError>;
}

macro_rules! impl_prop_target {
    ($($t:ty),* $(,)?) => {
        $(impl PropTarget for Box<$t> {
            fn specs(&self) -> &'static [PropertySpec] {
                (**self).properties()
            }
            fn set(&mut self, name: &str, value: PropValue) -> Result<(), PropError> {
                (**self).set_property(name, value)
            }
        })*
    };
}

impl_prop_target!(
    dyn DynSourceLoop,
    dyn DynAsyncElement,
    dyn crate::runtime::DynMultiInputElement,
    dyn crate::runtime::DynMultiOutputElement,
    dyn crate::fanout::DynMultiOutputSource,
);

/// Apply parsed `key=value` props to a node: look each name up in the element's
/// declared [`PropertySpec`]s, parse the text for that spec (which validates an
/// enum nick / flag set against its declared values), then set it.
fn apply_props<T: PropTarget>(
    target: &mut T,
    name: &str,
    props: &[(String, String)],
) -> Result<(), ParseError> {
    for (key, raw) in props {
        let bad_value = || ParseError::BadValue {
            element: name.into(),
            key: key.clone(),
            value: raw.clone(),
        };
        let spec = target.specs().iter().find(|s| s.name == key).copied();
        let value = match spec {
            Some(spec) => spec.parse_value(raw).map_err(|e| match e {
                ValueError::Kind(_) => bad_value(),
                ValueError::Nick(nick) => ParseError::BadEnumValue {
                    element: name.into(),
                    key: key.clone(),
                    value: nick,
                    values: spec.enum_values.unwrap_or(""),
                },
            })?,
            // The element takes names it cannot declare, so there is no kind to
            // parse for: the text goes on as written for whatever does know it.
            None if crate::property::takes_undeclared_properties(target.specs()) => {
                PropValue::Str(raw.clone())
            }
            None => {
                return Err(ParseError::UnknownProperty {
                    element: name.into(),
                    key: key.clone(),
                })
            }
        };
        target.set(key, value).map_err(|_| bad_value())?;
    }
    Ok(())
}

/// Build the runnable [`Graph`] from parsed chains: flatten elements, resolve
/// `t.` references into directed links, derive each element's role (and any tee's
/// fan-out width) from its link degree, then construct and wire the nodes.
/// `decodebin`: not an element, but a macro that expands, at parse time, into the
/// chain of decoders / parsers the auto-plug search finds from its upstream caps
/// down to raw video or audio (M193).
fn is_decodebin(name: &str) -> bool {
    matches!(name, "decodebin")
}

/// Depth bound for the decodebin auto-plug search: a parse + decode (+ a spare
/// hop) is 2-3, so this leaves headroom without letting an unsatisfiable target
/// wander.
const DECODEBIN_MAX_DEPTH: usize = 6;

/// The decode chain to raw for `input`, decoding into `preferred` (the memory
/// the element right after the macro takes).
///
/// The chain is returned built, not named: the caps the search chose a decoder
/// to produce reach the element only through the factory that takes them, and a
/// parameterless launch factory would build the decoder's default format again.
/// That is how `decodebin ! wgpusink` reached a strict-NV12 sink with a decoder
/// left on I420; the solver re-fixates among the decoder's advertised formats
/// once the chosen one arrives.
fn autoplug_for_consumer(
    registry: &Registry,
    input: &Caps,
    preferred: MemoryDomainKind,
    avoided: &[&str],
) -> Option<Vec<Box<dyn DynAsyncElement>>> {
    let raw = |c: &Caps| is_raw_video(c) || is_raw_audio(c);
    let mut chain =
        registry.autoplug_avoiding(input, &raw, DECODEBIN_MAX_DEPTH, preferred, avoided)?;
    // M421/M676: prepend the re-framing parser ahead of a real decode of an
    // elementary stream, like the boxed `decodebin` splice (the caps-identity
    // parser is invisible to the shortest-chain search, so it never appears in
    // the chain).
    if !chain.is_empty() {
        if let Some(parser) = registry
            .parser_name(input)
            .and_then(|p| registry.make_element(p))
        {
            chain.insert(0, parser);
        }
    }
    Some(chain)
}

/// Expand every `decodebin` node into the decoder chain the registry auto-plugs
/// from its predecessor's declared caps down to raw (video or audio). An empty
/// chain (the input is already raw) drops the node entirely, so its predecessor
/// links straight to its consumer. The predecessor is the element immediately
/// before the `decodebin` in the same chain; a `decodebin` with no upstream
/// element (chain head, or after a bare `name.` reference) is a loud error,
/// since it has nothing to take its input caps from.
fn expand_decodebin(
    registry: &Registry,
    chains: Vec<Chain>,
    avoided: &[&str],
) -> Result<Vec<Chain>, ParseError> {
    // Names referenced as `name.` somewhere: a `decodebin name=d` with such refs is
    // a FAN-OUT node (M482), not the inline linear case, so it is left unexpanded
    // here and handled by the decodebin-select path in `build_graph` (which probes
    // the file, demuxes, and decodes each requested port). Only the unreferenced
    // inline `... ! decodebin ! ...` expands to a linear decode chain below.
    let mut referenced: alloc::collections::BTreeSet<String> = alloc::collections::BTreeSet::new();
    for chain in &chains {
        for item in chain {
            if let Item::Ref { name, .. } = item {
                referenced.insert(name.clone());
            }
        }
    }
    let is_fanout_decodebin = |spec: &ElementSpec| {
        is_decodebin(&spec.name)
            && spec
                .instance
                .as_deref()
                .map(|n| referenced.contains(n))
                .unwrap_or(false)
    };

    let mut out = Vec::with_capacity(chains.len());
    for chain in chains {
        let mut new_chain: Chain = Vec::with_capacity(chain.len());
        // The element (name + props) whose output caps feed the next decodebin:
        // the most recent real element. A `Ref` clears it (its caps live in
        // another chain). Props matter because they can re-type the output (a
        // `filesrc`'s `bytestream-format` selects the container).
        let mut upstream: Option<(String, Vec<(String, String)>)> = None;
        let mut items = chain.into_iter().peekable();
        while let Some(item) = items.next() {
            match item {
                // A fan-out `decodebin name=d` is left for `build_graph`'s
                // decodebin-select path; it is not linearly expandable.
                Item::Element(spec) if is_fanout_decodebin(&spec) => {
                    upstream = Some((spec.name.clone(), spec.props.clone()));
                    new_chain.push(Item::Element(spec));
                }
                Item::Element(spec) if is_decodebin(&spec.name) => {
                    let (pred, props) = upstream.as_ref().ok_or(ParseError::DecodebinNoUpstream)?;
                    let caps = resolve_upstream_caps(registry, pred, props)?;
                    // M746: a single-stream demux fixes its output pad before parsing
                    // any byte, so it defaults to a video port; on an audio-only
                    // container the default auto-plug would pick a video decoder and
                    // fail "no caps overlap". If a primary-stream hook sniffs the file
                    // and names the real (audio) stream, plug that demux with its
                    // stream selection and auto-plug the decoder from the elementary
                    // caps instead. A hook declines a container with a video track (the
                    // default video path is right) or one it does not parse.
                    let location = props
                        .iter()
                        .find(|(k, _)| k == "location")
                        .map(|(_, v)| v.as_str());
                    let chain_input =
                        match location.and_then(|loc| registry.primary_stream(loc, &caps)) {
                            Some(primary) => {
                                new_chain.push(Item::Element(ElementSpec {
                                    name: primary.demux.to_string(),
                                    props: primary.props.clone(),
                                    instance: None,
                                    log_category: None,
                                }));
                                primary.caps
                            }
                            None => caps,
                        };
                    // M1018: the element right after the `decodebin` decides what
                    // memory the decoder should decode into, so a GPU-resident
                    // consumer gets a decoder that decodes into its domain rather
                    // than one whose frames have to be downloaded. Only the
                    // immediate consumer counts, the rule the graph-side
                    // derivation follows.
                    let preferred = match items.peek() {
                        Some(Item::Element(consumer)) => {
                            registry.declared_memory_preference(&consumer.name)
                        }
                        _ => MemoryDomainKind::System,
                    };
                    let decoders =
                        autoplug_for_consumer(registry, &chain_input, preferred, avoided)
                            .ok_or_else(|| {
                                ParseError::NoDecodeChain(alloc::format!("{chain_input:?}"))
                            })?;
                    // Built here rather than named, so each element is
                    // constructed for the caps the search chose it to produce (a
                    // multi-format decoder emits the format the consumer takes,
                    // not the first one it lists). A pre-built node carries no
                    // name, so `upstream` clears as it does after a
                    // `uridecodebin`.
                    for element in decoders {
                        new_chain.push(Item::Prebuilt(PrebuiltNode::Element(element)));
                    }
                    upstream = None;
                }
                Item::Element(spec) => {
                    upstream = Some((spec.name.clone(), spec.props.clone()));
                    new_chain.push(Item::Element(spec));
                }
                Item::Ref { name, pad } => {
                    upstream = None;
                    new_chain.push(Item::Ref { name, pad });
                }
                // A pre-built node (from `uridecodebin` / `playbin`) carries no
                // name to take declared caps from, so a `decodebin` cannot follow
                // it. Clear the upstream; if a decodebin does follow, it reports
                // the missing-upstream error.
                prebuilt @ Item::Prebuilt(_) => {
                    upstream = None;
                    new_chain.push(prebuilt);
                }
            }
        }
        out.push(new_chain);
    }
    Ok(out)
}

/// `encodebin`: not an element either, but a macro that expands, at parse time,
/// into one encoder per profile stream plus the muxer that writes the profile's
/// container (M1089). `encodebin2` is gst's name for the same thing.
fn is_encodebin(name: &str) -> bool {
    matches!(name, "encodebin" | "encodebin2")
}

/// The separator between an encoding profile's container and its streams, the
/// same string form gst's `GstEncodingProfile` serializes to.
const PROFILE_SEPARATOR: char = ':';

/// A parsed `profile=`: the container to mux into (`None` for a container-less
/// profile, which is one coded stream and no muxer), and each stream it carries.
struct EncodingProfile {
    container: Option<Caps>,
    streams: Vec<ProfileStream>,
}

/// One stream of a profile: the coded form it names, plus whatever its part
/// pins (M1097). A pinned raw field is honoured by the converter spliced ahead
/// of the encoder; a pinned bitrate is a property on the encoder itself.
struct ProfileStream {
    caps: Caps,
    width: Option<u32>,
    height: Option<u32>,
    /// Frames per second as the `numerator/denominator` `videorate` takes.
    framerate: Option<(u32, u32)>,
    sample_rate: Option<u32>,
    channels: Option<u8>,
    /// Bits per second, gst's `bitrate` unit.
    bitrate: Option<u64>,
}

/// The encoder property a profile's pinned bitrate is applied through.
const BITRATE_PROPERTY: &str = "bitrate";

/// Parse `container-caps:stream-caps[:stream-caps...]`.
///
/// The media type of each part chooses the element: the container names the
/// muxer, a stream names the codec its encoder must produce. A stream part may
/// also pin a width / height / framerate (or a sample `rate` / `channels`),
/// which splices the converter that reaches it, and a `bitrate`, which is set on
/// the encoder. Any other field is refused rather than dropped.
fn parse_encoding_profile(text: &str) -> Result<EncodingProfile, ParseError> {
    let bad = |msg: &str| ParseError::EncodingProfile(msg.to_string());
    let mut parts = text.split(PROFILE_SEPARATOR).map(str::trim).peekable();
    let first = *parts
        .peek()
        .filter(|p| !p.is_empty())
        .ok_or_else(|| bad("empty"))?;
    let head = single_caps(first).ok_or_else(|| {
        ParseError::EncodingProfile(alloc::format!("unknown media type `{first}`"))
    })?;
    // A leading container is the muxing form; a leading stream is gst's
    // container-less profile, which encodes and stops.
    let container = match head {
        Caps::ByteStream { .. } => {
            parts.next();
            Some(head)
        }
        _ => None,
    };
    let mut streams = Vec::new();
    for part in parts {
        if part.is_empty() {
            return Err(bad("empty stream"));
        }
        streams.push(parse_profile_stream(part)?);
    }
    if streams.is_empty() {
        return Err(bad("no stream"));
    }
    Ok(EncodingProfile { container, streams })
}

/// Parse one stream part: its media type plus the fields it pins.
///
/// The caps parser ignores a field it has no home for, so the pinned settings
/// are read from the text here instead: an audio caps carries a channel count
/// and a sample rate whether or not the part named them, and `bitrate` is not a
/// caps field at all.
fn parse_profile_stream(part: &str) -> Result<ProfileStream, ParseError> {
    let caps = single_caps(part)
        .ok_or_else(|| ParseError::EncodingProfile(alloc::format!("unknown stream `{part}`")))?;
    let video = matches!(caps, Caps::CompressedVideo { .. } | Caps::RawVideo { .. });
    let audio = matches!(caps, Caps::Audio { .. });
    let mut stream = ProfileStream {
        caps,
        width: None,
        height: None,
        framerate: None,
        sample_rate: None,
        channels: None,
        bitrate: None,
    };
    for field in crate::caps_parse::split_top_commas(part)
        .into_iter()
        .skip(1)
    {
        let (key, value) = field.split_once('=').ok_or_else(|| {
            ParseError::EncodingProfile(alloc::format!(
                "`{part}` has a field `{}` that is not key=value",
                field.trim()
            ))
        })?;
        let (key, value) = (key.trim(), value.trim());
        match key {
            "width" if video => {
                stream.width = Some(profile_number(part, key, value, MAX_PROFILE_VALUE)? as u32)
            }
            "height" if video => {
                stream.height = Some(profile_number(part, key, value, MAX_PROFILE_VALUE)? as u32)
            }
            "framerate" if video => stream.framerate = Some(profile_framerate(part, value)?),
            "rate" if audio => {
                stream.sample_rate = Some(profile_number(part, key, value, MAX_PROFILE_VALUE)? as u32)
            }
            "channels" if audio => {
                stream.channels = Some(profile_number(part, key, value, MAX_CHANNELS)? as u8)
            }
            BITRATE_PROPERTY => {
                stream.bitrate = Some(profile_number(part, key, value, MAX_PROFILE_VALUE)?)
            }
            _ => {
                return Err(ParseError::EncodingProfile(alloc::format!(
                    "`{part}` pins `{key}`, which a profile cannot apply: set it on an element ahead of the encodebin instead"
                )))
            }
        }
    }
    // The scaler takes a whole output geometry, so half of one has nothing to
    // scale to.
    if stream.width.is_some() != stream.height.is_some() {
        return Err(ParseError::EncodingProfile(alloc::format!(
            "`{part}` pins one of width / height: a scale needs both, so pin both or neither"
        )));
    }
    Ok(stream)
}

/// Upper bounds on a pinned profile field, so a bogus value fails the parse
/// rather than reaching an element as a truncated cast. Every field but the
/// channel count is a `u32` where it lands.
const MAX_PROFILE_VALUE: u64 = u32::MAX as u64;
const MAX_CHANNELS: u64 = u8::MAX as u64;

/// A pinned profile field's number. Zero is rejected everywhere it appears
/// (a zero geometry, rate, channel count or bitrate pins nothing).
fn profile_number(part: &str, key: &str, value: &str, max: u64) -> Result<u64, ParseError> {
    value
        .parse::<u64>()
        .ok()
        .filter(|n| (1..=max).contains(n))
        .ok_or_else(|| {
            ParseError::EncodingProfile(alloc::format!(
                "`{part}` has a bad `{key}={value}`: expected a whole number 1..={max}"
            ))
        })
}

/// A pinned `framerate=N/D` (or a bare `framerate=N`, gst's shorthand), in the
/// fraction form `videorate`'s property takes.
fn profile_framerate(part: &str, value: &str) -> Result<(u32, u32), ParseError> {
    let (numerator, denominator) = value.split_once('/').unwrap_or((value, "1"));
    Ok((
        profile_number(part, "framerate", numerator.trim(), MAX_PROFILE_VALUE)? as u32,
        profile_number(part, "framerate", denominator.trim(), MAX_PROFILE_VALUE)? as u32,
    ))
}

/// One caps from a media-type description. A format-less raw description
/// (`video/x-raw`, `audio/x-raw`) expands to every format it could be, and an
/// uncompressed profile stream means "store what arrives" rather than any one of
/// them, so its first alternative stands for the whole set: the expansion reads
/// only which kind of stream it is. `None` when the text names no media type
/// this build knows.
fn single_caps(desc: &str) -> Option<Caps> {
    let set = CapsSet::from_gst_string(desc)?;
    match set.alternatives() {
        [only] => Some(only.clone()),
        [first, ..] if is_raw_video(first) || is_raw_audio(first) => Some(first.clone()),
        _ => None,
    }
}

/// The profile stream that encodes `input`: a raw video input takes the first
/// coded video stream, raw audio the first coded audio one. `None` when the
/// profile carries nothing of that kind.
fn stream_for_input<'a>(profile: &'a EncodingProfile, input: &Caps) -> Option<&'a ProfileStream> {
    let want_video = is_raw_video(input);
    let want_audio = is_raw_audio(input);
    profile.streams.iter().find(|stream| match stream.caps {
        Caps::CompressedVideo { .. } => want_video,
        Caps::Audio { .. } => want_audio,
        _ => false,
    })
}

/// The converters an encoder needs ahead of it, when the raw stream reaching it
/// is not in a form it takes: a pixel-format convert for video, a sample-format
/// convert plus a resampler for audio (either the format or the rate can be the
/// mismatch). Asked of the encoder itself rather than guessed from a template,
/// so an encoder that already accepts what arrives gets nothing spliced.
///
/// With no upstream caps to check (a decode chain the parser expanded has no
/// name to ask), the converters go in anyway: they pass a matching format
/// through, where a missing one fails the negotiation.
fn conversion_specs(
    registry: &Registry,
    encoder: &ElementSpec,
    input: Option<&Caps>,
    stream: &Caps,
) -> Vec<ElementSpec> {
    let plain = |name: &str| ElementSpec {
        name: name.to_string(),
        props: Vec::new(),
        instance: None,
        log_category: None,
    };
    if let Some(input) = input {
        let accepted = registry
            .make_element(&encoder.name)
            .is_some_and(|element| element.intercept_caps(input).is_ok());
        if accepted {
            return Vec::new();
        }
    }
    // Which converters belong to this stream is the stream's own kind: the
    // encoder for a video stream takes raw video, whatever reaches it.
    match stream {
        Caps::CompressedVideo { .. } => Vec::from([plain("videoconvert")]),
        Caps::Audio { .. } => Vec::from([plain("audioconvert"), plain("audioresample")]),
        _ => Vec::new(),
    }
}

/// The encoder element for one profile stream, as a spec the builder constructs
/// like any named element. The properties come from the registry's choice (a
/// multi-codec encoder is pinned to this codec with `codec=`), plus the
/// profile's own `bitrate` where it pinned one.
fn encoder_spec(
    registry: &Registry,
    stream: &ProfileStream,
) -> Result<Option<ElementSpec>, ParseError> {
    // An uncompressed stream profile (`audio/x-raw` in a WAV, `video/x-raw` in a
    // y4m) stores what arrives, so there is nothing to encode.
    if is_raw_video(&stream.caps) || is_raw_audio(&stream.caps) {
        if let Some(bitrate) = stream.bitrate {
            return Err(ParseError::EncodingProfile(alloc::format!(
                "pins `bitrate={bitrate}` on an uncompressed stream, which has no encoder to set it on"
            )));
        }
        return Ok(None);
    }
    let choice = registry.encoder_choice(&stream.caps).ok_or_else(|| {
        ParseError::NoEncoder(alloc::format!(
            "no encoder for {} is compiled in (build the feature for it, or name an encoder explicitly)",
            stream.caps.to_gst_string()
        ))
    })?;
    let mut props: Vec<(String, String)> = choice
        .props
        .iter()
        .map(|(k, v)| (k.to_string(), v.to_string()))
        .collect();
    if let Some(bitrate) = stream.bitrate {
        if !declares_bitrate(registry, choice.element) {
            return Err(ParseError::EncodingProfile(alloc::format!(
                "pins `bitrate={bitrate}`, which `{}` has no property for: drop it, or name the encoder yourself with the knob it does have",
                choice.element
            )));
        }
        props.push((BITRATE_PROPERTY.to_string(), alloc::format!("{bitrate}")));
    }
    Ok(Some(ElementSpec {
        name: choice.element.to_string(),
        props,
        instance: None,
        log_category: None,
    }))
}

/// Whether an encoder takes a target bitrate at all, asked of the element rather
/// than assumed, so a profile that pins one on a fixed-rate encoder fails the
/// parse instead of encoding at some other rate.
fn declares_bitrate(registry: &Registry, element: &str) -> bool {
    registry.make_element(element).is_some_and(|element| {
        element
            .properties()
            .iter()
            .any(|spec| spec.name == BITRATE_PROPERTY)
    })
}

/// The elements one branch of a profile expands to: the converters its encoder
/// needs, the converters its pinned settings need, then the encoder itself. A
/// pinned setting on an uncompressed stream still gets its converter, there is
/// just nothing after it but the muxer.
fn stream_specs(
    registry: &Registry,
    stream: &ProfileStream,
    upstream: Option<&(String, Vec<(String, String)>)>,
) -> Result<Vec<ElementSpec>, ParseError> {
    let encoder = encoder_spec(registry, stream)?;
    let mut specs = match &encoder {
        Some(encoder) => {
            let input = upstream
                .and_then(|(name, props)| resolve_upstream_caps(registry, name, props).ok());
            conversion_specs(registry, encoder, input.as_ref(), &stream.caps)
        }
        None => Vec::new(),
    };
    splice_pinned_settings(&mut specs, stream);
    specs.extend(encoder);
    Ok(specs)
}

/// Apply a stream's pinned settings to the converters ahead of its encoder: a
/// geometry sets `videoscale`, a framerate `videorate`, a channel count
/// `audioconvert` and a sample rate `audioresample`, adding that converter when
/// the encoder did not already need one.
fn splice_pinned_settings(specs: &mut Vec<ElementSpec>, stream: &ProfileStream) {
    if let (Some(width), Some(height)) = (stream.width, stream.height) {
        pin_on_converter(specs, "videoscale", "width", alloc::format!("{width}"));
        pin_on_converter(specs, "videoscale", "height", alloc::format!("{height}"));
    }
    if let Some((numerator, denominator)) = stream.framerate {
        pin_on_converter(
            specs,
            "videorate",
            "framerate",
            alloc::format!("{numerator}/{denominator}"),
        );
    }
    if let Some(channels) = stream.channels {
        pin_on_converter(
            specs,
            "audioconvert",
            "channels",
            alloc::format!("{channels}"),
        );
    }
    if let Some(sample_rate) = stream.sample_rate {
        pin_on_converter(
            specs,
            "audioresample",
            "samplerate",
            alloc::format!("{sample_rate}"),
        );
    }
}

/// Set a property on the named converter, appending the converter when the list
/// does not already carry it.
fn pin_on_converter(specs: &mut Vec<ElementSpec>, element: &str, key: &str, value: String) {
    match specs.iter_mut().find(|spec| spec.name == element) {
        Some(spec) => spec.props.push((key.to_string(), value)),
        None => specs.push(ElementSpec {
            name: element.to_string(),
            props: Vec::from([(key.to_string(), value)]),
            instance: None,
            log_category: None,
        }),
    }
}

/// `transcodebin`: decode whatever arrives and re-encode it to a profile, which
/// is exactly `decodebin ! encodebin profile=...`. Rewritten into those two
/// macros before either expands, so it inherits both behaviours (and both error
/// messages) rather than repeating them.
fn expand_transcodebin(chains: Vec<Chain>) -> Vec<Chain> {
    let mut out = Vec::with_capacity(chains.len());
    for chain in chains {
        let mut new_chain: Chain = Vec::with_capacity(chain.len() + 1);
        for item in chain {
            match item {
                Item::Element(spec) if spec.name == "transcodebin" => {
                    new_chain.push(Item::Element(ElementSpec {
                        name: "decodebin".to_string(),
                        props: Vec::new(),
                        instance: None,
                        log_category: spec.log_category.clone(),
                    }));
                    new_chain.push(Item::Element(ElementSpec {
                        name: "encodebin".to_string(),
                        ..spec
                    }));
                }
                other => new_chain.push(other),
            }
        }
        out.push(new_chain);
    }
    out
}

/// Expand every `encodebin` into `<encoder> ! <muxer>`: the encoder for the
/// stream matching that chain's input, and the muxer for the profile's container.
/// The muxer inherits the bin's `name=`, so a second branch reaching it as `e.`
/// links into the muxer the way any fan-in does; each such branch gets the
/// encoder for its own kind of input inserted ahead of the reference.
fn expand_encodebin(registry: &Registry, chains: Vec<Chain>) -> Result<Vec<Chain>, ParseError> {
    // Nothing to do for the common line with no encodebin in it.
    if !chains
        .iter()
        .flatten()
        .any(|item| matches!(item, Item::Element(spec) if is_encodebin(&spec.name)))
    {
        return Ok(chains);
    }

    // Every encodebin's profile and instance name, so a branch referencing the
    // name knows which profile to encode for.
    let mut profiles: Vec<(Option<String>, EncodingProfile)> = Vec::new();
    for chain in &chains {
        for item in chain {
            if let Item::Element(spec) = item {
                if is_encodebin(&spec.name) {
                    let text = prop(spec, "profile")
                        .ok_or_else(|| ParseError::EncodingProfile("missing".to_string()))?;
                    for (key, _) in &spec.props {
                        if !matches!(key.as_str(), "profile") {
                            return Err(ParseError::EncodingProfile(alloc::format!(
                                "unknown property `{key}`"
                            )));
                        }
                    }
                    profiles.push((spec.instance.clone(), parse_encoding_profile(text)?));
                }
            }
        }
    }

    let mut out = Vec::with_capacity(chains.len());
    for chain in chains {
        let mut new_chain: Chain = Vec::with_capacity(chain.len() + 1);
        let mut upstream: Option<(String, Vec<(String, String)>)> = None;
        for item in chain {
            match item {
                Item::Element(spec) if is_encodebin(&spec.name) => {
                    let profile = profiles
                        .iter()
                        .find(|(instance, _)| *instance == spec.instance)
                        .map(|(_, profile)| profile)
                        .expect("every encodebin's profile was parsed above");
                    // The container is the profile's head, so a missing muxer is
                    // reported before a missing encoder.
                    let muxer = muxer_spec(registry, profile, &spec)?;
                    let stream = profile_stream(registry, profile, upstream.as_ref())?;
                    for element in stream_specs(registry, stream, upstream.as_ref())? {
                        new_chain.push(Item::Element(element));
                    }
                    if let Some(muxer) = muxer {
                        new_chain.push(Item::Element(muxer));
                    }
                    upstream = None;
                }
                // A branch feeding an encodebin by name encodes its own input
                // first: the reference links into the muxer, which takes coded
                // streams.
                Item::Ref { name, pad } => {
                    if let Some((_, profile)) = profiles
                        .iter()
                        .find(|(instance, _)| instance.as_deref() == Some(name.as_str()))
                    {
                        let stream = profile_stream(registry, profile, upstream.as_ref())?;
                        for element in stream_specs(registry, stream, upstream.as_ref())? {
                            new_chain.push(Item::Element(element));
                        }
                    }
                    upstream = None;
                    new_chain.push(Item::Ref { name, pad });
                }
                Item::Element(spec) => {
                    upstream = Some((spec.name.clone(), spec.props.clone()));
                    new_chain.push(Item::Element(spec));
                }
                prebuilt @ Item::Prebuilt(_) => {
                    upstream = None;
                    new_chain.push(prebuilt);
                }
            }
        }
        out.push(new_chain);
    }
    Ok(out)
}

/// The profile stream a branch encodes into. A single-stream profile leaves no
/// choice, so it needs no input caps at all (which is what lets `decodebin !
/// encodebin` work: an expanded decode chain has no name to ask for caps). With
/// several streams the branch's own input decides which one, so the element
/// ahead of it has to declare its output.
fn profile_stream<'a>(
    registry: &Registry,
    profile: &'a EncodingProfile,
    upstream: Option<&(String, Vec<(String, String)>)>,
) -> Result<&'a ProfileStream, ParseError> {
    if let [only] = &profile.streams[..] {
        return Ok(only);
    }
    let input = encodebin_input_caps(registry, upstream)?;
    stream_for_input(profile, &input).ok_or_else(|| {
        ParseError::EncodebinInput(alloc::format!(
            "the profile carries no stream that encodes {}",
            input.to_gst_string()
        ))
    })
}

/// The raw caps reaching an `encodebin` (or a branch that references one): the
/// declared output of the element ahead of it.
fn encodebin_input_caps(
    registry: &Registry,
    upstream: Option<&(String, Vec<(String, String)>)>,
) -> Result<Caps, ParseError> {
    let (name, props) = upstream.ok_or_else(|| {
        ParseError::EncodebinInput(
            "no upstream element to encode; it must follow a source or element with declared caps"
                .to_string(),
        )
    })?;
    resolve_upstream_caps(registry, name, props).map_err(|_| {
        ParseError::EncodebinInput(alloc::format!("`{name}` declares no output caps to encode"))
    })
}

/// The muxer element for a profile's container, carrying the bin's instance name
/// so a `name.` reference resolves to it.
fn muxer_spec(
    registry: &Registry,
    profile: &EncodingProfile,
    bin: &ElementSpec,
) -> Result<Option<ElementSpec>, ParseError> {
    let Some(container) = &profile.container else {
        return Ok(None);
    };
    let name = registry.muxer_name(container).ok_or_else(|| {
        ParseError::NoEncoder(alloc::format!(
            "no muxer for {} is compiled in",
            container.to_gst_string()
        ))
    })?;
    Ok(Some(ElementSpec {
        name: name.to_string(),
        props: Vec::new(),
        instance: bin.instance.clone(),
        log_category: bin.log_category.clone(),
    }))
}

/// The caps a `decodebin` predecessor produces, used as the auto-plug input
/// (M195). For a registered source, build it and apply its properties so a
/// property that re-types the output (a `filesrc`'s `bytestream-format`) is
/// reflected via [`SourceLoop::probe_output_caps`]; fall back to the registry's
/// declared caps (a fixed source, or a transform's source-pad template).
/// `probe_output_caps` also sniffs a `bytestream-format=auto` source's header at
/// parse time (M480), so `decodebin` picks the demuxer from the real content even
/// when the file extension is wrong; only an unreadable / unrecognized file falls
/// back to the declared default.
fn resolve_upstream_caps(
    registry: &Registry,
    name: &str,
    props: &[(String, String)],
) -> Result<Caps, ParseError> {
    if let Some(mut src) = registry.make_source(name) {
        apply_props(&mut src, name, props)?;
        // `probe_output_caps` may sniff the header (a `bytestream-format=auto`
        // source), so `decodebin` picks the demuxer from the real content, not a
        // mislabeled extension; it falls back to the no-I/O caps otherwise.
        if let Some(caps) = src.probe_output_caps() {
            return Ok(caps);
        }
    }
    registry
        .declared_output_caps(name)
        .ok_or(ParseError::DecodebinNoUpstream)
}

/// `uridecodebin` / `playbin`: a source-providing macro. `uridecodebin uri=X`
/// builds the source from the URI scheme handler and auto-plugs the decode chain
/// to raw; `playbin uri=X` is that plus an auto sink (`autovideosink`, or the
/// `video-sink=` override), i.e. a complete pipeline.
fn is_uri_source(name: &str) -> bool {
    matches!(name, "uridecodebin" | "playbin")
}

/// The value of a spec property by key, if present.
fn prop<'a>(spec: &'a ElementSpec, key: &str) -> Option<&'a str> {
    spec.props
        .iter()
        .find(|(k, _)| k == key)
        .map(|(_, v)| v.as_str())
}

/// Expand every `uridecodebin` / `playbin` (a source position element) into the
/// `uri=` scheme handler's source plus the auto-plugged decode chain, as
/// pre-built nodes spliced straight into the chain. `playbin` additionally
/// appends an auto sink so the line is a complete pipeline. The element must head
/// its chain (it provides the source).
fn expand_uri_sources(
    registry: &Registry,
    chains: Vec<Chain>,
    avoided: &[&str],
) -> Result<Vec<Chain>, ParseError> {
    let mut out = Vec::with_capacity(chains.len());
    for chain in chains {
        let mut new_chain: Chain = Vec::with_capacity(chain.len());
        let mut items = chain.into_iter().enumerate().peekable();
        while let Some((i, item)) = items.next() {
            let spec = match item {
                Item::Element(spec) if is_uri_source(&spec.name) => spec,
                other => {
                    new_chain.push(other);
                    continue;
                }
            };
            if i != 0 {
                return Err(ParseError::UriSourceNotAtHead(spec.name));
            }
            let is_playbin = spec.name.starts_with("playbin");
            let uri =
                prop(&spec, "uri").ok_or_else(|| ParseError::MissingUri(spec.name.clone()))?;
            let (source, caps) = registry
                .build_uri_source(uri)
                .map_err(|e: UriError| ParseError::Uri(alloc::format!("{uri}: {e:?}")))?;
            // `playbin` names its own sink, a bare `uridecodebin` is followed by
            // one: either way the consumer's declared input memory picks the
            // decoder (M1018), as it does after a `decodebin`.
            let sink = is_playbin.then(|| {
                prop(&spec, "video-sink")
                    .unwrap_or("autovideosink")
                    .to_string()
            });
            let consumer = match (&sink, items.peek()) {
                (Some(sink), _) => Some(sink.as_str()),
                (None, Some((_, Item::Element(consumer)))) => Some(consumer.name.as_str()),
                _ => None,
            };
            let preferred = consumer.map_or(MemoryDomainKind::System, |name| {
                registry.declared_memory_preference(name)
            });
            let target = |c: &Caps| is_raw_video(c) || is_raw_audio(c);
            let decoders = registry
                .autoplug_avoiding(&caps, &target, DECODEBIN_MAX_DEPTH, preferred, avoided)
                .ok_or_else(|| ParseError::NoDecodeChain(alloc::format!("{caps:?}")))?;
            new_chain.push(Item::Prebuilt(PrebuiltNode::Source(source)));
            for dec in decoders {
                new_chain.push(Item::Prebuilt(PrebuiltNode::Element(dec)));
            }
            if let Some(sink) = sink {
                new_chain.push(Item::Element(ElementSpec {
                    name: sink,
                    props: Vec::new(),
                    instance: None,
                    log_category: None,
                }));
            }
        }
        out.push(new_chain);
    }
    Ok(out)
}

fn build_graph(
    registry: &Registry,
    chains: Vec<Chain>,
    avoided: &[&str],
) -> Result<Graph<GraphNode>, ParseError> {
    // Expand the source-providing (uridecodebin / playbin) and mid-chain
    // (decodebin) macros into concrete nodes before the structural build, so the
    // rest of the builder sees only real elements and pre-built nodes.
    let chains = expand_transcodebin(chains);
    let chains = expand_uri_sources(registry, chains, avoided)?;
    let chains = expand_decodebin(registry, chains, avoided)?;
    let chains = expand_encodebin(registry, chains)?;

    // A chain endpoint after flattening: a concrete element index, or a still
    // unresolved reference by name.
    enum Endpoint {
        Element(usize),
        Ref { name: String, pad: String },
    }

    let mut specs: Vec<ElementSpec> = Vec::new();
    // Parallel to `specs`: the pre-built node for that index (a `uridecodebin` /
    // `playbin` source or decoder), or `None` for a normal name-built element.
    // The placeholder spec for a pre-built node carries a benign name so the
    // structural closures (`is_queue` / `is_tee`) never match it, and node
    // construction uses the pre-built node instead of looking the name up.
    let mut prebuilt: Vec<Option<PrebuiltNode>> = Vec::new();
    let mut names: Vec<(String, usize)> = Vec::new();
    let mut chain_eps: Vec<Vec<Endpoint>> = Vec::with_capacity(chains.len());

    for chain in chains {
        let mut eps = Vec::with_capacity(chain.len());
        for item in chain {
            match item {
                Item::Element(spec) => {
                    let ei = specs.len();
                    if let Some(inst) = &spec.instance {
                        if names.iter().any(|(n, _)| n == inst) {
                            return Err(ParseError::DuplicateName(inst.clone()));
                        }
                        names.push((inst.clone(), ei));
                    }
                    specs.push(spec);
                    prebuilt.push(None);
                    eps.push(Endpoint::Element(ei));
                }
                Item::Prebuilt(node) => {
                    let ei = specs.len();
                    let name = match node {
                        PrebuiltNode::Source(_) => "uridecodebin",
                        PrebuiltNode::Element(_) => "(decoder)",
                    };
                    specs.push(ElementSpec {
                        name: name.to_string(),
                        props: Vec::new(),
                        instance: None,
                        log_category: None,
                    });
                    prebuilt.push(Some(node));
                    eps.push(Endpoint::Element(ei));
                }
                Item::Ref { name, pad } => eps.push(Endpoint::Ref { name, pad }),
            }
        }
        chain_eps.push(eps);
    }

    if specs.len() < 2 {
        return Err(ParseError::TooFewStages);
    }

    // Resolve references, collect the directed links (by element index), and
    // record each demux output-pad request (M476): a head-ref `d.video_0` that
    // sources a link contributes a `PadRequest` to `demux_pads[d]`, in link (port)
    // order, so a demux-select hook can map port i to the requested stream.
    let mut raw_links: Vec<(usize, usize)> = Vec::new();
    let mut demux_pads: Vec<Vec<PadRequest>> = alloc::vec![Vec::new(); specs.len()];
    // The DESTINATION pad request per raw link (M481): a named input-pad ref
    // (`... ! mux.audio_0`) carries a request; a bare `mux.` or an inline consumer
    // carries `None` (positional). The transpose of `demux_pads` (output side).
    let mut raw_dest_req: Vec<Option<PadRequest>> = Vec::new();
    for eps in &chain_eps {
        let mut idxs: Vec<usize> = Vec::with_capacity(eps.len());
        // The pad suffix of each endpoint that is a reference (`None` for a
        // concrete element), parallel to `idxs`.
        let mut pads: Vec<Option<&str>> = Vec::with_capacity(eps.len());
        for ep in eps {
            match ep {
                Endpoint::Element(ei) => {
                    idxs.push(*ei);
                    pads.push(None);
                }
                Endpoint::Ref { name, pad } => {
                    let i = names
                        .iter()
                        .find(|(n, _)| n == name)
                        .map(|(_, i)| *i)
                        .ok_or_else(|| ParseError::UnknownReference(name.clone()))?;
                    idxs.push(i);
                    pads.push(Some(pad.as_str()));
                }
            }
        }
        for w in 0..idxs.len().saturating_sub(1) {
            let (s, d) = (idxs[w], idxs[w + 1]);
            raw_links.push((s, d));
            // Record the source's output-pad request in port order: a pad-ref
            // source (`d.video_0`) carries a named request; an inline output
            // (`d ! x`) takes the positional Nth-forwardable-stream request. Only
            // consulted for an explicit-demux fan-out node (M476); harmless noise
            // for a normal element or a tee.
            let ordinal = demux_pads[s].len();
            let req = match pads[w] {
                Some(pad) => parse_pad_request(pad, ordinal),
                None => PadRequest {
                    kind: PadKind::Any,
                    index: ordinal,
                },
            };
            demux_pads[s].push(req);
            // The destination's input-pad request: a named ref (`mux.audio_0`)
            // parses; a bare `mux.` (empty suffix) or an inline consumer is `None`
            // (positional, resolved by the sequential input counter below).
            raw_dest_req.push(match pads[w + 1] {
                Some(pad) if !pad.is_empty() => Some(parse_pad_request(pad, 0)),
                _ => None,
            });
        }
    }

    // M190: `queue` / `queue2` is not an element in g2g. Per the design,
    // per-edge `LinkPolicy` (Block / DropOldest / DropNewest) is the leaky-queue
    // analog, so a queue node collapses into the backpressure policy of the edge
    // it sits on rather than becoming a buffering element + extra hop. Validate
    // each queue is 1-in/1-out, read its `leaky=`, then contract it out of the
    // link list, walking chains of queues to the first real consumer and keeping
    // the downstream-most leaky policy.
    let is_queue = |ei: usize| matches!(specs[ei].name.as_str(), "queue" | "queue2");
    let mut raw_in = alloc::vec![0usize; specs.len()];
    let mut raw_out = alloc::vec![0usize; specs.len()];
    for &(s, d) in &raw_links {
        raw_out[s] += 1;
        raw_in[d] += 1;
    }
    let mut queue_succ: Vec<Option<usize>> = alloc::vec![None; specs.len()];
    let mut queue_policy = alloc::vec![LinkPolicy::Block; specs.len()];
    let mut queue_capacity: Vec<Option<usize>> = alloc::vec![None; specs.len()];
    for ei in 0..specs.len() {
        if is_queue(ei) {
            if raw_in[ei] != 1 || raw_out[ei] != 1 {
                return Err(ParseError::QueueRole(specs[ei].name.clone()));
            }
            queue_policy[ei] = queue_leaky_policy(&specs[ei]);
            queue_capacity[ei] = queue_capacity_of(&specs[ei]);
            queue_succ[ei] = raw_links.iter().find(|(s, _)| *s == ei).map(|(_, d)| *d);
        }
    }
    // Each edge whose source is a real element walks through any run of queues to
    // its terminal consumer, carrying the accumulated policy; edges out of a queue
    // are consumed by that walk (skipped here).
    let mut links: Vec<(usize, usize, LinkPolicy, Option<usize>)> = Vec::new();
    // The destination input-pad request per contracted link (M481), aligned with
    // `links`; taken from the raw link that lands on the terminal consumer (so a
    // `... ! queue ! mux.audio_0` keeps its named pad through the queue contraction).
    let mut link_dest_req: Vec<Option<PadRequest>> = Vec::new();
    for (li, &(s, d)) in raw_links.iter().enumerate() {
        if is_queue(s) {
            continue;
        }
        let (mut cur, mut src_li) = (d, li);
        let mut policy = LinkPolicy::Block;
        let mut capacity: Option<usize> = None;
        while is_queue(cur) {
            if queue_policy[cur] != LinkPolicy::Block {
                policy = queue_policy[cur];
            }
            // A run of queues carries the last explicit depth (rare to chain them).
            if let Some(c) = queue_capacity[cur] {
                capacity = Some(c);
            }
            let next = queue_succ[cur].expect("queue validated 1-out above");
            // The named-pad suffix lives on the raw link that enters the terminal
            // consumer, i.e. `(cur -> next)`; find it for the request.
            src_li = raw_links
                .iter()
                .position(|&(a, b)| a == cur && b == next)
                .unwrap_or(src_li);
            cur = next;
        }
        links.push((s, cur, policy, capacity));
        link_dest_req.push(raw_dest_req[src_li].clone());
    }

    // Link degree per element fixes its role and any tee's output width. Computed
    // over the contracted links, so queue indices drop to degree 0 and are skipped
    // as nodes below.
    let mut in_deg = alloc::vec![0usize; specs.len()];
    let mut out_deg = alloc::vec![0usize; specs.len()];
    for &(s, d, _, _) in &links {
        out_deg[s] += 1;
        in_deg[d] += 1;
    }

    let is_tee = |ei: usize| specs[ei].name == "tee";
    // A non-tee node with several inbound links is a muxer (built from the
    // registry with that input count); a tee has a single input pad.
    let is_muxer = |ei: usize| !is_tee(ei) && in_deg[ei] > 1;
    // A node registered as a demuxer with several outbound links is a fan-out
    // demux (M210): the transpose of a muxer. A registered name with one output
    // falls back to its single-output launch element (e.g. `tsdemux`), the way a
    // one-input muxer name falls back to its single-input element.
    let is_demux = |ei: usize| !is_tee(ei) && out_deg[ei] > 1 && registry.is_demux(&specs[ei].name);
    // Explicit-demux fan-out (M476): a non-tee, non-registered-demux element that
    // fans out to several pads and is fed by a file source is built by a registered
    // demux-select hook, which probes the file (`location=`) and returns a
    // multi-output demuxer with one port per pad request (in reference order). This
    // is how `matroskademux` / `tsdemux` / `qtdemux` in a launch line split a file
    // into its elementary streams, honoring `d.video_0` / `d.audio_0` selection.
    let mut demux_select_node: Vec<Option<Box<dyn crate::runtime::DynMultiOutputElement>>> =
        (0..specs.len()).map(|_| None).collect();
    if !registry.demux_select_hooks().is_empty() {
        for ei in 0..specs.len() {
            if is_tee(ei) || is_demux(ei) || out_deg[ei] <= 1 || prebuilt[ei].is_some() {
                continue;
            }
            // The upstream file location (the source linking into this demux).
            let Some(location) = links
                .iter()
                .find(|(_, d, _, _)| *d == ei)
                .and_then(|(s, _, _, _)| prop(&specs[*s], "location"))
            else {
                continue;
            };
            for hook in registry.demux_select_hooks() {
                if let Some(demux) = hook(&specs[ei].name, location, &demux_pads[ei]) {
                    demux_select_node[ei] = Some(demux);
                    break;
                }
            }
        }
    }
    // `decodebin name=d` fan-out (M482): a `decodebin` node left unexpanded (it has
    // named refs) with a file source upstream probes the file, builds the
    // multi-output demuxer (stored like a demux-select node so it fans out on its
    // own pads), and records each port's elementary caps so the wiring below splices
    // a decoder onto every port (the decode-per-port that makes it `decodebin`, not a
    // bare demuxer). Declining hooks leave it unbuilt (a loud error at node build).
    let mut decode_fanout_caps: Vec<Option<Vec<Caps>>> = (0..specs.len()).map(|_| None).collect();
    if !registry.decodebin_select_hooks().is_empty() {
        for ei in 0..specs.len() {
            if !is_decodebin(&specs[ei].name) || out_deg[ei] == 0 || prebuilt[ei].is_some() {
                continue;
            }
            let Some(location) = links
                .iter()
                .find(|(_, d, _, _)| *d == ei)
                .and_then(|(s, _, _, _)| prop(&specs[*s], "location"))
            else {
                continue;
            };
            for hook in registry.decodebin_select_hooks() {
                if let Some((demux, caps)) = hook(location, &demux_pads[ei]) {
                    demux_select_node[ei] = Some(demux);
                    decode_fanout_caps[ei] = Some(caps);
                    break;
                }
            }
        }
    }
    let is_select: Vec<bool> = demux_select_node.iter().map(|d| d.is_some()).collect();
    // Auto-tee (M473): a non-tee, non-demux node whose single output fans out to
    // several consumers gets an implicit `tee` spliced in below, so a gst-launch
    // line that omits the explicit tee still builds. `tee`, registered demuxers,
    // and explicit-demux fan-out nodes fan out on their own pads and are left alone.
    let needs_tee = |ei: usize| {
        !is_tee(ei)
            && !is_demux(ei)
            && !is_select[ei]
            && !registry.is_fanout_src(&specs[ei].name)
            && out_deg[ei] > 1
    };
    // A fan-in element with no output is checked at construction below: a
    // terminal session (`is_terminal`) legally ends the graph (M713), a
    // merging muxer without a downstream stays `MuxerWithoutOutput`.
    for (ei, spec) in specs.iter().enumerate() {
        if is_tee(ei) && !spec.props.is_empty() {
            // The structural tee carries no element, so it has no properties.
            return Err(ParseError::UnknownProperty {
                element: "tee".to_string(),
                key: spec.props[0].0.clone(),
            });
        }
    }

    // Construct nodes in element-index order so `node_of[ei]` lines up. Queue
    // indices were contracted into edge policies above, so they get no node
    // (`None`); they never appear as an endpoint in the contracted links.
    let mut graph: Graph<GraphNode> = Graph::new();
    let mut node_of: Vec<Option<NodeId>> = Vec::with_capacity(specs.len());
    // The resolved muxer input-pad index per contracted link (M481), aligned with
    // `links`; filled at muxer construction from the element's `input_pad_index`
    // scheme. `None` for a non-muxer link or an unnamed ref (sequential fallback).
    let mut mux_pad_of_link: Vec<Option<u8>> = alloc::vec![None; links.len()];
    for ei in 0..specs.len() {
        if is_queue(ei) {
            node_of.push(None);
            continue;
        }
        // A pre-built node (uridecodebin / playbin source or decoder) is spliced
        // in directly; its role still follows link degree (a source has no input,
        // a terminal decoder no output).
        if let Some(node) = prebuilt[ei].take() {
            let nid = match node {
                PrebuiltNode::Source(src) => graph.add_source(GraphNodeRef::Source(src)),
                PrebuiltNode::Element(el) if out_deg[ei] == 0 => {
                    graph.add_sink(GraphNodeRef::Element(el))
                }
                PrebuiltNode::Element(el) => graph.add_transform(GraphNodeRef::Element(el)),
            };
            node_of.push(Some(nid));
            continue;
        }
        let spec = &specs[ei];
        let node = if is_tee(ei) {
            graph.add_tee(out_deg[ei] as u8).node()
        } else if in_deg[ei] == 0 && registry.is_fanout_src(&spec.name) {
            // Terminal fan-out source (M727): 0 inputs, one output per named
            // pad reference (`s. ! ...`). The element's intrinsic port count
            // must match the linked outputs.
            let mut src = registry
                .make_fanout_src(&spec.name, out_deg[ei])
                .ok_or_else(|| ParseError::UnknownElement(spec.name.clone()))?;
            if src.output_count() != out_deg[ei] {
                return Err(ParseError::UnknownInputPad(spec.name.clone()));
            }
            apply_props(&mut src, &spec.name, &spec.props)?;
            graph
                .add_fanout_src(GraphNodeRef::FanoutSource(src), out_deg[ei] as u8)
                .node()
        } else if in_deg[ei] == 0 {
            let mut src = registry
                .make_source(&spec.name)
                .ok_or_else(|| ParseError::UnknownSource(spec.name.clone()))?;
            apply_props(&mut src, &spec.name, &spec.props)?;
            graph.add_source(GraphNodeRef::Source(src))
        } else if is_muxer(ei) {
            let mut mux = registry
                .make_muxer(&spec.name, in_deg[ei])
                .ok_or_else(|| ParseError::NotAMuxer(spec.name.clone()))?;
            apply_props(&mut mux, &spec.name, &spec.props)?;
            // Resolve named input-pad refs (M481) to concrete indices via the
            // muxer's own scheme, so `... ! mux.audio_0  ... ! mux.video_0` routes
            // by name regardless of order. Named refs claim their index first;
            // bare refs fill the remaining slots in link order (the historical
            // positional behavior).
            let n = in_deg[ei];
            let incoming: Vec<usize> = links
                .iter()
                .enumerate()
                .filter(|(_, (_, d, _, _))| *d == ei)
                .map(|(k, _)| k)
                .collect();
            let mut used = alloc::vec![false; n];
            for (ord, &k) in incoming.iter().enumerate() {
                let Some(req) = &link_dest_req[k] else {
                    continue;
                };
                let idx = mux
                    .input_pad_index(req, ord)
                    .filter(|&i| i < n)
                    .ok_or_else(|| ParseError::UnknownInputPad(spec.name.clone()))?;
                if core::mem::replace(&mut used[idx], true) {
                    return Err(ParseError::DuplicateInputPad(spec.name.clone()));
                }
                mux_pad_of_link[k] = Some(idx as u8);
            }
            for &k in &incoming {
                if link_dest_req[k].is_none() {
                    let idx = used
                        .iter()
                        .position(|u| !u)
                        .expect("in_deg matches link count");
                    used[idx] = true;
                    mux_pad_of_link[k] = Some(idx as u8);
                }
            }
            if out_deg[ei] == 0 {
                // Nothing downstream: legal only for a terminal fan-in session
                // (M713), whose element consumes its inputs with no merged
                // output. A merging muxer here would silently drop its output.
                if !mux.is_terminal() {
                    return Err(ParseError::MuxerWithoutOutput(spec.name.clone()));
                }
                graph
                    .add_fanin_sink(GraphNodeRef::Muxer(mux), in_deg[ei] as u8)
                    .node()
            } else {
                graph
                    .add_muxer(GraphNodeRef::Muxer(mux), in_deg[ei] as u8)
                    .node()
            }
        } else if is_select[ei] {
            // M476: a demux-select hook already built the multi-output demuxer
            // (probing the upstream file); splice it in with one port per pad.
            let demux = demux_select_node[ei]
                .take()
                .expect("select demux built above");
            graph
                .add_demux(GraphNodeRef::Demux(demux), out_deg[ei] as u8)
                .node()
        } else if is_demux(ei) {
            let mut demux = registry
                .make_demux(&spec.name, out_deg[ei])
                .ok_or_else(|| ParseError::UnknownElement(spec.name.clone()))?;
            apply_props(&mut demux, &spec.name, &spec.props)?;
            graph
                .add_demux(GraphNodeRef::Demux(demux), out_deg[ei] as u8)
                .node()
        } else if out_deg[ei] == 0 {
            let mut el = registry
                .make_element(&spec.name)
                .ok_or_else(|| ParseError::UnknownElement(spec.name.clone()))?;
            apply_props(&mut el, &spec.name, &spec.props)?;
            graph.add_sink(GraphNodeRef::Element(el))
        } else {
            let mut el = registry
                .make_element(&spec.name)
                .ok_or_else(|| ParseError::UnknownElement(spec.name.clone()))?;
            apply_props(&mut el, &spec.name, &spec.props)?;
            graph.add_transform(GraphNodeRef::Element(el))
        };
        // M842: a `name=` is the element's instance name at run time, not just
        // the handle pad references resolve against.
        if let Some(inst) = &spec.instance {
            graph.set_node_name(node, inst.clone());
        }
        // M847: `log-category=` renames this instance's log category, which the
        // runner hands to the element before naming it.
        if let Some(cat) = &spec.log_category {
            graph.set_node_log_category(node, cat.clone());
        }
        node_of.push(Some(node));
    }

    // Auto-tee (M473): for each fan-out node that is not itself a tee/demux, splice
    // an implicit `tee` onto its output. The node's single output pad feeds the
    // tee (a plain blocking link); the tee's `out_deg` pads feed the consumers,
    // which the edge loop below sources from the tee instead of the node.
    let mut implicit_tee: Vec<Option<NodeId>> = alloc::vec![None; specs.len()];
    for ei in 0..specs.len() {
        if needs_tee(ei) {
            if let Some(src_node) = node_of[ei] {
                let tee = graph.add_tee(out_deg[ei] as u8).node();
                graph.link_with(PadId::from(src_node), PadId::from(tee), LinkPolicy::Block)?;
                implicit_tee[ei] = Some(tee);
            }
        }
    }

    // Wire edges. Each tee or demux branch takes a distinct output pad (0..n) and
    // each muxer input a distinct input pad (0..n); every other output and input
    // is pad 0. A queue's `leaky=` rides along as the edge's `LinkPolicy`.
    let mut tee_next = alloc::vec![0u8; specs.len()];
    for (k, &(s, d, policy, capacity)) in links.iter().enumerate() {
        let node_s = node_of[s].expect("contracted link source is a real node");
        let node_d = node_of[d].expect("contracted link destination is a real node");
        let src = if let Some(tee) = implicit_tee[s] {
            // The fan-out node's consumers source from its spliced-in tee's pads.
            let index = tee_next[s];
            tee_next[s] += 1;
            PadId { node: tee, index }
        } else if is_tee(s) || is_demux(s) || is_select[s] || registry.is_fanout_src(&specs[s].name)
        {
            let index = tee_next[s];
            tee_next[s] += 1;
            PadId {
                node: node_s,
                index,
            }
        } else {
            PadId::from(node_s)
        };
        let dst = if is_muxer(d) {
            // The input index was resolved at construction (named pads via the
            // muxer's scheme, bare refs sequentially); every muxer link has one.
            let index = mux_pad_of_link[k].expect("muxer link assigned an input pad");
            PadId {
                node: node_d,
                index,
            }
        } else {
            PadId::from(node_d)
        };
        // `decodebin` fan-out (M482): splice a decoder between the demux port and the
        // branch consumer instead of a bare link, so each `d.video_0` / `d.audio_0`
        // branch receives DECODED (raw) frames. A text port carries `Text{Utf8}`
        // already, so it links straight through (no codec).
        if let Some(caps) = &decode_fanout_caps[s] {
            let port = src.index as usize;
            let kind = demux_pads[s]
                .get(port)
                .map(|r| r.kind)
                .unwrap_or(PadKind::Any);
            if !matches!(kind, PadKind::Text) {
                let target: &dyn Fn(&Caps) -> bool = match kind {
                    PadKind::Audio => &is_raw_audio,
                    _ => &is_raw_video,
                };
                registry
                    .decodebin(
                        &mut graph,
                        src,
                        dst,
                        &caps[port],
                        target,
                        DECODEBIN_MAX_DEPTH,
                    )
                    .map_err(|_| ParseError::NoDecodeChain(alloc::format!("{:?}", caps[port])))?;
                continue;
            }
        }
        graph.link_full(src, dst, policy, capacity)?;
    }

    Ok(graph)
}

/// Map a `queue` / `queue2` node's `leaky=` property to the edge backpressure
/// policy it stands in for (M190). gst accepts the enum by value or nick:
/// `0`/`no` (lossless, the default), `1`/`upstream` (drop the newest incoming
/// buffer), `2`/`downstream` (drop the oldest queued buffer). The other buffering
/// bounds (`max-size-bytes` / `max-size-time`, `min-threshold-*`, `silent`) are
/// accepted but not modeled (g2g's link depth is a buffer count), so they are
/// ignored for paste compatibility rather than rejected; `max-size-buffers` maps
/// to the edge depth, see [`queue_capacity_of`].
fn queue_leaky_policy(spec: &ElementSpec) -> LinkPolicy {
    for (k, v) in &spec.props {
        if k == "leaky" {
            return match v.as_str() {
                "1" | "upstream" => LinkPolicy::DropNewest,
                "2" | "downstream" => LinkPolicy::DropOldest,
                // "0" / "no" and anything unrecognized: lossless block.
                _ => LinkPolicy::Block,
            };
        }
    }
    LinkPolicy::Block
}

/// A `queue max-size-buffers=N` sets the depth of the edge it contracts to (the
/// gst per-queue buffer bound), overriding the runner's graph-wide `link_capacity`
/// for just that link. `0` in gst means "unbounded"; g2g has no unbounded channel,
/// so a `0` is ignored (falls back to the default depth). An unparseable value is
/// ignored too (paste compatibility).
fn queue_capacity_of(spec: &ElementSpec) -> Option<usize> {
    spec.props
        .iter()
        .find(|(k, _)| k == "max-size-buffers")
        .and_then(|(_, v)| v.parse::<usize>().ok())
        .filter(|&n| n > 0)
}

/// Parse a `gst-launch`-style pipeline string into a runnable [`Graph`], building
/// each element by name from `registry`, applying its `key=value` properties, and
/// linking the chains (including `tee` branches) into the DAG. Roles follow
/// connectivity. The result drops straight onto
/// [`run_graph`](crate::runtime::run_graph).
///
/// ```text
/// videotestsrc num-buffers=3 ! tee name=t ! fakesink   t. ! videoflip ! fakesink
/// ```
pub fn parse_launch(registry: &Registry, pipeline: &str) -> Result<Graph<GraphNode>, ParseError> {
    parse_launch_avoiding(registry, pipeline, &[])
}

/// [`parse_launch`], with the auto-plug search forbidden from picking any
/// factory named in `avoided` (M1023). The line is otherwise parsed identically,
/// so an element the text names explicitly is still built: this bounds only what
/// a `decodebin` / `uridecodebin` / `playbin` may choose. An application retries
/// through here after a chosen decoder turned out not to decode the stream.
pub fn parse_launch_avoiding(
    registry: &Registry,
    pipeline: &str,
    avoided: &[&str],
) -> Result<Graph<GraphNode>, ParseError> {
    // The parser's own built-ins are not registry factories, but they are
    // element names all the same, so a chain may start on one.
    let knows = |name: &str| {
        registry.knows_element(name)
            || matches!(
                name,
                "queue"
                    | "queue2"
                    | "decodebin"
                    | "uridecodebin"
                    | "playbin"
                    | "encodebin"
                    | "encodebin2"
                    | "transcodebin"
            )
    };
    let chains = parse_chains_with(pipeline, &knows)?;
    // playbin uri=X auto-fan-out (M382): a lone `playbin uri=` probes the
    // container via the registered hook and auto-builds source -> demux ->
    // per-stream decode -> auto sinks (multi-stream). Without a hook (or if it
    // declines, e.g. a non-Matroska file), fall through to build_graph, which
    // expands `playbin` to the single-stream pipeline (M196).
    if let Some(uri) = lone_playbin_uri(&chains) {
        // Try each registered hook (one per container type) until one handles the
        // URI; a hook returns Ok(None) to decline a container it does not parse.
        for hook in registry.playbin_hooks() {
            if let Some(graph) = hook(registry, uri)? {
                return Ok(splice_domain_converters(registry, graph));
            }
        }
    }
    Ok(splice_domain_converters(
        registry,
        build_graph(registry, chains, avoided)?,
    ))
}

/// The factory to bar from the auto-plug search after a run ended in
/// `failed_element`, or `None` when re-plugging cannot help (M1023). Pass the
/// result to [`parse_launch_avoiding`] (accumulating it into `avoided`) and run
/// the line again; `None` means report the failure instead.
///
/// The caller owns the rest of the policy, because only it knows whether a
/// restart is safe: retry only while nothing has been presented (a run that
/// output nothing restarts invisibly, one that did not would repeat itself) and
/// not for a live source ([`has_live_source`], where a restart reopens the
/// connection and drops what was in flight).
///
/// `None` when the failing element is unknown to the registry, when the line
/// names it explicitly (the user asked for that element by name, so substituting
/// another would defy them), or when it has already been barred once.
pub fn fallback_factory(
    registry: &Registry,
    pipeline: &str,
    failed_element: &str,
    avoided: &[&str],
) -> Option<&'static str> {
    let factory = registry.factory_of_instance(failed_element)?;
    if avoided.contains(&factory) {
        return None;
    }
    let named_in_line = tokenize(pipeline).iter().any(|t| t == factory);
    (!named_in_line).then_some(factory)
}

/// Whether any element in `graph` offers a live-source clock: a capture or
/// network source, which a restart would reopen at the cost of whatever was in
/// flight. The signal a caller checks before re-running a pipeline it has already
/// started (see [`fallback_factory`]).
pub fn has_live_source(graph: &Graph<GraphNode>) -> bool {
    (0..graph.node_count())
        .filter_map(|i| graph.element(NodeId(i as u32)))
        .filter_map(|node| match node {
            GraphNodeRef::Source(source) => source.provide_clock(),
            GraphNodeRef::Element(element) => element.provide_clock(),
            _ => None,
        })
        .any(|candidate| candidate.priority == crate::clock::ClockPriority::LiveSource)
}

/// Splice memory-domain converters into a just-parsed graph (M1017), where the
/// registry carries a factory: a text pipeline that links a GPU decoder to a sink
/// in another domain gets the bridge (or the download) it needs without naming
/// it. A no-op when the domains already agree, or when no factory is registered.
fn splice_domain_converters(registry: &Registry, graph: Graph<GraphNode>) -> Graph<GraphNode> {
    match registry.domain_converter() {
        Some(factory) => crate::runtime::auto_plug_domain_converters(graph, &factory),
        None => graph,
    }
}

/// The `uri=` of a pipeline that is a single bare `playbin uri=X` element (and
/// nothing else), the M382 multi-stream auto-fan-out trigger. `None` for any
/// other shape: a `playbin` mid-pipeline, alongside other elements, or without a
/// `uri=` is left to the normal builder (the M196 single-stream expansion).
fn lone_playbin_uri(chains: &[Chain]) -> Option<&str> {
    let [chain] = chains else { return None };
    let [Item::Element(spec)] = chain.as_slice() else {
        return None;
    };
    if spec.name != "playbin" {
        return None;
    }
    prop(spec, "uri")
}

#[cfg(test)]
mod tests {
    use super::*;

    fn item_names(chain: &Chain) -> Vec<&str> {
        chain
            .iter()
            .map(|i| match i {
                Item::Element(s) => s.name.as_str(),
                Item::Ref { name, .. } => name.as_str(),
                Item::Prebuilt(_) => "(prebuilt)",
            })
            .collect()
    }

    #[test]
    fn parse_chains_splits_names_and_props() {
        let chains = parse_chains(
            "videotestsrc num-buffers=3 pattern=snow ! videoflip method=rotate-180 ! fakesink",
        )
        .unwrap();
        assert_eq!(chains.len(), 1);
        assert_eq!(
            item_names(&chains[0]),
            ["videotestsrc", "videoflip", "fakesink"]
        );
        let Item::Element(src) = &chains[0][0] else {
            panic!("first is an element")
        };
        assert_eq!(
            src.props,
            [
                ("num-buffers".to_string(), "3".to_string()),
                ("pattern".into(), "snow".into())
            ]
        );
        let Item::Element(sink) = &chains[0][2] else {
            panic!("last is an element")
        };
        assert!(sink.props.is_empty());
    }

    #[test]
    fn parse_chains_strips_quoted_values() {
        // A double-quoted value has its quotes stripped.
        let chains = parse_chains("filesrc location=\"file.mp4\" ! fakesink").unwrap();
        let Item::Element(src) = &chains[0][0] else {
            panic!("element")
        };
        assert_eq!(
            src.props[0],
            ("location".to_string(), "file.mp4".to_string())
        );
    }

    #[test]
    fn parse_chains_keeps_spaces_in_quoted_values() {
        // The quote-aware tokenizer keeps spaces inside a value, so a nested
        // element description (the `gstwrap` case) survives as one property.
        let chains = parse_chains("gstwrap element=\"x264enc bitrate=4000\" ! fakesink").unwrap();
        assert_eq!(item_names(&chains[0]), ["gstwrap", "fakesink"]);
        let Item::Element(w) = &chains[0][0] else {
            panic!("element")
        };
        assert_eq!(
            w.props[0],
            ("element".to_string(), "x264enc bitrate=4000".to_string())
        );
    }

    #[test]
    fn parse_chains_keeps_bang_inside_quotes() {
        // A `!` inside a quoted value is literal, not a stage separator: one
        // element with one property, not two chained nodes.
        let chains = parse_chains("gstwrap element=\"a ! b\" ! fakesink").unwrap();
        assert_eq!(item_names(&chains[0]), ["gstwrap", "fakesink"]);
        let Item::Element(w) = &chains[0][0] else {
            panic!("element")
        };
        assert_eq!(w.props[0], ("element".to_string(), "a ! b".to_string()));
    }

    #[test]
    fn tokenize_treats_quoted_region_as_one_token() {
        assert_eq!(
            tokenize("gstwrap element=\"x y\" ! sink"),
            ["gstwrap", "element=\"x y\"", "!", "sink"]
        );
    }

    #[test]
    fn tokenize_treats_single_quoted_region_as_one_token() {
        assert_eq!(
            tokenize("filesink location='/my file.ts' ! sink"),
            ["filesink", "location='/my file.ts'", "!", "sink"]
        );
    }

    #[test]
    fn single_quoted_value_is_unquoted() {
        let chains = parse_chains("videotestsrc ! identity note='a b c' ! fakesink").unwrap();
        let Item::Element(id) = &chains[0][1] else {
            panic!("element")
        };
        assert_eq!(id.props, [("note".to_string(), "a b c".to_string())]);
    }

    #[test]
    fn escaped_space_keeps_a_value_whole() {
        // gst-launch 1.26 does the same: `\ ` is a literal space, not a token
        // break (verified against `filesink location=`).
        assert_eq!(
            tokenize(r"filesink location=/my\ file.ts ! sink"),
            ["filesink", r"location=/my\ file.ts", "!", "sink"]
        );
        let chains = parse_chains(r"videotestsrc ! filesink location=/my\ file.ts").unwrap();
        let Item::Element(fs) = &chains[0][1] else {
            panic!("element")
        };
        assert_eq!(fs.props, [("location".to_string(), "/my file.ts".into())]);
    }

    #[test]
    fn escaped_quote_does_not_close_the_region() {
        // `"a\"b"` is one token holding `a"b`, as in gst-launch.
        assert_eq!(
            tokenize(r#"identity note="a\" b" ! sink"#),
            ["identity", r#"note="a\" b""#, "!", "sink"]
        );
        let chains = parse_chains(r#"videotestsrc ! identity note="a\" b""#).unwrap();
        let Item::Element(id) = &chains[0][1] else {
            panic!("element")
        };
        assert_eq!(id.props, [("note".to_string(), "a\" b".into())]);
    }

    #[test]
    fn quoted_region_may_sit_mid_value() {
        // gst-launch rejects this ("no element ..."): its lexer only quotes a
        // whole value. g2g accepts the superset, so a path with one spaced
        // component needs no quoting of the rest.
        let chains =
            parse_chains(r#"videotestsrc ! filesink location=/tmp/"my dir"/a.ts"#).unwrap();
        let Item::Element(fs) = &chains[0][1] else {
            panic!("element")
        };
        assert_eq!(
            fs.props,
            [("location".to_string(), "/tmp/my dir/a.ts".into())]
        );
    }

    #[test]
    fn unescaped_backslash_stays_literal() {
        // A Windows path survives unescaped. gst-launch would drop these
        // backslashes (`videos\a.ts` opens a file called `videosa.ts`), so this
        // is a deliberate deviation: `\` escapes only the launch-special
        // characters.
        assert_eq!(unquote_value(r"C:\videos\a.ts"), r"C:\videos\a.ts");
        assert_eq!(unquote_value(r"a\ b"), "a b");
        assert_eq!(unquote_value(r"a\\b"), r"a\b");
    }

    #[test]
    fn hash_starts_a_comment_to_end_of_line() {
        // Trailing comment on one line, and a comment mid-pipeline across lines.
        let chains = parse_chains("videotestsrc ! fakesink # trailing note").unwrap();
        assert_eq!(item_names(&chains[0]), ["videotestsrc", "fakesink"]);
        let chains = parse_chains("videotestsrc  # the source\n  ! fakesink").unwrap();
        assert_eq!(item_names(&chains[0]), ["videotestsrc", "fakesink"]);
    }

    #[test]
    fn hash_inside_a_value_is_literal_not_a_comment() {
        // A URI fragment (`#closed-captions=cc1`, `#t=10`) must survive; `#` is a
        // comment only at a token boundary.
        assert_eq!(
            tokenize("uridecodebin uri=file:///v.mp4#closed-captions=cc1 ! sink"),
            [
                "uridecodebin",
                "uri=file:///v.mp4#closed-captions=cc1",
                "!",
                "sink"
            ]
        );
    }

    #[test]
    fn caps_range_value_error_carries_a_syntax_hint() {
        let e = ParseError::BadValue {
            element: "capsfilter".to_string(),
            key: "caps".to_string(),
            value: "video/x-raw,width=[1,1920]".to_string(),
        };
        let msg = alloc::format!("{e}");
        assert!(msg.contains("ranges"), "caps range hint present: {msg}");
        // A plain element property error keeps the bare message (no caps hint).
        let plain = ParseError::BadValue {
            element: "videobox".to_string(),
            key: "top".to_string(),
            value: "abc".to_string(),
        };
        assert!(!alloc::format!("{plain}").contains("ranges"));
    }

    #[test]
    fn enum_value_error_lists_the_valid_nicks() {
        let e = ParseError::BadEnumValue {
            element: "videoflip".to_string(),
            key: "method".to_string(),
            value: "sideways".to_string(),
            values: "none | clockwise",
        };
        let msg = alloc::format!("{e}");
        assert!(msg.contains("videoflip"), "{msg}");
        assert!(msg.contains("'method'"), "{msg}");
        assert!(msg.contains("valid: none | clockwise"), "{msg}");
        assert!(!msg.contains("flag set"), "no flag hint for a plain enum");
        // A malformed flag set (the whole value is reported) also hints at the
        // `+` syntax.
        let flags = ParseError::BadEnumValue {
            element: "rtspsrc".to_string(),
            key: "protocols".to_string(),
            value: "udp+".to_string(),
            values: "udp | tcp",
        };
        assert!(alloc::format!("{flags}").contains("flag set"));
    }

    #[test]
    fn caps_description_becomes_capsfilter() {
        // A bare `media/type,...` node is the inline caps-filter shorthand.
        let chains =
            parse_chains("videotestsrc ! video/x-raw,format=nv12,width=320 ! fakesink").unwrap();
        assert_eq!(
            item_names(&chains[0]),
            ["videotestsrc", "capsfilter", "fakesink"]
        );
        let Item::Element(caps) = &chains[0][1] else {
            panic!("element")
        };
        assert_eq!(
            caps.props,
            [(
                "caps".to_string(),
                "video/x-raw,format=nv12,width=320".to_string()
            )]
        );
    }

    #[test]
    fn tee_branch_parses_into_two_chains() {
        // `name=` is the instance handle (not a property); `t.` opens the branch.
        let chains =
            parse_chains("videotestsrc ! tee name=t ! fakesink t. ! videoflip ! fakesink").unwrap();
        assert_eq!(chains.len(), 2);
        assert_eq!(item_names(&chains[0]), ["videotestsrc", "tee", "fakesink"]);
        assert_eq!(item_names(&chains[1]), ["t", "videoflip", "fakesink"]);
        let Item::Element(tee) = &chains[0][1] else {
            panic!("element")
        };
        assert_eq!(tee.instance.as_deref(), Some("t"));
        assert!(tee.props.is_empty(), "name= is the handle, not a property");
        assert!(matches!(&chains[1][0], Item::Ref { name, .. } if name == "t"));
    }

    #[test]
    fn empty_and_too_few_stages_error() {
        let reg = Registry::new();
        assert!(matches!(parse_launch(&reg, "   "), Err(ParseError::Empty)));
        assert!(matches!(
            parse_launch(&reg, "videotestsrc"),
            Err(ParseError::TooFewStages)
        ));
    }

    #[test]
    fn malformed_property_is_reported() {
        assert!(matches!(
            parse_chains("videotestsrc bogus ! fakesink"),
            Err(ParseError::MalformedProperty { .. })
        ));
    }

    /// gst-launch runs several top-level chains in one pipeline
    /// (`videotestsrc ! xvimagesink audiotestsrc ! pulsesink`); a bare element
    /// name after a completed chain opens the next one. Before, the sink ate it
    /// as a property and reported it malformed.
    #[test]
    fn a_second_chain_may_follow_a_sink() {
        let known = |n: &str| matches!(n, "videotestsrc" | "fakesink" | "filesink");
        let chains = parse_chains_with("videotestsrc ! fakesink videotestsrc ! fakesink", &known)
            .expect("two chains parse");
        let shapes: Vec<Vec<&str>> = chains.iter().map(item_names).collect();
        assert_eq!(
            shapes,
            [["videotestsrc", "fakesink"], ["videotestsrc", "fakesink"]],
            "one chain per source"
        );
    }

    /// The property case must not regress: a bare token naming nothing
    /// registered is still the typo it looks like, not a new chain.
    #[test]
    fn a_bare_token_that_names_nothing_is_still_a_malformed_property() {
        let known = |n: &str| matches!(n, "videotestsrc" | "fakesink" | "filesink");
        assert!(matches!(
            parse_chains_with("videotestsrc bogus ! fakesink", &known),
            Err(ParseError::MalformedProperty { .. })
        ));
        // And after a sink, where the new-chain rule applies.
        assert!(matches!(
            parse_chains_with("videotestsrc ! fakesink bogus", &known),
            Err(ParseError::MalformedProperty { .. })
        ));
    }

    /// A property whose value happens to name an element is still a property:
    /// the `=` decides.
    #[test]
    fn a_key_value_token_is_never_a_chain_head() {
        let known = |n: &str| matches!(n, "videotestsrc" | "fakesink" | "filesink");
        let chains =
            parse_chains_with("videotestsrc ! filesink location=fakesink", &known).expect("parses");
        let shapes: Vec<Vec<&str>> = chains.iter().map(item_names).collect();
        assert_eq!(shapes, [["videotestsrc", "filesink"]], "one chain");
    }

    #[test]
    fn unknown_reference_is_reported() {
        // The degree / reference checks precede registry construction, so an
        // empty registry still surfaces them.
        let reg = Registry::new();
        let err = parse_launch(
            &reg,
            "videotestsrc ! tee name=t ! fakesink nope. ! fakesink",
        )
        .unwrap_err();
        assert_eq!(err, ParseError::UnknownReference("nope".to_string()));
    }

    #[test]
    fn duplicate_name_is_reported() {
        let reg = Registry::new();
        let err =
            parse_launch(&reg, "videotestsrc name=x ! videoflip name=x ! fakesink").unwrap_err();
        assert_eq!(err, ParseError::DuplicateName("x".to_string()));
    }

    // Auto-tee (M473): fan-out without an explicit `tee` no longer errors; the
    // parser splices one in. Covered end to end (build + run + topology) in
    // g2g-plugins/tests/m118_launch_branching.rs, where real elements exist.
}