fix(neuron): cover TpForwardLogitsWithImages in drain_poisoned match

The CUDA type-check caught a non-exhaustive match: drain_poisoned()
must reply an error to every Job variant's reply channel, including the
new cuda-gated TpForwardLogitsWithImages. The non-cuda build couldn't
see it — the variant is #[cfg(feature = "cuda")], so the match is
exhaustive without it on CPU.

Refs TP-vision plan Stage 2.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/neuron/src/harness/arch/qwen3_5/mod.rs

@@ -236,7 +236,11 @@ fn default_partial_rotary_factor() -> f32 {
 /// `slice_assign` per run. For typical Qwen3.6 requests this is one
 /// or two runs per image; `slice_assign` does one tensor copy per
 /// run, which is cheap relative to the decoder forward pass.
-fn splice_runs(h: &Tensor, img: &Tensor, positions: &[u32]) -> candle_core::Result<Tensor> {
+pub(crate) fn splice_runs(
+    h: &Tensor,
+    img: &Tensor,
+    positions: &[u32],
+) -> candle_core::Result<Tensor> {
     debug_assert!(
         !positions.is_empty(),
         "splice_runs precondition: non-empty positions"

crates/neuron/src/harness/candle.rs

@@ -106,18 +106,18 @@ impl LoadedHandle {
         }
     }
 
-    /// Modalities the loaded model supports. Stage B7. TP models are
-    /// always text-only today — TP-vision is tracked under issue #12.
+    /// Modalities the loaded model supports. Stage B7 (single-GPU) +
+    /// TP-vision (#12) — both single-GPU and TP loads advertise
+    /// `"vision"` when a replicated vision tower materialised.
     pub fn capabilities(&self) -> Vec<String> {
         let mut caps = vec!["text".to_string()];
-        match self {
-            LoadedHandle::Single(m) => {
-                if m.has_vision {
-                    caps.push("vision".to_string());
-                }
-            }
+        let has_vision = match self {
+            LoadedHandle::Single(m) => m.has_vision,
             #[cfg(feature = "cuda")]
-            LoadedHandle::Tp(_) => {}
+            LoadedHandle::Tp(m) => m.has_vision,
+        };
+        if has_vision {
+            caps.push("vision".to_string());
         }
         caps
     }
@@ -281,6 +281,16 @@ pub struct TpLoadedModel {
     pub tool_call_tokens: Option<ToolCallTokenPair>,
     /// Same shape as [`LoadedModel::chat_template`].
     pub chat_template: Option<String>,
+    /// Vision capability flag (TP-vision). `true` iff every rank
+    /// materialised a replicated vision tower. Mirrors
+    /// [`LoadedModel::has_vision`]; drives capability advertising and
+    /// the TP vision dispatch.
+    pub has_vision: bool,
+    /// `<|image_pad|>` token id — same as [`LoadedModel::image_token_id`].
+    pub image_token_id: Option<u32>,
+    /// LM-side tokens per image at the fixed 448×448 resolution — same
+    /// as [`LoadedModel::lm_tokens_per_image`].
+    pub lm_tokens_per_image: Option<usize>,
 }
 
 #[cfg(feature = "cuda")]
@@ -2675,6 +2685,20 @@ impl CandleHarness {
             );
         }
 
+        // Vision metadata from the same config.json the shards loaded
+        // from. The TP model builder (Stage 1) materialises a replicated
+        // vision tower on every rank when `vision_config` is present, so
+        // `has_vision` here is consistent with what each rank loaded.
+        let vision_meta = VisionMeta::from_config_path(&config_path);
+        if vision_meta.has_vision {
+            tracing::info!(
+                model = %spec.model_id,
+                image_token_id = ?vision_meta.image_token_id,
+                lm_tokens_per_image = ?vision_meta.lm_tokens_per_image,
+                "TP load: vision tower present, advertising vision capability"
+            );
+        }
+
         let tp_loaded = StdArc::new(TpLoadedModel {
             model_id: spec.model_id.clone(),
             tokenizer,
@@ -2690,6 +2714,9 @@ impl CandleHarness {
             reasoning_tokens,
             tool_call_tokens,
             chat_template,
+            has_vision: vision_meta.has_vision,
+            image_token_id: vision_meta.image_token_id,
+            lm_tokens_per_image: vision_meta.lm_tokens_per_image,
         });
 
         let mut models = self.models.write().await;
@@ -2739,15 +2766,15 @@ impl CandleHarness {
             return Err(poisoned_error(&model_id));
         }
 
-        // Stage 0 (TP-vision): the TP path has no vision tower yet, so
-        // an image-bearing request can't be honoured. Reject it cleanly
-        // with `vision_unsupported` instead of silently dropping the
-        // image and answering from text alone (the issue-#3 confident-
-        // hallucination pattern). Made conditional on the TP model's
-        // `has_vision` once Stage 3 wires real TP-vision.
-        if request_has_images(&request) {
+        // Reject image-bearing requests against a TP model with no
+        // vision tower, cleanly (`vision_unsupported`) rather than
+        // silently dropping the image. Vision-capable TP loads fall
+        // through to the image-aware prefill in chat_completion_tp_inner.
+        if request_has_images(&request) && !tp.has_vision {
             let _g = span.enter();
-            tracing::warn!("TP chat_completion: rejecting image request, TP vision unsupported");
+            tracing::warn!(
+                "TP chat_completion: rejecting image request, model has no vision tower"
+            );
             return Err(InferenceError::VisionUnsupported { model_id });
         }
 
@@ -2828,14 +2855,12 @@ impl CandleHarness {
             return Err(poisoned_error(&request.model));
         }
 
-        // Stage 0 (TP-vision): reject image requests on the TP streaming
-        // path before opening the SSE stream — the TP path has no vision
-        // tower yet, so honouring the image is impossible and silently
-        // dropping it would hallucinate. Returns a clean 400; made
-        // conditional on `has_vision` in Stage 3.
-        if request_has_images(&request) {
+        // Reject image requests against a non-vision TP model before
+        // opening the SSE stream. Vision-capable TP loads fall through
+        // to the image-aware prefill in the orchestration task below.
+        if request_has_images(&request) && !tp.has_vision {
             tracing::warn!(
-                "TP chat_completion (stream): rejecting image request, TP vision unsupported"
+                "TP chat_completion (stream): rejecting image request, model has no vision tower"
             );
             return Err(InferenceError::VisionUnsupported {
                 model_id: request.model.clone(),
@@ -2847,7 +2872,44 @@ impl CandleHarness {
             .tokenizer
             .encode(prompt.as_str(), true)
             .map_err(|e| InferenceError::Other(anyhow::anyhow!("tokenize: {e}")))?;
-        let prompt_tokens: Vec<u32> = encoding.get_ids().to_vec();
+        let mut prompt_tokens: Vec<u32> = encoding.get_ids().to_vec();
+
+        // TP-vision (streaming): same detection + pad expansion as the
+        // non-streaming path. The resulting `vision_route` moves into
+        // the orchestration task, which runs a single-shot image prefill
+        // when present. Returning early here keeps a rejected request
+        // from opening the SSE stream.
+        let vision_route: Option<(Vec<String>, u32)> = if request_has_images(&request) {
+            if !tp.has_vision {
+                return Err(InferenceError::VisionUnsupported {
+                    model_id: request.model.clone(),
+                });
+            }
+            let image_token_id =
+                tp.image_token_id
+                    .ok_or_else(|| InferenceError::VisionUnsupported {
+                        model_id: request.model.clone(),
+                    })?;
+            let patches_per_image =
+                tp.lm_tokens_per_image
+                    .ok_or_else(|| InferenceError::VisionUnsupported {
+                        model_id: request.model.clone(),
+                    })?;
+            let data_uris = extract_image_data_uris(&request);
+            if data_uris.is_empty() {
+                return Err(InferenceError::Other(anyhow::anyhow!(
+                    "request has image content but extractor produced zero data URIs"
+                )));
+            }
+            let per_image_counts: Vec<usize> = vec![patches_per_image; data_uris.len()];
+            prompt_tokens =
+                expand_image_pad_tokens(&prompt_tokens, image_token_id, &per_image_counts)
+                    .map_err(InferenceError::Other)?;
+            Some((data_uris, image_token_id))
+        } else {
+            None
+        };
+
         let prompt_len = prompt_tokens.len();
 
         let temperature = request.temperature.unwrap_or(0.7);
@@ -2961,14 +3023,27 @@ impl CandleHarness {
                     // chunk fans out to every rank with a growing
                     // offset; only the final chunk's logits are kept
                     // for the first sample.
-                    let logits_vec = match chunked_prefill_tp(
-                        &mut pool,
-                        &model_id,
-                        leader_handle,
-                        &prompt_tokens,
-                    )
-                    .await
-                    {
+                    // Vision requests do a single-shot image prefill;
+                    // text requests chunk it. `vision_route` was moved
+                    // into this task from the synchronous setup above.
+                    let prefill_result = match &vision_route {
+                        Some((data_uris, image_token_id)) => {
+                            pool.generate_step_with_images(
+                                &model_id,
+                                leader_handle,
+                                prompt_tokens.clone(),
+                                0,
+                                *image_token_id,
+                                data_uris.clone(),
+                            )
+                            .await
+                        }
+                        None => {
+                            chunked_prefill_tp(&mut pool, &model_id, leader_handle, &prompt_tokens)
+                                .await
+                        }
+                    };
+                    let logits_vec = match prefill_result {
                         Ok(l) => l,
                         Err(e) => {
                             failure = Some(format!("prefill: {e:#}"));
@@ -3311,7 +3386,43 @@ async fn chat_completion_tp_inner(
         .tokenizer
         .encode(prompt.as_str(), true)
         .map_err(|e| InferenceError::Other(anyhow::anyhow!("tokenize: {e}")))?;
-    let prompt_tokens: Vec<u32> = encoding.get_ids().to_vec();
+    let mut prompt_tokens: Vec<u32> = encoding.get_ids().to_vec();
+
+    // TP-vision: when the request carries images (and the model has a
+    // replicated tower — enforced by the caller's guard), expand each
+    // `<|image_pad|>` sentinel to the per-image patch count and carry
+    // the source data URIs through to the single-shot image prefill.
+    // Mirrors the single-GPU `chat_completion` vision_route block.
+    let vision_route: Option<(Vec<String>, u32)> = if request_has_images(&request) {
+        if !tp.has_vision {
+            return Err(InferenceError::VisionUnsupported {
+                model_id: request.model.clone(),
+            });
+        }
+        let image_token_id =
+            tp.image_token_id
+                .ok_or_else(|| InferenceError::VisionUnsupported {
+                    model_id: request.model.clone(),
+                })?;
+        let patches_per_image =
+            tp.lm_tokens_per_image
+                .ok_or_else(|| InferenceError::VisionUnsupported {
+                    model_id: request.model.clone(),
+                })?;
+        let data_uris = extract_image_data_uris(&request);
+        if data_uris.is_empty() {
+            return Err(InferenceError::Other(anyhow::anyhow!(
+                "request has image content but extractor produced zero data URIs"
+            )));
+        }
+        let per_image_counts: Vec<usize> = vec![patches_per_image; data_uris.len()];
+        prompt_tokens = expand_image_pad_tokens(&prompt_tokens, image_token_id, &per_image_counts)
+            .map_err(InferenceError::Other)?;
+        Some((data_uris, image_token_id))
+    } else {
+        None
+    };
+
     let prompt_len = prompt_tokens.len();
 
     let temperature = request.temperature.unwrap_or(0.7);
@@ -3381,9 +3492,24 @@ async fn chat_completion_tp_inner(
     // spread across multiple `generate_step` calls with monotonically
     // growing offsets.
     let prefill_start = std::time::Instant::now();
-    let logits_vec = chunked_prefill_tp(&mut pool, &model_id, leader_handle, &prompt_tokens)
-        .await
-        .map_err(InferenceError::Other)?;
+    // Vision requests do a single-shot image prefill (every rank encodes
+    // + splices its replicated tower); text requests chunk the prefill.
+    let logits_vec = match &vision_route {
+        Some((data_uris, image_token_id)) => pool
+            .generate_step_with_images(
+                &model_id,
+                leader_handle,
+                prompt_tokens.clone(),
+                0,
+                *image_token_id,
+                data_uris.clone(),
+            )
+            .await
+            .map_err(InferenceError::Other)?,
+        None => chunked_prefill_tp(&mut pool, &model_id, leader_handle, &prompt_tokens)
+            .await
+            .map_err(InferenceError::Other)?,
+    };
     let (post_prefill_vram_free_mb, _) = tp.query_vram().await;
     tracing::info!(
         model = %model_id,
@@ -3841,6 +3967,37 @@ fn extract_images_from_request(
     Ok(out)
 }
 
+/// Collect the raw `image_url.url` strings (data URIs) from a chat
+/// request, in prompt order. The TP vision path (Stage C / TP-vision)
+/// ships these verbatim to every rank, which each preprocess + encode
+/// identically — so unlike `extract_images_from_request` (which
+/// preprocesses on the leader for the single-GPU worker job) this
+/// keeps the source form for replicated per-rank encoding.
+///
+/// Cuda-gated: the only callers are the TP entry points, which compile
+/// only under the `cuda` feature.
+#[cfg(feature = "cuda")]
+fn extract_image_data_uris(request: &ChatCompletionRequest) -> Vec<String> {
+    let mut out = Vec::new();
+    for msg in &request.messages {
+        if let MessageContent::Parts(parts) = &msg.content {
+            for part in parts {
+                if part.get("type").and_then(|v| v.as_str()) != Some("image_url") {
+                    continue;
+                }
+                if let Some(url) = part
+                    .get("image_url")
+                    .and_then(|v| v.get("url"))
+                    .and_then(|v| v.as_str())
+                {
+                    out.push(url.to_string());
+                }
+            }
+        }
+    }
+    out
+}
+
 /// Expand each occurrence of `image_token_id` in `input_ids` into
 /// `patches_per_image[i]` copies (one expansion per image, in order).
 /// Stage B4 helper.

crates/neuron/src/harness/device_worker/dispatch.rs

@@ -262,6 +262,25 @@ pub(crate) fn run(device_index: u32, rx: Receiver<Job>, poisoned: Arc<AtomicBool
                 let result = tp_forward_logits(&mut state, handle, &tokens, offset);
                 let _ = reply.send(result);
             }
+            #[cfg(feature = "cuda")]
+            Job::TpForwardLogitsWithImages {
+                handle,
+                tokens,
+                offset,
+                image_token_id,
+                image_data_uris,
+                reply,
+            } => {
+                let result = tp_forward_logits_with_images(
+                    &mut state,
+                    handle,
+                    &tokens,
+                    offset,
+                    image_token_id,
+                    &image_data_uris,
+                );
+                let _ = reply.send(result);
+            }
             // Handled by the matches!() check above; reaching here
             // means a Shutdown slipped past which is a bug.
             Job::Shutdown => unreachable!("Shutdown should break above"),
@@ -734,6 +753,61 @@ fn tp_forward_logits(
     Ok(values)
 }
 
+/// Image-bearing leader forward (rank 0). Preprocesses each source
+/// `image_data_uris` entry through the same deterministic
+/// `preprocess_data_uri` every rank runs, uploads to the leader's
+/// device, encodes + splices + forwards via
+/// `TpLeaderModel::forward_with_images`, and copies the `[vocab]`
+/// logits to CPU. Mirrors the single-GPU `forward_logits_with_images`
+/// but on the TP leader's replicated tower.
+#[cfg(feature = "cuda")]
+fn tp_forward_logits_with_images(
+    state: &mut DeviceWorkerState,
+    handle: TpHandle,
+    tokens: &[u32],
+    offset: usize,
+    image_token_id: u32,
+    image_data_uris: &[String],
+) -> anyhow::Result<Vec<f32>> {
+    use crate::harness::preprocess::{PreprocessProfile, preprocess_data_uri};
+    use candle_core::{DType, Tensor};
+
+    if image_data_uris.is_empty() {
+        anyhow::bail!("TpForwardLogitsWithImages dispatched with zero images");
+    }
+
+    // Preprocess every image into a device-resident (C, H, W) tensor.
+    // Same fixed-resolution profile + decode path the subprocess workers
+    // run, so the encoded embeddings match across ranks bit-for-bit.
+    let profile = PreprocessProfile::qwen3_6();
+    let (h, w) = (
+        profile.target_height as usize,
+        profile.target_width as usize,
+    );
+    let mut pixels: Vec<Tensor> = Vec::with_capacity(image_data_uris.len());
+    for (idx, uri) in image_data_uris.iter().enumerate() {
+        let px = preprocess_data_uri(uri, &profile)
+            .with_context(|| format!("preprocess image[{idx}] (TP leader)"))?;
+        let t = Tensor::from_vec(px, (3, h, w), &state.device)?;
+        pixels.push(t);
+    }
+
+    let input = Tensor::new(tokens, &state.device)?.unsqueeze(0)?;
+
+    let model = state.tp_models.get_mut(&handle).ok_or_else(|| {
+        anyhow::anyhow!(
+            "TpForwardLogitsWithImages: no model for handle {}",
+            handle.0
+        )
+    })?;
+
+    let logits = model.forward_with_images(&input, offset, &pixels, image_token_id)?;
+    let logits = logits.squeeze(0)?.squeeze(0)?;
+    let logits = logits.to_dtype(DType::F32)?.flatten_all()?;
+    let values = logits.to_vec1::<f32>()?;
+    Ok(values)
+}
+
 /// Forward step + copy the `[vocab]` logits to a CPU `Vec<f32>` ready
 /// for sampling on the async caller. The model's `device()` (CUDA or
 /// CPU) determines where the kernel runs; this fn doesn't care.
@@ -941,6 +1015,10 @@ fn drain_poisoned(job: Job, device_index: u32) {
         Job::TpForwardLogits { reply, .. } => {
             let _ = reply.send(Err(err()));
         }
+        #[cfg(feature = "cuda")]
+        Job::TpForwardLogitsWithImages { reply, .. } => {
+            let _ = reply.send(Err(err()));
+        }
         Job::Shutdown => {
             // Filtered by the matches!() guard in run(); reaching
             // here would be a logic error.

crates/neuron/src/harness/device_worker/jobs.rs

@@ -231,6 +231,23 @@ pub enum Job {
         offset: usize,
         reply: oneshot::Sender<Result<Vec<f32>>>,
     },
+    /// Image-bearing leader (rank 0) forward for the single-shot vision
+    /// prefill. The handler preprocesses each `image_data_uris` entry
+    /// (the same deterministic path every rank runs), encodes through
+    /// the leader's replicated tower, splices at `image_token_id`, and
+    /// returns CPU-side `[vocab]` logits. Image tensors never escape the
+    /// worker thread. Caller fans out `GenerateStepWithImages` to the
+    /// subprocess ranks and drains them; only the leader forward moves
+    /// here.
+    #[cfg(feature = "cuda")]
+    TpForwardLogitsWithImages {
+        handle: TpHandle,
+        tokens: Vec<u32>,
+        offset: usize,
+        image_token_id: u32,
+        image_data_uris: Vec<String>,
+        reply: oneshot::Sender<Result<Vec<f32>>>,
+    },
     /// Tell the worker to break its dispatch loop and exit. Any jobs
     /// queued after this in the channel reply `Err` to their oneshot
     /// senders (the senders are dropped on the worker's exit, which

crates/neuron/src/harness/device_worker/mod.rs

@@ -572,6 +572,47 @@ impl DeviceWorkerHandle {
         }
     }
 
+    /// Image-bearing TP leader forward (single-shot vision prefill).
+    /// Routes `Job::TpForwardLogitsWithImages` onto the worker thread;
+    /// the handler preprocesses + encodes + splices + forwards and
+    /// returns CPU-side `[vocab]` logits. The `WorkerPool` fans the
+    /// matching `GenerateStepWithImages` out to subprocess ranks so the
+    /// row-parallel collectives complete.
+    #[cfg(feature = "cuda")]
+    pub async fn tp_forward_logits_with_images(
+        &self,
+        handle: TpHandle,
+        tokens: Vec<u32>,
+        offset: usize,
+        image_token_id: u32,
+        image_data_uris: Vec<String>,
+    ) -> Result<Vec<f32>, WorkerError> {
+        if self.poisoned.load(Ordering::Acquire) {
+            return Err(WorkerError::Poisoned {
+                device_index: self.device_index,
+            });
+        }
+        let (reply_tx, reply_rx) = oneshot::channel();
+        self.tx
+            .send(Job::TpForwardLogitsWithImages {
+                handle,
+                tokens,
+                offset,
+                image_token_id,
+                image_data_uris,
+                reply: reply_tx,
+            })
+            .map_err(|_| WorkerError::Gone {
+                device_index: self.device_index,
+            })?;
+        match reply_rx.await {
+            Ok(result) => result.map_err(WorkerError::from),
+            Err(_) => Err(WorkerError::Gone {
+                device_index: self.device_index,
+            }),
+        }
+    }
+
     /// Send `Job::Shutdown` and join the thread. Idempotent — calling
     /// twice is a no-op the second time.
     pub fn shutdown(&self) -> anyhow::Result<()> {

crates/neuron/src/harness/tp/mod.rs

@@ -62,6 +62,25 @@ impl TpLeaderModel {
         }
     }
 
+    /// Image-bearing forward on rank 0. Only the vision-capable
+    /// `qwen3_5` arch supports it; the dense `qwen3` arch has no tower.
+    pub fn forward_with_images(
+        &mut self,
+        input: &candle_core::Tensor,
+        offset: usize,
+        image_pixels: &[candle_core::Tensor],
+        image_token_id: u32,
+    ) -> candle_core::Result<candle_core::Tensor> {
+        match self {
+            TpLeaderModel::Qwen3_5(m) => {
+                m.forward_with_images(input, offset, image_pixels, image_token_id)
+            }
+            TpLeaderModel::Qwen3(_) => {
+                candle_core::bail!("forward_with_images: qwen3 (dense) has no vision tower")
+            }
+        }
+    }
+
     pub fn clear_kv_cache(&mut self) {
         match self {
             TpLeaderModel::Qwen3(m) => m.clear_kv_cache(),
@@ -687,6 +706,129 @@ impl WorkerPool {
         }
     }
 
+    /// Image-bearing variant of [`Self::generate_step`] for the
+    /// single-shot vision prefill. Identical fan-out / leader-forward /
+    /// drain shape, but every rank runs the encode + splice path:
+    ///
+    /// - subprocess workers get `GenerateStepWithImages` (carrying the
+    ///   source `image_data_uris`); each preprocesses + encodes through
+    ///   its replicated tower and splices locally;
+    /// - the leader runs the same encode + splice + forward on its
+    ///   device worker thread via `tp_forward_logits_with_images`.
+    ///
+    /// The row-parallel `AllReduce`s synchronise the ranks exactly as in
+    /// the text path. Because the tower is replicated and the preprocess
+    /// is deterministic, every rank's spliced hidden state matches — no
+    /// embedding broadcast. Only used for prefill; decode reuses
+    /// `generate_step`.
+    #[cfg(feature = "cuda")]
+    pub async fn generate_step_with_images(
+        &mut self,
+        model_id: &str,
+        leader_handle: super::device_worker::TpHandle,
+        tokens: Vec<u32>,
+        offset: usize,
+        image_token_id: u32,
+        image_data_uris: Vec<String>,
+    ) -> Result<Vec<f32>> {
+        let step_start = std::time::Instant::now();
+        let tokens_len = tokens.len();
+        tracing::debug!(
+            model = %model_id,
+            tokens = tokens_len,
+            offset,
+            images = image_data_uris.len(),
+            "WorkerPool::generate_step_with_images: fan-out"
+        );
+
+        // 1. Fan-out the image-bearing prefill to subprocess workers.
+        for w in &mut self.workers {
+            w.send_only(&WorkerRequest::GenerateStepWithImages {
+                model_id: model_id.to_string(),
+                tokens: tokens.clone(),
+                offset,
+                image_token_id,
+                image_data_uris: image_data_uris.clone(),
+            })
+            .await?;
+        }
+
+        // 2. Leader's image forward on its device worker thread. The
+        //    AllReduce CustomOps block until every worker issues the
+        //    matching collective; CPU-side logits keep the device tensor
+        //    from escaping the worker thread.
+        let leader_start = std::time::Instant::now();
+        let leader_result = self
+            .leader_worker
+            .tp_forward_logits_with_images(
+                leader_handle,
+                tokens,
+                offset,
+                image_token_id,
+                image_data_uris,
+            )
+            .await;
+        let leader_ok = leader_result.is_ok();
+        let leader_ms = leader_start.elapsed().as_millis();
+        if !leader_ok {
+            let detail = leader_result
+                .as_ref()
+                .err()
+                .map(|e| format!("{e:#}"))
+                .unwrap_or_default();
+            tracing::warn!(
+                model = %model_id,
+                tokens = tokens_len,
+                offset,
+                leader_ms,
+                error = %detail,
+                "WorkerPool::generate_step_with_images: leader forward failed"
+            );
+        }
+
+        // 3. ALWAYS drain worker responses, regardless of the leader's
+        //    outcome, so stale GenerateStepOk replies don't poison the
+        //    next request's recv (same invariant as generate_step).
+        let worker_errors = drain_workers(&mut self.workers, |r| match r {
+            WorkerResponse::GenerateStepOk => Ok(()),
+            WorkerResponse::Error { kind, message } => Err(format!("[{kind}]: {message}")),
+            other => Err(format!("expected GenerateStepOk, got {other:?}")),
+        })
+        .await;
+        tracing::debug!(
+            model = %model_id,
+            leader_ms,
+            leader_ok,
+            errors = worker_errors.len(),
+            total_ms = step_start.elapsed().as_millis(),
+            "WorkerPool::generate_step_with_images: workers drained"
+        );
+
+        match leader_result {
+            Ok(values) => {
+                if worker_errors.is_empty() {
+                    Ok(values)
+                } else {
+                    anyhow::bail!(
+                        "GenerateStepWithImages: leader succeeded but workers failed: {}",
+                        worker_errors.join("; ")
+                    )
+                }
+            }
+            Err(e) => {
+                if worker_errors.is_empty() {
+                    Err(anyhow::Error::new(e)
+                        .context("GenerateStepWithImages: leader forward failed"))
+                } else {
+                    Err(anyhow::Error::new(e).context(format!(
+                        "GenerateStepWithImages: leader forward failed and workers also failed: {}",
+                        worker_errors.join("; ")
+                    )))
+                }
+            }
+        }
+    }
+
     /// Reset the KV cache for `model_id` on every rank. Called at the
     /// start of every inference so a fresh request doesn't attend over
     /// the previous one's tokens.

crates/neuron/src/harness/tp/rpc.rs

@@ -88,6 +88,29 @@ pub enum WorkerRequest {
         offset: usize,
     },
 
+    /// Like `GenerateStep` but the prefill carries image content. Every
+    /// rank preprocesses the same `image_data_uris` through its
+    /// *replicated* vision tower, splices the resulting patch embeddings
+    /// at `image_token_id` positions, and runs the forward — the
+    /// row-parallel `AllReduce`s still synchronise every rank. Because
+    /// the tower is replicated and `preprocess_data_uri` is
+    /// deterministic, the spliced hidden state is identical on every
+    /// rank, so no embedding broadcast is needed. Sent only for the
+    /// (single-shot) image-bearing prefill; decode steps use plain
+    /// `GenerateStep`. Worker replies with the same `GenerateStepOk`.
+    GenerateStepWithImages {
+        model_id: String,
+        tokens: Vec<u32>,
+        offset: usize,
+        /// `<|image_pad|>` sentinel id (248056 for Qwen3.6); splice
+        /// target in the expanded token stream.
+        image_token_id: u32,
+        /// Source image data URIs (`data:image/...;base64,...`), one per
+        /// image in prompt order. Each rank decodes + preprocesses these
+        /// identically; tens of KB each, so cheap over the stdin pipe.
+        image_data_uris: Vec<String>,
+    },
+
     /// Reset the KV cache for this model on this rank. Sent at the
     /// start of every inference so a fresh request doesn't accidentally
     /// attend over the previous one's tokens.
@@ -191,6 +214,32 @@ mod tests {
         assert_eq!(wire, r#"{"op":"init","comm_id":"deadbeef"}"#);
     }
 
+    #[test]
+    fn request_generate_step_with_images_round_trip() {
+        let req = WorkerRequest::GenerateStepWithImages {
+            model_id: "Qwen/Qwen3.6-27B".into(),
+            tokens: vec![1, 2, 248056, 3],
+            offset: 0,
+            image_token_id: 248056,
+            image_data_uris: vec!["data:image/png;base64,AAA=".into()],
+        };
+        let wire = serde_json::to_string(&req).unwrap();
+        assert!(wire.contains(r#""op":"generate_step_with_images""#));
+        match roundtrip(&req) {
+            WorkerRequest::GenerateStepWithImages {
+                tokens,
+                image_token_id,
+                image_data_uris,
+                ..
+            } => {
+                assert_eq!(tokens, vec![1, 2, 248056, 3]);
+                assert_eq!(image_token_id, 248056);
+                assert_eq!(image_data_uris.len(), 1);
+            }
+            other => panic!("expected GenerateStepWithImages, got {other:?}"),
+        }
+    }
+
     #[test]
     fn request_shutdown_round_trip() {
         assert_eq!(

crates/neuron/src/harness/tp/tp_qwen3_5.rs

@@ -46,6 +46,8 @@ use super::tp_linear::{ColumnParallelLinear, RowParallelLinear};
 use crate::harness::arch::qwen3_5::linear_attn::repeat_interleave;
 use crate::harness::arch::qwen3_5::rmsnorm::{Qwen3_5RmsNorm, Qwen3_5RmsNormGated, l2norm};
 use crate::harness::arch::qwen3_5::rope::RotaryEmbedding;
+use crate::harness::arch::qwen3_5::splice_runs;
+use crate::harness::arch::qwen3_5::vision::VisionTower;
 pub use crate::harness::arch::qwen3_5::{Config, TextConfig};
 
 // ─── linear-attention (Gated DeltaNet) ──────────────────────────────
@@ -990,11 +992,103 @@ impl TpQwen3_5Model {
         }
         self.norm.forward(&h)
     }
+
+    /// Forward with image-embedding splice (TP, replicated tower).
+    ///
+    /// Mirrors the single-GPU `Qwen3_5Model::forward_inner` splice:
+    /// embed locally, replace the rows at `image_token_id` positions
+    /// with the image patch embeddings, then run the sharded decoder
+    /// stack. The TP invariant is that every rank holds an identical
+    /// hidden state (only the attention/MLP matmuls shard, with a
+    /// trailing `AllReduce`). That holds here because every rank
+    /// encodes the *same* pixels through its *replicated* vision tower
+    /// and so produces identical `image_embeds` — no broadcast needed.
+    pub fn forward_with_vision(
+        &mut self,
+        input: &Tensor,
+        offset: usize,
+        image_embeds: &Tensor,
+        image_token_id: u32,
+    ) -> candle_core::Result<Tensor> {
+        let (b, l) = input.dims2()?;
+        let mut h = self.embed_tokens.forward(input)?;
+
+        // Locate the image-token positions in the (pre-expanded) input
+        // ids and splice the patch rows in. Same CPU-side scan as the
+        // single-GPU path; the count must match the patch dimension or
+        // the prompt expansion is wrong.
+        let ids: Vec<u32> = input.flatten_all()?.to_vec1()?;
+        let mut positions: Vec<u32> = Vec::with_capacity(image_embeds.dim(0)?);
+        for (idx, id) in ids.iter().enumerate() {
+            if *id == image_token_id {
+                positions.push(idx as u32);
+            }
+        }
+        let n_img_tokens = image_embeds.dim(0)?;
+        if positions.len() != n_img_tokens {
+            candle_core::bail!(
+                "TP forward_with_vision: prompt has {} image-token positions but \
+                 image_embeds carries {} tokens — ensure the per-image patch-count \
+                 expansion has been applied",
+                positions.len(),
+                n_img_tokens,
+            );
+        }
+        if !positions.is_empty() {
+            let img = image_embeds.to_dtype(self.dtype)?;
+            h = splice_runs(&h, &img, &positions)?;
+        }
+
+        let causal = if l == 1 {
+            None
+        } else {
+            Some(self.causal_mask(b, l, offset)?)
+        };
+        for layer in &mut self.layers {
+            h = layer.forward(&h, causal.as_ref(), offset)?;
+        }
+        self.norm.forward(&h)
+    }
 }
 
 pub struct TpQwen3_5ForCausalLM {
     base: TpQwen3_5Model,
     lm_head: super::tp_linear::MaybeQuantLinear,
+    /// Replicated vision tower (TP-vision). Loaded on every rank from
+    /// the full, unsharded `model.visual.*` weights; `None` for
+    /// text-only checkpoints. Each rank encodes the same image
+    /// independently — no sharding, no broadcast — which keeps the
+    /// spliced input embeddings identical across ranks (the
+    /// replicated-hidden-state invariant the sharded layers rely on).
+    vision: Option<VisionTower>,
+    /// `<|image_pad|>` sentinel id (mirrors `Config::image_token_id`);
+    /// the splice target for `forward_with_vision`.
+    image_token_id: Option<u32>,
+}
+
+/// Load the replicated vision tower from the unsharded `model.visual.*`
+/// weights when the config carries a `vision_config` block. Shared by
+/// the cuda and non-cuda `load` variants. `vb.pp("model.visual")`
+/// resolves against the same full safetensors every rank mmaps; plain
+/// `.get()` on a `ShardedVarBuilder` returns the full (replicated)
+/// tensor, so this loads identically regardless of `world_size`.
+fn load_replicated_vision_tower(
+    config: &Config,
+    vb: &ShardedVarBuilder,
+) -> Result<Option<VisionTower>> {
+    match config.vision_config.clone() {
+        Some(vcfg) => {
+            tracing::info!(
+                depth = vcfg.depth,
+                hidden_size = vcfg.hidden_size,
+                "loading qwen3_5 vision tower (TP replicated)"
+            );
+            let tower = VisionTower::load(vcfg, vb.pp("model.visual"))
+                .context("load qwen3_5 vision tower (model.visual.*) [TP replicated]")?;
+            Ok(Some(tower))
+        }
+        None => Ok(None),
+    }
 }
 
 impl TpQwen3_5ForCausalLM {
@@ -1012,7 +1106,14 @@ impl TpQwen3_5ForCausalLM {
         let cfg = &config.text_config;
         let base = TpQwen3_5Model::load(cfg, vb, mmap, rank, world_size, comm, quant)?;
         let lm_head = build_lm_head(cfg, vb, &base, quant)?;
-        let model = Self { base, lm_head };
+        let vision = load_replicated_vision_tower(&config, vb)?;
+        let image_token_id = config.image_token_id;
+        let model = Self {
+            base,
+            lm_head,
+            vision,
+            image_token_id,
+        };
         log_construction_complete(cfg, rank, world_size, quant, model.device());
         Ok(model)
     }
@@ -1029,17 +1130,100 @@ impl TpQwen3_5ForCausalLM {
         let cfg = &config.text_config;
         let base = TpQwen3_5Model::load(cfg, vb, mmap, rank, world_size, quant)?;
         let lm_head = build_lm_head(cfg, vb, &base, quant)?;
-        let model = Self { base, lm_head };
+        let vision = load_replicated_vision_tower(&config, vb)?;
+        let image_token_id = config.image_token_id;
+        let model = Self {
+            base,
+            lm_head,
+            vision,
+            image_token_id,
+        };
         log_construction_complete(cfg, rank, world_size, quant, model.device());
         Ok(model)
     }
 
+    /// True when this TP load materialised a replicated vision tower.
+    /// Drives capability advertising and the Stage 3 vision dispatch.
+    pub fn has_vision(&self) -> bool {
+        self.vision.is_some()
+    }
+
+    /// `<|image_pad|>` sentinel id, when known.
+    pub fn image_token_id(&self) -> Option<u32> {
+        self.image_token_id
+    }
+
+    /// Encode one preprocessed `(C, H, W)` image into LM-side patch
+    /// embeddings `(N_lm, hidden)` via this rank's replicated tower.
+    /// Errors when loaded without a vision tower.
+    pub fn encode_image(&self, image: &Tensor) -> Result<Tensor> {
+        self.vision
+            .as_ref()
+            .ok_or_else(|| {
+                anyhow::anyhow!(
+                    "encode_image: this TP Qwen3.6 load has no vision tower \
+                     (config.json::vision_config absent or weights missing)"
+                )
+            })?
+            .forward(image)
+    }
+
     pub fn forward(&mut self, input: &Tensor, offset: usize) -> candle_core::Result<Tensor> {
         let (_, l) = input.dims2()?;
         let hidden = self.base.forward(input, offset)?;
         hidden.i((.., l - 1.., ..))?.apply(&self.lm_head)
     }
 
+    /// Forward with image-embedding splice (TP). Mirrors `forward` but
+    /// routes through `TpQwen3_5Model::forward_with_vision` so the
+    /// per-rank input embeddings get the image patches spliced in at
+    /// `image_token_id` positions before the sharded decoder stack.
+    pub fn forward_with_vision(
+        &mut self,
+        input: &Tensor,
+        offset: usize,
+        image_embeds: &Tensor,
+        image_token_id: u32,
+    ) -> candle_core::Result<Tensor> {
+        let (_, l) = input.dims2()?;
+        let hidden = self
+            .base
+            .forward_with_vision(input, offset, image_embeds, image_token_id)?;
+        hidden.i((.., l - 1.., ..))?.apply(&self.lm_head)
+    }
+
+    /// End-to-end image prefill on one rank: encode each preprocessed
+    /// `(C, H, W)` pixel tensor through this rank's replicated tower,
+    /// concatenate the per-image embeddings along the patch axis, and
+    /// forward with the splice. Shared by the leader (`TpLeaderModel`)
+    /// and the subprocess worker (`WorkerModel`) so every rank runs the
+    /// identical encode → splice → forward and keeps the replicated
+    /// hidden state in lockstep. Returns last-position logits
+    /// `(B, 1, vocab)`, same contract as `forward`.
+    pub fn forward_with_images(
+        &mut self,
+        input: &Tensor,
+        offset: usize,
+        image_pixels: &[Tensor],
+        image_token_id: u32,
+    ) -> candle_core::Result<Tensor> {
+        if image_pixels.is_empty() {
+            candle_core::bail!("forward_with_images: called with zero images");
+        }
+        // Encode each image (immutable borrows of the tower) before the
+        // mutable forward below; the borrows end as each owned embedding
+        // is pushed.
+        let mut per_image = Vec::with_capacity(image_pixels.len());
+        for (idx, img) in image_pixels.iter().enumerate() {
+            let embed = self
+                .encode_image(img)
+                .map_err(|e| candle_core::Error::Msg(format!("encode image[{idx}]: {e:#}")))?;
+            per_image.push(embed);
+        }
+        let image_embeds = Tensor::cat(&per_image.iter().collect::<Vec<_>>(), 0)?;
+        self.forward_with_vision(input, offset, &image_embeds, image_token_id)
+    }
+
     pub fn clear_kv_cache(&mut self) {
         self.base.clear_kv_cache();
     }

crates/neuron/src/harness/tp/worker.rs

@@ -47,6 +47,28 @@ impl WorkerModel {
         }
     }
 
+    /// Image-bearing forward on this rank. Only the vision-capable
+    /// `qwen3_5` arch has a replicated tower; the dense `qwen3` arch
+    /// errors. The returned logits are discarded by the caller (the
+    /// leader samples from its own rank-0 copy) — the value is the NCCL
+    /// collectives the forward issues.
+    fn forward_with_images(
+        &mut self,
+        input: &candle_core::Tensor,
+        offset: usize,
+        image_pixels: &[candle_core::Tensor],
+        image_token_id: u32,
+    ) -> candle_core::Result<candle_core::Tensor> {
+        match self {
+            WorkerModel::Qwen3_5(m) => {
+                m.forward_with_images(input, offset, image_pixels, image_token_id)
+            }
+            WorkerModel::Qwen3(_) => {
+                candle_core::bail!("forward_with_images: qwen3 (dense) has no vision tower")
+            }
+        }
+    }
+
     fn clear_kv_cache(&mut self) {
         match self {
             WorkerModel::Qwen3(m) => m.clear_kv_cache(),
@@ -167,6 +189,19 @@ impl WorkerState {
                 tokens,
                 offset,
             } => self.handle_generate_step(&model_id, tokens, offset),
+            WorkerRequest::GenerateStepWithImages {
+                model_id,
+                tokens,
+                offset,
+                image_token_id,
+                image_data_uris,
+            } => self.handle_generate_step_with_images(
+                &model_id,
+                tokens,
+                offset,
+                image_token_id,
+                image_data_uris,
+            ),
             WorkerRequest::ClearKvCache { model_id } => self.handle_clear_kv_cache(&model_id),
             WorkerRequest::UnloadModel { model_id } => self.handle_unload_model(&model_id),
             WorkerRequest::Shutdown => WorkerResponse::Bye,
@@ -418,6 +453,124 @@ impl WorkerState {
         }
     }
 
+    /// Image-bearing prefill on this rank. Preprocesses each source data
+    /// URI through the same deterministic `preprocess_data_uri` the
+    /// leader runs, encodes through this rank's replicated tower, and
+    /// splices + forwards. The logits are discarded (the leader samples
+    /// from rank 0); the row-parallel `AllReduce`s are the point.
+    #[cfg(feature = "cuda")]
+    fn handle_generate_step_with_images(
+        &mut self,
+        model_id: &str,
+        tokens: Vec<u32>,
+        offset: usize,
+        image_token_id: u32,
+        image_data_uris: Vec<String>,
+    ) -> WorkerResponse {
+        use crate::harness::preprocess::{PreprocessProfile, preprocess_data_uri};
+        use candle_core::Tensor;
+
+        if image_data_uris.is_empty() {
+            return WorkerResponse::Error {
+                kind: "bad_request".into(),
+                message: "GenerateStepWithImages with zero images".into(),
+            };
+        }
+        let Some(model) = self.models.get_mut(model_id) else {
+            return WorkerResponse::Error {
+                kind: "model_not_loaded".into(),
+                message: format!("model '{model_id}' not loaded on rank {}", self.config.rank),
+            };
+        };
+        let device = model.device().clone();
+
+        // Preprocess each image identically to the leader so the encoded
+        // embeddings — and thus the spliced hidden state — match across
+        // ranks. Fixed 448×448 profile.
+        let profile = PreprocessProfile::qwen3_6();
+        let (h, w) = (
+            profile.target_height as usize,
+            profile.target_width as usize,
+        );
+        let mut pixels: Vec<Tensor> = Vec::with_capacity(image_data_uris.len());
+        for (idx, uri) in image_data_uris.iter().enumerate() {
+            let px = match preprocess_data_uri(uri, &profile) {
+                Ok(p) => p,
+                Err(e) => {
+                    return WorkerResponse::Error {
+                        kind: "bad_request".into(),
+                        message: format!("preprocess image[{idx}]: {e:#}"),
+                    };
+                }
+            };
+            match Tensor::from_vec(px, (3, h, w), &device) {
+                Ok(t) => pixels.push(t),
+                Err(e) => {
+                    return WorkerResponse::Error {
+                        kind: "forward_failed".into(),
+                        message: format!("build image[{idx}] tensor: {e}"),
+                    };
+                }
+            }
+        }
+
+        let input = match Tensor::new(tokens.as_slice(), &device).and_then(|t| t.unsqueeze(0)) {
+            Ok(t) => t,
+            Err(e) => {
+                return WorkerResponse::Error {
+                    kind: "forward_failed".into(),
+                    message: format!("build input tensor: {e}"),
+                };
+            }
+        };
+
+        let start = std::time::Instant::now();
+        tracing::debug!(
+            rank = self.config.rank,
+            model = %model_id,
+            tokens = tokens.len(),
+            offset,
+            images = pixels.len(),
+            "worker GenerateStepWithImages: forward starting"
+        );
+        // Drop the logits — the leader samples from its own rank-0 copy.
+        if let Err(e) = model.forward_with_images(&input, offset, &pixels, image_token_id) {
+            tracing::warn!(
+                rank = self.config.rank,
+                model = %model_id,
+                elapsed_ms = start.elapsed().as_millis(),
+                error = %e,
+                "worker GenerateStepWithImages: forward failed"
+            );
+            return WorkerResponse::Error {
+                kind: "forward_failed".into(),
+                message: format!("TP image forward: {e}"),
+            };
+        }
+        tracing::debug!(
+            rank = self.config.rank,
+            model = %model_id,
+            elapsed_ms = start.elapsed().as_millis(),
+            "worker GenerateStepWithImages: forward done"
+        );
+        WorkerResponse::GenerateStepOk
+    }
+
+    #[cfg(not(feature = "cuda"))]
+    fn handle_generate_step_with_images(
+        &mut self,
+        _model_id: &str,
+        _tokens: Vec<u32>,
+        _offset: usize,
+        _image_token_id: u32,
+        _image_data_uris: Vec<String>,
+    ) -> WorkerResponse {
+        WorkerResponse::Error {
+            kind: "cuda_feature_not_enabled".into(),
+            message: "GenerateStepWithImages requires --features cuda".into(),
+        }
+    }
+
     #[cfg(feature = "cuda")]
     fn handle_clear_kv_cache(&mut self, model_id: &str) -> WorkerResponse {
         let Some(model) = self.models.get_mut(model_id) else {