streamer45
diff --git a/‎crates/api/src/yaml.rs‎
Lines changed: 19 additions & 0 deletions b/‎crates/api/src/yaml.rs‎
Lines changed: 19 additions & 0 deletions
diff --git a/‎crates/core/src/types.rs‎
Lines changed: 1 addition & 0 deletions b/‎crates/core/src/types.rs‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎crates/engine/src/oneshot.rs‎
Lines changed: 10 additions & 5 deletions b/‎crates/engine/src/oneshot.rs‎
Lines changed: 10 additions & 5 deletions
diff --git a/‎crates/nodes/src/containers/mp4.rs‎
Lines changed: 255 additions & 42 deletions b/‎crates/nodes/src/containers/mp4.rs‎
Lines changed: 255 additions & 42 deletions
diff --git a/‎crates/nodes/src/transport/moq/constants.rs‎
Lines changed: 68 additions & 2 deletions b/‎crates/nodes/src/transport/moq/constants.rs‎
Lines changed: 68 additions & 2 deletions
diff --git a/‎crates/nodes/src/transport/moq/peer/config.rs‎
Lines changed: 41 additions & 1 deletion b/‎crates/nodes/src/transport/moq/peer/config.rs‎
Lines changed: 41 additions & 1 deletion
@@ -1636,6 +1636,25 @@ nodes:
         );
     }
 
