feat(qwen3-next): add pipeline-parallel stage support

src/distributed/pipeline/stage_executor/family_registry_tests.rs

@@ -64,6 +64,19 @@ fn new_issue_345_families_are_registered() {
     }
 }
 
+#[test]
+fn qwen3_next_family_is_registered() {
+    // Floor test for the Qwen3-Next pipeline-parallel stage family. Removing
+    // it (or its stable name) without bumping the pipeline capability protocol
+    // version would break running multi-host qwen3-next deployments.
+    let families: HashSet<StageFamily> = supported_families().iter().copied().collect();
+    assert!(
+        families.contains(&StageFamily::Qwen3Next),
+        "StageFamily::Qwen3Next must appear in supported_families()",
+    );
+    assert_eq!(StageFamily::Qwen3Next.name(), "qwen3_next");
+}
+
 #[test]
 fn all_stage_family_variants_have_stable_names() {
     // If a new variant is added, this match must be updated to give it a

src/distributed/pipeline/stage_executor/mod.rs

@@ -41,6 +41,7 @@ mod mixtral;
 mod nemotron_h;
 mod qwen3;
 mod qwen35;
+mod qwen3_next;
 
 #[cfg(test)]
 #[path = "family_registry_tests.rs"]
@@ -69,6 +70,7 @@ use mistral::MistralStageExecutor;
 use mixtral::MixtralStageExecutor;
 use nemotron_h::NemotronHStageExecutor;
 use qwen3::Qwen3StageExecutor;
+use qwen3_next::Qwen3NextStageExecutor;
 use qwen35::Qwen35StageExecutor;
 
 /// Input payload for a single stage-local forward.
@@ -238,6 +240,7 @@ pub enum StageFamily {
     Glm4MoeLite,
     GlmMoeDsa,
     Qwen3,
+    Qwen3Next,
     Qwen35,
     Qwen35Vlm,
     Qwen35Moe,
@@ -267,6 +270,7 @@ impl StageFamily {
             Self::Glm4MoeLite => "glm4_moe_lite",
             Self::GlmMoeDsa => "glm_moe_dsa",
             Self::Qwen3 => "qwen3",
+            Self::Qwen3Next => "qwen3_next",
             Self::Qwen35 => "qwen3_5",
             Self::Qwen35Vlm => "qwen3_5_vlm",
             Self::Qwen35Moe => "qwen3_5_moe",
@@ -318,6 +322,7 @@ pub fn supported_families() -> &'static [StageFamily] {
         StageFamily::Qwen35Moe,
         StageFamily::Qwen35MoeVlm,
         StageFamily::Qwen35Vlm,
+        StageFamily::Qwen3Next,
     ];
     FAMILIES
 }
@@ -348,6 +353,7 @@ fn resolve_stage_family(model_dir: &Path) -> Result<StageFamily> {
         ModelType::Glm4MoeLite => StageFamily::Glm4MoeLite,
         ModelType::GlmMoeDsa => StageFamily::GlmMoeDsa,
         ModelType::Qwen3 => StageFamily::Qwen3,
+        ModelType::Qwen3Next => StageFamily::Qwen3Next,
         ModelType::Qwen35 => StageFamily::Qwen35,
         ModelType::Qwen35VLM => StageFamily::Qwen35Vlm,
         ModelType::Qwen35Moe => StageFamily::Qwen35Moe,
@@ -483,6 +489,14 @@ fn load_family_backend(
                 stage_index,
             )?))
         }
+        StageFamily::Qwen3Next => {
+            ensure_no_adapter(adapter_path, family)?;
+            Ok(Box::new(Qwen3NextStageExecutor::load(
+                model_dir,
+                filter,
+                stage_index,
+            )?))
+        }
         StageFamily::Qwen35
         | StageFamily::Qwen35Vlm
         | StageFamily::Qwen35Moe

src/distributed/pipeline/stage_executor/qwen3_next.rs

@@ -0,0 +1,102 @@
+// Copyright 2025-2026 Lablup Inc. and Jeongkyu Shin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Pipeline-parallel stage executor for Qwen3-Next.
+//!
+//! Qwen3-Next is a hybrid model that interleaves linear (GatedDeltaNet,
+//! `GatedDeltaCache`) and full-attention (`KVCache`) layers. This executor is
+//! the sibling of the Qwen 3.5 stage executor (`qwen35.rs`): it wraps a
+//! [`Qwen3NextStageModel`] holding only the stage's layer range and keeps the
+//! stage-local heterogeneous cache vector in a
+//! [`PointerOwnedCacheStore`], keyed off the identity of the external
+//! `KVCache` handle slice the runtime hands in.
+//!
+//! The external handles carry only the sequence offset back to admission and
+//! scheduler code; the real conv-state / SSM-state / KV tensors live in the
+//! internal [`Qwen3NextCache`] vector, one entry per LOCAL layer, whose
+//! `Linear`/`Attention` variant is chosen from the GLOBAL layer index so the
+//! cache type and mask match the single-process path exactly.
+//!
+//! Like the Qwen 3.5 executor, the speculative-decoding hooks are not
+//! propagated across the stage boundary: mlxcel's pipeline-parallel runner
+//! does not support speculative decoding for any family today.
+
+use std::path::Path;
+
+use anyhow::Result;
+use mlxcel_core::layers::KVCache;
+use mlxcel_core::{MlxArray, copy};
+
+use crate::models::qwen3_next::{Qwen3NextCache, Qwen3NextStageModel};
+
+use super::common::PointerOwnedCacheStore;
+use super::{FamilyStageExecutor, LayerFilter, StageExecutionInput, StageExecutionOutput};
+
+pub struct Qwen3NextStageExecutor {
+    model: Qwen3NextStageModel,
+    cache_store: PointerOwnedCacheStore<Qwen3NextCache>,
+}
+
+impl Qwen3NextStageExecutor {
+    pub fn load(model_dir: &Path, filter: &LayerFilter, stage_index: usize) -> Result<Self> {
+        Ok(Self {
+            model: Qwen3NextStageModel::load(model_dir, filter, stage_index)
+                .map_err(anyhow::Error::msg)?,
+            cache_store: PointerOwnedCacheStore::default(),
+        })
+    }
+}
+
+impl FamilyStageExecutor for Qwen3NextStageExecutor {
+    fn make_caches(&self) -> Vec<KVCache> {
+        (0..self.model.num_layers())
+            .map(|_| KVCache::new())
+            .collect()
+    }
+
+    fn release_caches(&self, caches: &[KVCache]) {
+        self.cache_store.release_caches(caches);
+    }
+
+    fn execute(
+        &self,
+        input: StageExecutionInput<'_>,
+        caches: &mut [KVCache],
+        _mask: Option<&MlxArray>,
+    ) -> Result<StageExecutionOutput> {
+        let mut internal_caches = self.cache_store.caches_for_sequence(
+            caches,
+            || self.model.make_caches(),
+            Qwen3NextCache::offset,
+            "qwen3-next sequence cache entry must exist",
+        );
+
+        let output = match input {
+            StageExecutionInput::TokenIds(input_ids) => self
+                .model
+                .execute_from_token_ids(input_ids, &mut internal_caches),
+            StageExecutionInput::HiddenStates(hidden_states) => self
+                .model
+                .execute_from_hidden_states(copy(hidden_states), &mut internal_caches),
+        }
+        .map_err(anyhow::Error::msg)?;
+
+        PointerOwnedCacheStore::sync_external_offsets(
+            caches,
+            &internal_caches,
+            Qwen3NextCache::offset,
+        );
+        Ok(output)
+    }
+}