+    /// Regression test: `moq_aac_mixing.yml` must compile without errors.
+    ///
+    /// Previously the mixer's `needs` used bracket syntax (`mic_gain[in_0]`)
+    /// which the parser does not support, causing a "references non-existent
+    /// node" error.  The fix switched to simple array syntax so that
+    /// `Needs::Multiple` auto-generates `in_0`, `in_1` by index.
+    #[test]
+    fn test_sample_moq_aac_mixing_compiles() {
+        let yaml = include_str!("../../../samples/pipelines/dynamic/moq_aac_mixing.yml");
+        let user_pipeline = parse_yaml(yaml).unwrap();
+        let result = compile(user_pipeline);
+
+        assert!(
+            result.is_ok(),
+            "Sample pipeline moq_aac_mixing.yml should compile: {:?}",
+            result.err()
+        );
+    }
+
     #[test]
     #[allow(clippy::unwrap_used, clippy::expect_used)]
     fn test_multiple_inputs_numbered_pins() {
 
@@ -63,6 +63,7 @@ pub struct RawVideoFormat {
 #[non_exhaustive]
 pub enum AudioCodec {
     Opus,
+    Aac,
 }
 
 /// Supported encoded video codecs.
 
@@ -477,16 +477,21 @@ impl Engine {
         // --- 5.5. Start source / generator nodes ---
         // File readers need an explicit Start signal, and so do generator nodes
         // (e.g. video::colorbars) that follow the Ready → Start lifecycle.
-        // In generator mode we find root nodes (never a to_node in any connection)
-        // and send them Start as well.
+        // We always scan for root nodes (never a to_node in any connection) so
+        // that mixed pipelines (e.g. http_input + colorbars) work correctly.
+        // http_input nodes are excluded because they are driven by the incoming
+        // HTTP stream rather than a Start signal.
         let mut start_node_ids: Vec<String> = source_node_ids.clone();
 
-        if !has_http_input && source_node_ids.is_empty() {
-            // Generator mode — find root nodes that need a Start signal.
+        {
             let downstream_nodes: std::collections::HashSet<&str> =
                 definition.connections.iter().map(|c| c.to_node.as_str()).collect();
             for name in definition.nodes.keys() {
-                if name != &output_node_id && !downstream_nodes.contains(name.as_str()) {
+                if name != &output_node_id
+                    && !downstream_nodes.contains(name.as_str())
+                    && !start_node_ids.contains(name)
+                    && !http_input_nodes.contains(name)
+                {
                     start_node_ids.push(name.clone());
                 }
             }
 
@@ -12,13 +12,22 @@ use streamkit_core::types::{
     AudioCodec, EncodedAudioFormat, EncodedVideoFormat, PacketMetadata, PacketType, VideoCodec,
 };
 
-pub const DEFAULT_AUDIO_FRAME_DURATION_US: u64 = 20_000;
+pub const DEFAULT_AUDIO_FRAME_DURATION_US_OPUS: u64 = 20_000;
+pub const DEFAULT_AUDIO_FRAME_DURATION_US_AAC: u64 = 21_333;
+
+/// Return the default audio frame duration for the given codec.
+pub const fn default_audio_frame_duration_us(codec: AudioCodec) -> u64 {
+    match codec {
+        AudioCodec::Aac => DEFAULT_AUDIO_FRAME_DURATION_US_AAC,
+        _ => DEFAULT_AUDIO_FRAME_DURATION_US_OPUS,
+    }
+}
 
 pub fn packet_duration_us(metadata: Option<&PacketMetadata>) -> Option<u64> {
     metadata.and_then(|m| m.duration_us).filter(|d| *d > 0)
 }
 
-/// Return the accepted media types for dynamic MoQ pins (Opus audio + VP9/AV1/H264 video).
+/// Return the accepted media types for dynamic MoQ pins (Opus/AAC audio + VP9/AV1/H264 video).
 ///
 /// This is shared across `moq_peer` and `moq_push` to avoid duplicating the
 /// type construction in every `RequestAddInputPin` / `RequestAddOutputPin` handler.
@@ -28,6 +37,10 @@ pub fn moq_accepted_media_types() -> Vec<PacketType> {
             codec: AudioCodec::Opus,
             codec_private: None,
         }),
+        PacketType::EncodedAudio(EncodedAudioFormat {
+            codec: AudioCodec::Aac,
+            codec_private: None,
+        }),
         PacketType::EncodedVideo(EncodedVideoFormat {
             codec: VideoCodec::Vp9,
             bitstream_format: None,
@@ -132,6 +145,59 @@ pub fn parse_video_codec_config(s: &str) -> Option<VideoCodec> {
     }
 }
 
+/// Build the [`hang::catalog::AudioCodec`] entry for a given [`AudioCodec`].
+///
+/// Centralises the AAC/Opus catalog construction, mirroring
+/// [`catalog_video_codec`] for the video side.
+pub fn catalog_audio_codec(codec: AudioCodec) -> hang::catalog::AudioCodec {
+    match codec {
+        AudioCodec::Opus => hang::catalog::AudioCodec::Opus,
+        AudioCodec::Aac => hang::catalog::AudioCodec::AAC(hang::catalog::AAC { profile: 2 }),
+        // Future-proof: fall back to Opus and warn.
+        _ => {
+            tracing::warn!(?codec, "unsupported AudioCodec for MoQ catalog, defaulting to Opus");
+            hang::catalog::AudioCodec::Opus
+        },
+    }
+}
+
+/// Resolve the audio codec from config → input_types → default (Opus).
+///
+/// Priority order:
+/// 1. Explicit `audio_codec` config param (required for dynamic pipelines)
+/// 2. Auto-detected from `input_types` (static pipelines)
+/// 3. Default: Opus
+///
+/// Shared by `moq_peer` and `moq_push` to avoid duplicating the resolution chain.
+pub fn resolve_audio_codec(
+    config_codec: Option<&str>,
+    input_types: &std::collections::HashMap<String, PacketType>,
+) -> AudioCodec {
+    config_codec
+        .and_then(parse_audio_codec_config)
+        .or_else(|| {
+            input_types.iter().find_map(|(_, pt)| match pt {
+                PacketType::EncodedAudio(fmt) => Some(fmt.codec),
+                _ => None,
+            })
+        })
+        .unwrap_or(AudioCodec::Opus)
+}
+
+/// Parse an `audio_codec` config string (e.g. `"opus"`, `"aac"`) into the
+/// corresponding [`AudioCodec`].  Returns `None` for unrecognised values so
+/// the caller can fall back to auto-detection.
+pub fn parse_audio_codec_config(s: &str) -> Option<AudioCodec> {
+    match s.to_ascii_lowercase().as_str() {
+        "opus" => Some(AudioCodec::Opus),
+        "aac" => Some(AudioCodec::Aac),
+        _ => {
+            tracing::warn!(audio_codec = %s, "unrecognised audio_codec config value — ignoring");
+            None
+        },
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
@@ -11,7 +11,7 @@ use serde::Deserialize;
 use std::collections::HashMap;
 use std::sync::Arc;
 use streamkit_core::timing::MediaClock;
-use streamkit_core::types::{Packet, PacketType, VideoCodec};
+use streamkit_core::types::{AudioCodec, Packet, PacketType, VideoCodec};
 use tokio::sync::{broadcast, mpsc, watch, Semaphore};
 
 // ── Internal helper types ────────────────────────────────────────────────────
@@ -89,6 +89,16 @@ pub(super) enum PublisherEvent {
     Disconnected { path: String, error: Option<String> },
 }
 
+/// Resolved codec pair for video and audio.
+///
+/// Bundles the two codec fields that would otherwise be passed as separate
+/// parameters to `handle_pin_management` and friends.
+#[derive(Debug, Clone, Copy)]
+pub(super) struct MediaCodecConfig {
+    pub video: VideoCodec,
+    pub audio: AudioCodec,
+}
+
 /// Media and output configuration shared across subscriber-related functions.
 pub(super) struct SubscriberMediaConfig {
     pub has_video: bool,
@@ -98,6 +108,7 @@ pub(super) struct SubscriberMediaConfig {
     pub output_group_duration_ms: u64,
     pub output_initial_delay_ms: u64,
     pub video_codec: VideoCodec,
+    pub audio_codec: AudioCodec,
 }
 
 pub(super) struct BidirectionalTaskConfig {
@@ -157,6 +168,8 @@ pub(super) struct SubscriberSendCtx<'a> {
     pub last_audio_ts_ms: Option<u64>,
     pub last_video_ts_ms: Option<u64>,
     pub stats_delta_tx: &'a mpsc::Sender<NodeStatsDelta>,
+    /// Resolved audio codec — used for codec-aware default frame durations.
+    pub audio_codec: AudioCodec,
 }
 
 // ── Shared map of dynamic output pin senders ─────────────────────────────────
@@ -313,6 +326,31 @@ pub struct MoqPeerConfig {
     /// is auto-detected from `input_types` (static pipelines) and falls back
     /// to VP9.
     pub video_codec: Option<String>,
+    /// Audio codec for the MoQ catalog.
+    ///
+    /// Required for dynamic pipelines where `input_types` is not available at
+    /// startup.  Accepted values: `"opus"`, `"aac"`.  When `None`, the codec
+    /// is auto-detected from `input_types` (static pipelines) and falls back
+    /// to Opus.
+    ///
+    /// Controls the **publisher output pin** type (`audio/data`).  For
+    /// transcoding scenarios where the subscriber receives a different codec
+    /// (e.g. Opus in → AAC out), use [`subscriber_audio_codec`] to override
+    /// the subscriber catalog codec independently.
+    pub audio_codec: Option<String>,
+    /// Audio codec advertised in the **subscriber** MoQ catalog.
+    ///
+    /// When set, overrides [`audio_codec`] for the subscriber side only
+    /// (catalog, frame duration).  The publisher output pin (`audio/data`)
+    /// continues to use [`audio_codec`].
+    ///
+    /// Useful for transcoding pipelines where the publisher sends one codec
+    /// (e.g. Opus) but the pipeline re-encodes to another (e.g. AAC) before
+    /// feeding it back to subscribers.
+    ///
+    /// Accepted values: `"opus"`, `"aac"`.  When `None`, falls back to
+    /// [`audio_codec`].
+    pub subscriber_audio_codec: Option<String>,
 }
 
 impl Default for MoqPeerConfig {
@@ -327,6 +365,8 @@ impl Default for MoqPeerConfig {
             video_width: 640,
             video_height: 480,
             video_codec: None,
+            audio_codec: None,
+            subscriber_audio_codec: None,
         }
     }
 }
Original file line number	Diff line number	Diff line change
`@@ -63,6 +63,7 @@ pub struct RawVideoFormat {`
`63`	`63`	`#[non_exhaustive]`
`64`	`64`	`pub enum AudioCodec {`
`65`	`65`	`Opus,`
	`66`	`+ Aac,`
`66`	`67`	`}`
`67`	`68`
`68`	`69`	`/// Supported encoded video codecs.`