build(neuron): bump cudarc fork to 63327a2 (idempotent abort + Comm Send+Sync)

The fork's new commit makes `Comm: Send + Sync` (asserting NCCL's
thread-safety invariant upstream) and makes `Comm::abort` idempotent via
an `aborted` flag (so abort-then-Drop can't double-free) — strictly
better than the previous Drop-no-panic workaround, and the `abort()`
signature is unchanged so the watchdog call site is unaffected.

Because `Comm` is now `Send + Sync`, `Arc<Comm>` and the `SendComm` /
`NcclState` wrappers auto-derive `Send`/`Sync`, which conflicts (E0119)
with neuron's manual `unsafe impl`s. Remove the four now-redundant impls
— the safety assertion lives upstream in cudarc where it belongs. The
conflict is in cuda-gated code, so only the CUDA type-check catches it
(non-cuda build + clippy + tests stay green).

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

Cargo.lock

@@ -905,8 +905,7 @@ dependencies = [
 [[package]]
 name = "cudarc"
 version = "0.19.7"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1cea5f10a99e025c1b44ae2354c2d8326b25ddbd0baf76bde8e55cfd4018a2cc"
+source = "git+https://github.com/grenade/cudarc?rev=63327a256059f8252641ae46c6bb9eefe707f382#63327a256059f8252641ae46c6bb9eefe707f382"
 dependencies = [
  "float8",
  "half",

Cargo.toml

@@ -61,3 +61,12 @@ eventsource-stream = "0.2"
 # workspace crates
 cortex-core = { path = "crates/cortex-core" }
 cortex-gateway = { path = "crates/cortex-gateway" }
+
+# Patched cudarc (affects neuron's 0.19.x only; candle's 0.17.x is
+# untouched since the fork is 0.19.7 and doesn't satisfy a 0.17 req). Adds
+# Comm::abort / get_async_error / raw comm() — needed for #17 Stage 2 TP
+# hang-recovery (abort a wedged collective from another thread, then
+# rebuild the comm). Pinned to a fork revision pending upstream review
+# (grenade/cudarc @ nccl-comm-abort).
+[patch.crates-io]
+cudarc = { git = "https://github.com/grenade/cudarc", rev = "63327a256059f8252641ae46c6bb9eefe707f382" }

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

@@ -404,7 +404,7 @@ impl Qwen3_5Model {
     }
 
     pub fn forward(&mut self, input: &Tensor, offset: usize) -> candle_core::Result<Tensor> {
-        self.forward_inner(input, offset, None, None)
+        self.forward_inner(input, offset, None, None, &[])
     }
 
     /// Forward with image-embedding splice. Stage B of the vision plan.
@@ -422,23 +422,25 @@ impl Qwen3_5Model {
     ///
     /// The splice replaces the LM's text-side embedding at each
     /// `image_token_id` position with the corresponding row from
-    /// `image_embeds`. After the splice the decoder runs unchanged.
-    ///
-    /// **MRoPE gap.** Qwen3.6's `rope_parameters` declares MRoPE
-    /// (interleaved text/height/width axes); Stage B applies plain
-    /// text-position RoPE to image tokens. The model still attends
-    /// to image content but loses spatial structure that MRoPE-aware
-    /// position encoding would preserve. Tracked under issue #15
-    /// (numerical validation) — quality benchmark from Stage D should
-    /// surface the impact, and the fix lives in `rope::RotaryEmbedding`.
+    /// `image_embeds`. After the splice the decoder runs the interleaved
+    /// M-RoPE path: `grids` carries each image's post-merge LM grid
+    /// `(lm_gh, lm_gw)` so `get_rope_index` assigns image tokens their 2D
+    /// coordinates (dynamic resolution, #14).
     pub fn forward_with_vision(
         &mut self,
         input_ids: &Tensor,
         offset: usize,
         image_embeds: &Tensor,
         image_token_id: u32,
+        grids: &[(usize, usize)],
     ) -> candle_core::Result<Tensor> {
-        self.forward_inner(input_ids, offset, Some(image_embeds), Some(image_token_id))
+        self.forward_inner(
+            input_ids,
+            offset,
+            Some(image_embeds),
+            Some(image_token_id),
+            grids,
+        )
     }
 
     fn forward_inner(
@@ -447,13 +449,14 @@ impl Qwen3_5Model {
         offset: usize,
         image_embeds: Option<&Tensor>,
         image_token_id: Option<u32>,
+        grids: &[(usize, usize)],
     ) -> candle_core::Result<Tensor> {
         let (b, l) = input.dims2()?;
         let mut h = self.embed_tokens.forward(input)?;
 
         // Vision path: splice image embeddings at `image_token_id`
         // positions and build interleaved M-RoPE cos/sin so image tokens
-        // carry their 14×14 grid coordinates. Text / decode skip the
+        // carry their 2D (lm_gh × lm_gw) grid coordinates. Text / decode skip the
         // device→host id copy entirely and take the plain-RoPE fast path
         // — bit-for-bit the pre-M-RoPE behaviour when `rope_delta == 0`.
         let (cos, sin) = if let (Some(img), Some(tok_id)) = (image_embeds, image_token_id) {
@@ -483,7 +486,7 @@ impl Qwen3_5Model {
                 h = splice_runs(&h, &img, &positions)?;
             }
 
-            let (text, height, width, delta) = rope::get_rope_index(&ids, tok_id)
+            let (text, height, width, delta) = rope::get_rope_index(&ids, tok_id, grids)
                 .map_err(|e| candle_core::Error::Msg(format!("get_rope_index: {e}")))?;
             self.rope_delta = delta;
             let pos = rope::mrope_position_tensor(&text, &height, &width, &self.device)?;
@@ -603,11 +606,12 @@ impl Qwen3_5ForCausalLM {
         offset: usize,
         image_embeds: &Tensor,
         image_token_id: u32,
+        grids: &[(usize, usize)],
     ) -> candle_core::Result<Tensor> {
         let (_, l) = input.dims2()?;
-        let hidden = self
-            .base
-            .forward_with_vision(input, offset, image_embeds, image_token_id)?;
+        let hidden =
+            self.base
+                .forward_with_vision(input, offset, image_embeds, image_token_id, grids)?;
         hidden.i((.., l - 1.., ..))?.apply(&self.lm_head)
     }
 

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

@@ -260,28 +260,40 @@ pub(crate) fn mrope_enabled() -> bool {
 /// off, returns plain sequential identity positions on all three axes
 /// (`mrope_cos_sin` then reduces exactly to plain RoPE), restoring the
 /// pre-M-RoPE behaviour without touching the rest of the forward.
-pub(crate) fn get_rope_index(input_ids: &[u32], image_token_id: u32) -> Result<MRopeIndex> {
+pub(crate) fn get_rope_index(
+    input_ids: &[u32],
+    image_token_id: u32,
+    grids: &[(usize, usize)],
+) -> Result<MRopeIndex> {
     if !mrope_enabled() {
         let seq: Vec<i64> = (0..input_ids.len() as i64).collect();
         return Ok((seq.clone(), seq.clone(), seq, 0));
     }
-    compute_mrope_index(input_ids, image_token_id)
+    compute_mrope_index(input_ids, image_token_id, grids)
 }
 
 /// The real interleaved-M-RoPE position-id computation (always active in
 /// unit tests; gated behind [`get_rope_index`] at runtime).
 ///
-/// Fixed-resolution assumption (Stage C): each image run is a perfect
-/// square with `grid_t = 1` (still image) and `grid_h = grid_w =
-/// isqrt(run_len)` — 196 → 14×14. Dynamic resolution (#14) would thread
-/// real per-image grids instead.
-pub(crate) fn compute_mrope_index(input_ids: &[u32], image_token_id: u32) -> Result<MRopeIndex> {
+/// `grids` carries the post-merge LM grid `(lm_gh, lm_gw)` for each image
+/// run, in prompt order — a run length alone cannot recover its
+/// factorisation, so the grids must be passed (#14 dynamic resolution).
+/// Each image is a still frame (`grid_t = 1`); its tokens get
+/// `[base, base + hh, base + ww]` row-major and the shared counter
+/// resumes at `base + max(lm_gh, lm_gw)`. Multi-image is correct because
+/// the counter threads across images and interleaved text.
+pub(crate) fn compute_mrope_index(
+    input_ids: &[u32],
+    image_token_id: u32,
+    grids: &[(usize, usize)],
+) -> Result<MRopeIndex> {
     let n = input_ids.len();
     let mut text = Vec::with_capacity(n);
     let mut height = Vec::with_capacity(n);
     let mut width = Vec::with_capacity(n);
     let mut counter: i64 = 0;
     let mut i = 0;
+    let mut k = 0; // index into `grids`, one per image run
     while i < n {
         if input_ids[i] == image_token_id {
             let start = i;
@@ -289,25 +301,30 @@ pub(crate) fn compute_mrope_index(input_ids: &[u32], image_token_id: u32) -> Res
                 i += 1;
             }
             let run = i - start;
-            let g = run.isqrt();
-            if g * g != run {
+            let (grid_h, grid_w) = *grids.get(k).ok_or_else(|| {
+                anyhow::anyhow!(
+                    "get_rope_index: image run #{k} (len {run}) has no matching grid \
+                     ({} grids supplied)",
+                    grids.len()
+                )
+            })?;
+            k += 1;
+            if grid_h * grid_w != run {
                 anyhow::bail!(
-                    "get_rope_index: image run length {run} is not a perfect square \
-                     (fixed-resolution Stage C assumes a square grid; dynamic resolution is #14)"
+                    "get_rope_index: image run #{} length {run} != grid {grid_h}×{grid_w} = {}",
+                    k - 1,
+                    grid_h * grid_w
                 );
             }
-            let (grid_t, grid_h, grid_w) = (1usize, g, g);
             let base = counter;
-            for tt in 0..grid_t {
-                for hh in 0..grid_h {
-                    for ww in 0..grid_w {
-                        text.push(base + tt as i64);
-                        height.push(base + hh as i64);
-                        width.push(base + ww as i64);
-                    }
+            for hh in 0..grid_h {
+                for ww in 0..grid_w {
+                    text.push(base); // grid_t = 1 → temporal axis const
+                    height.push(base + hh as i64);
+                    width.push(base + ww as i64);
                 }
             }
-            counter = base + grid_t.max(grid_h).max(grid_w) as i64;
+            counter = base + grid_h.max(grid_w) as i64;
         } else {
             text.push(counter);
             height.push(counter);
@@ -316,6 +333,12 @@ pub(crate) fn compute_mrope_index(input_ids: &[u32], image_token_id: u32) -> Res
             i += 1;
         }
     }
+    if k != grids.len() {
+        anyhow::bail!(
+            "get_rope_index: prompt has {k} image run(s) but {} grid(s) were supplied",
+            grids.len()
+        );
+    }
     let delta = counter - n as i64;
     Ok((text, height, width, delta))
 }
@@ -447,7 +470,7 @@ mod tests {
 
     #[test]
     fn get_rope_index_text_only_is_sequential() {
-        let (t, h, w, delta) = compute_mrope_index(&[1, 2, 3, 4], 99).unwrap();
+        let (t, h, w, delta) = compute_mrope_index(&[1, 2, 3, 4], 99, &[]).unwrap();
         assert_eq!(t, vec![0, 1, 2, 3]);
         assert_eq!(h, vec![0, 1, 2, 3]);
         assert_eq!(w, vec![0, 1, 2, 3]);
@@ -456,12 +479,12 @@ mod tests {
 
     #[test]
     fn get_rope_index_text_image_text() {
-        // [text, image(2x2 run of 4), text]. image_token = 99.
+        // [text, image(2x2 run of 4), text]. image_token = 99, grid (2,2).
         let ids = [1u32, 99, 99, 99, 99, 2];
-        let (t, h, w, delta) = compute_mrope_index(&ids, 99).unwrap();
-        // token 0: text → 0. image base=1, grid 1x2x2:
+        let (t, h, w, delta) = compute_mrope_index(&ids, 99, &[(2, 2)]).unwrap();
+        // token 0: text → 0. image base=1, grid 2x2:
         //   t all = 1; h = base+row = [1,1,2,2]; w = base+col = [1,2,1,2].
-        // resume from base + max(1,2,2) = 3. trailing text → 3.
+        // resume from base + max(2,2) = 3. trailing text → 3.
         assert_eq!(t, vec![0, 1, 1, 1, 1, 3]);
         assert_eq!(h, vec![0, 1, 1, 2, 2, 3]);
         assert_eq!(w, vec![0, 1, 2, 1, 2, 3]);
@@ -472,25 +495,52 @@ mod tests {
         assert_eq!(6 + delta, 4);
     }
 
+    #[test]
+    fn get_rope_index_nonsquare_single_image() {
+        // text + image(2 rows × 3 cols = 6 tokens). grid (2,3).
+        let ids = [1u32, 99, 99, 99, 99, 99, 99];
+        let (t, h, w, delta) = compute_mrope_index(&ids, 99, &[(2, 3)]).unwrap();
+        // base = 1; row-major h = [0,0,0,1,1,1]+1, w = [0,1,2,0,1,2]+1.
+        assert_eq!(t, vec![0, 1, 1, 1, 1, 1, 1]);
+        assert_eq!(h, vec![0, 1, 1, 1, 2, 2, 2]);
+        assert_eq!(w, vec![0, 1, 2, 3, 1, 2, 3]);
+        // resume from base + max(2,3) = 4; seq_len 7, counter 4 → delta -3.
+        assert_eq!(delta, 4 - 7);
+    }
+
+    #[test]
+    fn get_rope_index_two_images_different_grids() {
+        // img(2x2)=4, text, img(1x3)=3. grids [(2,2),(1,3)].
+        let ids = [99, 99, 99, 99, 7, 99, 99, 99];
+        let (t, h, w, delta) = compute_mrope_index(&ids, 99, &[(2, 2), (1, 3)]).unwrap();
+        // img1 base=0 → t=0, h=[0,0,1,1], w=[0,1,0,1]; resume max(2,2)=2.
+        // text at counter 2. img2 base=3 → t=3, h=[3,3,3], w=[3,4,5];
+        // resume 3+max(1,3)=6.
+        assert_eq!(t, vec![0, 0, 0, 0, 2, 3, 3, 3]);
+        assert_eq!(h, vec![0, 0, 1, 1, 2, 3, 3, 3]);
+        assert_eq!(w, vec![0, 1, 0, 1, 2, 3, 4, 5]);
+        assert_eq!(delta, 6 - 8);
+    }
+
     #[test]
     fn get_rope_index_on_by_default() {
         // With NEURON_MROPE unset (default ON), the runtime path returns
-        // the real interleaved-M-RoPE positions, so image tokens carry
-        // their 2D grid coords (height differs from the text counter).
-        // (NEURON_MROPE=0 would fall back to identity; not asserted here
-        // since it depends on env.)
-        let (t, h, w, _delta) = get_rope_index(&[1, 99, 99, 99, 99, 2], 99).unwrap();
-        // Same as compute_mrope_index: 2x2 image after one text token.
+        // the real interleaved-M-RoPE positions. (NEURON_MROPE=0 would fall
+        // back to identity; not asserted here since it depends on env.)
+        let (t, h, w, _delta) = get_rope_index(&[1, 99, 99, 99, 99, 2], 99, &[(2, 2)]).unwrap();
         assert_eq!(t, vec![0, 1, 1, 1, 1, 3]);
         assert_eq!(h, vec![0, 1, 1, 2, 2, 3]);
         assert_eq!(w, vec![0, 1, 2, 1, 2, 3]);
     }
 
     #[test]
-    fn get_rope_index_rejects_non_square_image_run() {
-        // 196 is square (14x14) — ok. 195 is not.
-        assert!(compute_mrope_index(&[99u32; 196], 99).is_ok());
-        assert!(compute_mrope_index(&[99u32; 195], 99).is_err());
+    fn get_rope_index_grid_mismatches_error() {
+        // run length != grid product.
+        assert!(compute_mrope_index(&[99u32; 6], 99, &[(2, 2)]).is_err());
+        // too few grids for the number of image runs.
+        assert!(compute_mrope_index(&[99, 99, 7, 99], 99, &[(1, 2)]).is_err());
+        // too many grids.
+        assert!(compute_mrope_index(&[99, 99], 99, &[(1, 2), (1, 1)]).is_err());
     }
 
     #[test]
@@ -501,7 +551,7 @@ mod tests {
         let dev = Device::Cpu;
         let rope = RotaryEmbedding::new(DType::F32, &qwen36_cfg(), &dev).unwrap();
         let ids = [1u32, 99, 99, 99, 99]; // text + 2x2 image
-        let (t, h, w, _d) = compute_mrope_index(&ids, 99).unwrap();
+        let (t, h, w, _d) = compute_mrope_index(&ids, 99, &[(2, 2)]).unwrap();
         let pos = mrope_position_tensor(&t, &h, &w, &dev).unwrap();
         assert_eq!(pos.dims(), &[3, 5]);
         let (cos, _sin) = rope.mrope_cos_sin(&pos).unwrap();
@@ -518,7 +568,7 @@ mod tests {
     fn get_rope_index_196_is_14x14() {
         let mut ids = vec![1u32]; // one text token
         ids.extend(std::iter::repeat_n(99u32, 196));
-        let (t, h, w, _delta) = compute_mrope_index(&ids, 99).unwrap();
+        let (t, h, w, _delta) = compute_mrope_index(&ids, 99, &[(14, 14)]).unwrap();
         // image base = 1. Last image token (index 196) is grid (h=13,w=13).
         assert_eq!(*t.last().unwrap(), 1, "grid_t=1 → temporal const at base");
         assert_eq!(h[1], 1, "first image row at base");

crates/neuron/src/harness/candle.rs

@@ -60,6 +60,17 @@ pub struct CandleHarness {
     /// can still load on CPU for tests, just without worker threads).
     #[allow(dead_code)]
     device_workers: Arc<RwLock<HashMap<u32, Arc<super::device_worker::DeviceWorkerHandle>>>>,
+    /// Auto-recovery (#17): model ids whose poisoned context is being
+    /// rebuilt via unload+reload. Insert is the single-flight gate (one
+    /// recovery per model in flight); membership also lets the request
+    /// path answer "recovering, retry shortly" during the reload gap
+    /// rather than a bare "not loaded".
+    recovering: Arc<RwLock<std::collections::HashSet<String>>>,
+    /// Sender to the background recovery task. The request path enqueues
+    /// a poisoned model id here; the task (holding a `Weak<Self>`) runs
+    /// the unload→reload→health-gate. Unbounded + tiny (model ids), and
+    /// the `recovering` set dedupes, so it can't back up.
+    recovery_tx: tokio::sync::mpsc::UnboundedSender<String>,
 }
 
 /// One entry in the harness's loaded-model registry. Single-GPU loads
@@ -86,6 +97,15 @@ impl LoadedHandle {
         }
     }
 
+    /// The spec this model was loaded from (for auto-recovery #17).
+    pub fn spec(&self) -> &ModelSpec {
+        match self {
+            LoadedHandle::Single(m) => &m.spec,
+            #[cfg(feature = "cuda")]
+            LoadedHandle::Tp(m) => &m.spec,
+        }
+    }
+
     pub fn devices(&self) -> Vec<u32> {
         match self {
             LoadedHandle::Single(m) => m.devices.clone(),
@@ -210,13 +230,15 @@ pub struct LoadedModel {
     /// targets and the worker forward uses it to locate splice
     /// positions in the LM input embeddings.
     pub image_token_id: Option<u32>,
-    /// LM-side tokens this model's vision tower emits per image at
-    /// the Stage B fixed resolution (448×448 → 196 for Qwen3.6).
-    /// `None` for text-only models. Set at load time so the
-    /// hot path doesn't recompute it per request. Stage B fixed
-    /// resolution → constant; dynamic resolution per #14 makes it
-    /// per-image.
-    pub lm_tokens_per_image: Option<usize>,
+    /// `patch_size × spatial_merge_size` — divides a resized pixel
+    /// dimension into LM-grid units. Per-image LM token count is
+    /// `(h/factor) × (w/factor)` (#14 dynamic resolution). `None` for
+    /// text-only models. Set at load time.
+    pub image_grid_factor: Option<usize>,
+    /// The spec this model was loaded from — retained so auto-recovery
+    /// (#17) can `unload_model` + `load_model(spec)` a poisoned model
+    /// without an operator reconstructing it.
+    pub spec: ModelSpec,
 }
 
 impl LoadedModel {
@@ -288,9 +310,12 @@ pub struct TpLoadedModel {
     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>,
+    /// Pixel→LM-grid divisor — same as
+    /// [`LoadedModel::image_grid_factor`].
+    pub image_grid_factor: Option<usize>,
+    /// Loading spec, retained for auto-recovery (#17) — see
+    /// [`LoadedModel::spec`].
+    pub spec: ModelSpec,
 }
 
 #[cfg(feature = "cuda")]
@@ -394,10 +419,11 @@ impl ModelArch {
         offset: usize,
         image_embeds: &Tensor,
         image_token_id: u32,
+        grids: &[(usize, usize)],
     ) -> Result<Tensor> {
         let raw = match self {
             ModelArch::Qwen3_5Dense(m) => {
-                m.forward_with_vision(input, offset, image_embeds, image_token_id)?
+                m.forward_with_vision(input, offset, image_embeds, image_token_id, grids)?
             }
             other => anyhow::bail!(
                 "forward_with_vision: architecture {} has no vision tower",
@@ -407,6 +433,20 @@ impl ModelArch {
         squeeze_to_vocab(&raw)
     }
 
+    /// `patch_size × spatial_merge_size` for the loaded vision tower —
+    /// divides a resized pixel dim into LM-grid units (an image of
+    /// resized `(h, w)` yields the LM grid `(h/factor, w/factor)`).
+    /// `None` for architectures/checkpoints without a vision tower.
+    pub fn vision_grid_factor(&self) -> Option<usize> {
+        match self {
+            ModelArch::Qwen3_5Dense(m) => m.vision().map(|v| {
+                let c = v.config();
+                c.patch_size * c.spatial_merge_size
+            }),
+            _ => None,
+        }
+    }
+
     /// Encode a preprocessed image into LM-side token embeddings via
     /// the loaded vision tower. Stage A5.
     ///
@@ -779,6 +819,46 @@ fn poisoned_error(model_id: &str) -> InferenceError {
     ))
 }
 
+/// Reported while auto-recovery (#17) is rebuilding a poisoned model's
+/// context. Unlike [`poisoned_error`] this is a *transient* state — the
+/// model is being reloaded automatically; the client should retry.
+fn recovering_error(model_id: &str) -> InferenceError {
+    InferenceError::Other(anyhow::anyhow!(
+        "model '{model_id}' is recovering (its device context was poisoned \
+         by an earlier failure and is being automatically rebuilt); retry \
+         shortly"
+    ))
+}
+
+/// Verification hook for #17 auto-recovery. When `NEURON_DEBUG_POISON`
+/// names a model, the **first** request for it (process-wide) returns
+/// true, so the request path can trigger recovery as if a device fault
+/// had occurred — exercising the unload→reload→healthy cycle without
+/// corrupting the GPU. One-shot (a `swap` latch) so it can't loop the
+/// model through endless recoveries. No-op unless the env var is set.
+fn debug_poison_armed(model_id: &str) -> bool {
+    static FIRED: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
+    let armed = std::env::var("NEURON_DEBUG_POISON").ok().as_deref() == Some(model_id);
+    armed && !FIRED.swap(true, Ordering::Relaxed)
+}
+
+/// Background auto-recovery task (#17). Drains poisoned model ids and
+/// rebuilds each via [`CandleHarness::recover_one`]. Holds a `Weak` so a
+/// shutting-down harness lets the task exit; processes one id at a time,
+/// which (with the `recovering` set deduping enqueues) keeps recovery
+/// single-flight per model.
+async fn recovery_loop(
+    weak: std::sync::Weak<CandleHarness>,
+    mut rx: tokio::sync::mpsc::UnboundedReceiver<String>,
+) {
+    while let Some(model_id) = rx.recv().await {
+        let Some(this) = weak.upgrade() else {
+            break;
+        };
+        this.recover_one(&model_id).await;
+    }
+}
+
 /// Free/total VRAM on the candle `Device` in MiB. Returns `(0, 0)` if
 /// the query fails or the device is the CPU fallback so logging never
 /// crashes the request path. Mirrors the existing helper in
@@ -1133,7 +1213,7 @@ impl CandleHarness {
     /// Construct a new harness for `bind_url` using `config`. Resolves
     /// every configured source's auth env var and cache dir up front so
     /// the hot load path (`hf_api_for`) is a pure HashMap lookup.
-    pub fn new(bind_url: String, config: &crate::config::CandleHarnessConfig) -> Self {
+    pub fn new(bind_url: String, config: &crate::config::CandleHarnessConfig) -> Arc<Self> {
         let raw_sources = config.effective_sources();
         let default_source = config.effective_default_source().to_string();
         let mut sources = HashMap::with_capacity(raw_sources.len());
@@ -1183,13 +1263,25 @@ impl CandleHarness {
                  bare model ids will fail to resolve until this is fixed"
             );
         }
-        Self {
+        let (recovery_tx, recovery_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
+        let this = Arc::new(Self {
             models: Arc::new(RwLock::new(HashMap::new())),
             sources,
             default_source,
             bind_url,
             device_workers: Arc::new(RwLock::new(HashMap::new())),
+            recovering: Arc::new(RwLock::new(std::collections::HashSet::new())),
+            recovery_tx,
+        });
+        // Background auto-recovery task (#17). Holds a `Weak` so it can't
+        // keep the harness alive. Spawned only when a tokio runtime is
+        // present — sync unit tests that build a harness without one
+        // simply skip it (they don't exercise recovery).
+        if tokio::runtime::Handle::try_current().is_ok() {
+            let weak = Arc::downgrade(&this);
+            tokio::spawn(recovery_loop(weak, recovery_rx));
         }
+        this
     }
 
     /// Scheme to substitute for bare `org/name` model ids. Mirrors the
@@ -1614,7 +1706,17 @@ impl CandleHarness {
             let models = self.models.read().await;
             models.get(&request.model).cloned()
         };
-        let handle = handle.ok_or_else(|| InferenceError::ModelNotLoaded(request.model.clone()))?;
+        let handle = match handle {
+            Some(h) => h,
+            // Absent from the registry: distinguish a genuinely unloaded
+            // model from one whose slot is briefly gone mid auto-recovery
+            // (#17), so the client gets a transient "retry shortly" instead
+            // of a misleading "not loaded".
+            None if self.is_recovering(&request.model).await => {
+                return Err(recovering_error(&request.model));
+            }
+            None => return Err(InferenceError::ModelNotLoaded(request.model.clone())),
+        };
         // The match is technically infallible without `cuda` (only Single
         // exists), but the cfg-gated Tp arm makes this the right shape
         // under both feature flags.
@@ -1644,7 +1746,12 @@ impl CandleHarness {
         if loaded.poisoned.load(Ordering::Acquire) {
             let _g = span.enter();
             tracing::warn!("chat_completion: refusing request, model poisoned");
-            return Err(poisoned_error(&model_id));
+            return Err(self.trigger_recovery(&model_id).await);
+        }
+        if debug_poison_armed(&model_id) {
+            let _g = span.enter();
+            tracing::warn!("NEURON_DEBUG_POISON: forcing auto-recovery (#17 verification)");
+            return Err(self.trigger_recovery(&model_id).await);
         }
 
         // Serialise concurrent requests against this model. Holds for
@@ -1683,11 +1790,11 @@ impl CandleHarness {
                     .ok_or_else(|| InferenceError::VisionUnsupported {
                         model_id: request.model.clone(),
                     })?;
-                let patches_per_image = loaded
-                    .lm_tokens_per_image
-                    .ok_or_else(|| InferenceError::VisionUnsupported {
+                let factor = loaded.image_grid_factor.ok_or_else(|| {
+                    InferenceError::VisionUnsupported {
                         model_id: request.model.clone(),
-                    })?;
+                    }
+                })?;
                 let profile = super::preprocess::PreprocessProfile::qwen3_6();
                 let images = extract_images_from_request(&request, &profile).map_err(|e| {
                     InferenceError::Other(anyhow::anyhow!("extract_images: {e}"))
@@ -1699,7 +1806,12 @@ impl CandleHarness {
                         "request has image content but extractor produced zero images"
                     )));
                 }
-                let per_image_counts: Vec<usize> = vec![patches_per_image; images.len()];
+                // Per-image LM token count from each image's resized grid
+                // (#14 dynamic resolution; was a constant 196).
+                let per_image_counts: Vec<usize> = images
+                    .iter()
+                    .map(|im| (im.h / factor) * (im.w / factor))
+                    .collect();
                 prompt_tokens =
                     expand_image_pad_tokens(&prompt_tokens, image_token_id, &per_image_counts)
                         .map_err(InferenceError::Other)?;
@@ -2018,7 +2130,17 @@ impl CandleHarness {
             let models = self.models.read().await;
             models.get(&request.model).cloned()
         };
-        let handle = handle.ok_or_else(|| InferenceError::ModelNotLoaded(request.model.clone()))?;
+        let handle = match handle {
+            Some(h) => h,
+            // Absent from the registry: distinguish a genuinely unloaded
+            // model from one whose slot is briefly gone mid auto-recovery
+            // (#17), so the client gets a transient "retry shortly" instead
+            // of a misleading "not loaded".
+            None if self.is_recovering(&request.model).await => {
+                return Err(recovering_error(&request.model));
+            }
+            None => return Err(InferenceError::ModelNotLoaded(request.model.clone())),
+        };
         // The match is technically infallible without `cuda` (only Single
         // exists), but the cfg-gated Tp arm makes this the right shape
         // under both feature flags.
@@ -2059,11 +2181,12 @@ impl CandleHarness {
                         .ok_or_else(|| InferenceError::VisionUnsupported {
                             model_id: request.model.clone(),
                         })?;
-                let patches_per_image = loaded.lm_tokens_per_image.ok_or_else(|| {
-                    InferenceError::VisionUnsupported {
-                        model_id: request.model.clone(),
-                    }
-                })?;
+                let factor =
+                    loaded
+                        .image_grid_factor
+                        .ok_or_else(|| InferenceError::VisionUnsupported {
+                            model_id: request.model.clone(),
+                        })?;
                 let profile = super::preprocess::PreprocessProfile::qwen3_6();
                 let images = extract_images_from_request(&request, &profile)
                     .map_err(|e| InferenceError::Other(anyhow::anyhow!("extract_images: {e}")))?;
@@ -2072,7 +2195,11 @@ impl CandleHarness {
                         "request has image content but extractor produced zero images"
                     )));
                 }
-                let per_image_counts: Vec<usize> = vec![patches_per_image; images.len()];
+                // Per-image LM token count from each image's resized grid (#14).
+                let per_image_counts: Vec<usize> = images
+                    .iter()
+                    .map(|im| (im.h / factor) * (im.w / factor))
+                    .collect();
                 prompt_tokens =
                     expand_image_pad_tokens(&prompt_tokens, image_token_id, &per_image_counts)
                         .map_err(InferenceError::Other)?;
@@ -2106,7 +2233,7 @@ impl CandleHarness {
         // Refuse if the model is already poisoned. No point opening
         // an SSE stream just to send the Start event and then bail.
         if loaded.poisoned.load(Ordering::Acquire) {
-            return Err(poisoned_error(&model_id));
+            return Err(self.trigger_recovery(&model_id).await);
         }
 
         // Start event: tells the wire projector to emit its
@@ -2324,6 +2451,69 @@ pub struct InferenceStream {
     pub reasoning_markers: Option<ReasoningTokenPair>,
 }
 
+/// Auto-recovery (#17) — rebuild a poisoned model's device context
+/// automatically instead of leaving it bricked until a human reloads.
+impl CandleHarness {
+    /// True while `model_id` is being auto-recovered (its slot is briefly
+    /// absent from the registry during the reload).
+    pub async fn is_recovering(&self, model_id: &str) -> bool {
+        self.recovering.read().await.contains(model_id)
+    }
+
+    /// Single-flight trigger from the request path: enqueue a rebuild for a
+    /// poisoned model (only the first caller per model enqueues) and return
+    /// the transient "recovering" error to hand back to the client.
+    async fn trigger_recovery(&self, model_id: &str) -> InferenceError {
+        let newly = self.recovering.write().await.insert(model_id.to_string());
+        if newly {
+            tracing::warn!(model = %model_id, "auto-recovery: poisoned, enqueueing rebuild");
+            if self.recovery_tx.send(model_id.to_string()).is_err() {
+                // Background task gone (harness shutting down). Drop the
+                // marker and fall back to the manual-reload message.
+                self.recovering.write().await.remove(model_id);
+                tracing::error!(model = %model_id, "auto-recovery: task unavailable");
+                return poisoned_error(model_id);
+            }
+        }
+        recovering_error(model_id)
+    }
+
+    /// Rebuild a poisoned model: `unload_model` (drops it → cudarc aborts
+    /// NCCL + releases the context) then `load_model` from the retained
+    /// spec. A successful reload re-runs NCCL init + sanity inside the load
+    /// path, so it returns a fresh, healthy model; a failed reload leaves
+    /// the model unloaded (recoverable by the next load), never poisoned
+    /// forever. Runs on the background task — never inline on the request
+    /// path (would deadlock on the `models` write lock).
+    async fn recover_one(&self, model_id: &str) {
+        let spec = {
+            let models = self.models.read().await;
+            models.get(model_id).map(|h| h.spec().clone())
+        };
+        let Some(spec) = spec else {
+            self.recovering.write().await.remove(model_id);
+            return;
+        };
+        tracing::warn!(model = %model_id, "auto-recovery: unload+reload starting");
+        if let Err(e) = self.unload_model(model_id).await {
+            tracing::error!(
+                model = %model_id,
+                error = %format!("{e:#}"),
+                "auto-recovery: unload failed (continuing to reload)"
+            );
+        }
+        match self.load_model(&spec).await {
+            Ok(()) => tracing::info!(model = %model_id, "auto-recovery: reloaded; model healthy"),
+            Err(e) => tracing::error!(
+                model = %model_id,
+                error = %format!("{e:#}"),
+                "auto-recovery: reload failed; model left unloaded"
+            ),
+        }
+        self.recovering.write().await.remove(model_id);
+    }
+}
+
 #[async_trait]
 impl Harness for CandleHarness {
     fn name(&self) -> &str {
@@ -2526,7 +2716,8 @@ impl Harness for CandleHarness {
             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,
+            image_grid_factor: vision_meta.image_grid_factor,
+            spec: spec.clone(),
         });
 
         let mut models = self.models.write().await;
@@ -2742,7 +2933,7 @@ impl CandleHarness {
             tracing::info!(
                 model = %spec.model_id,
                 image_token_id = ?vision_meta.image_token_id,
-                lm_tokens_per_image = ?vision_meta.lm_tokens_per_image,
+                image_grid_factor = ?vision_meta.image_grid_factor,
                 "TP load: vision tower present, advertising vision capability"
             );
         }
@@ -2764,7 +2955,8 @@ impl CandleHarness {
             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,
+            image_grid_factor: vision_meta.image_grid_factor,
+            spec: spec.clone(),
         });
 
         let mut models = self.models.write().await;
@@ -2811,7 +3003,12 @@ impl CandleHarness {
         if tp.poisoned.load(Ordering::Acquire) {
             let _g = span.enter();
             tracing::warn!("TP chat_completion: refusing request, model poisoned");
-            return Err(poisoned_error(&model_id));
+            return Err(self.trigger_recovery(&model_id).await);
+        }
+        if debug_poison_armed(&model_id) {
+            let _g = span.enter();
+            tracing::warn!("NEURON_DEBUG_POISON: forcing auto-recovery (#17 verification)");
+            return Err(self.trigger_recovery(&model_id).await);
         }
 
         // Reject image-bearing requests against a TP model with no
@@ -2900,7 +3097,7 @@ impl CandleHarness {
         request: ChatCompletionRequest,
     ) -> Result<InferenceStream, InferenceError> {
         if tp.poisoned.load(Ordering::Acquire) {
-            return Err(poisoned_error(&request.model));
+            return Err(self.trigger_recovery(&request.model).await);
         }
 
         // Reject image requests against a non-vision TP model before
@@ -2938,18 +3135,32 @@ impl CandleHarness {
                     .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 factor = tp
+                .image_grid_factor
+                .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()];
+            // Per-image LM token count from each image's resized grid (#14).
+            // Decode header + smart_resize only; the workers re-derive the
+            // same dims when they preprocess for the replicated tower.
+            let profile = super::preprocess::PreprocessProfile::qwen3_6();
+            let per_image_counts: Vec<usize> = data_uris
+                .iter()
+                .enumerate()
+                .map(|(i, uri)| {
+                    let (h, w) =
+                        super::preprocess::resized_dims_for_uri(uri, &profile).map_err(|e| {
+                            InferenceError::Other(anyhow::anyhow!("resized_dims image #{i}: {e}"))
+                        })?;
+                    Ok::<usize, InferenceError>((h as usize / factor) * (w as usize / factor))
+                })
+                .collect::<Result<Vec<_>, _>>()?;
             prompt_tokens =
                 expand_image_pad_tokens(&prompt_tokens, image_token_id, &per_image_counts)
                     .map_err(InferenceError::Other)?;
@@ -3457,18 +3668,30 @@ async fn chat_completion_tp_inner(
                 .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 factor = tp
+            .image_grid_factor
+            .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()];
+        // Per-image LM token count from each image's resized grid (#14).
+        let profile = super::preprocess::PreprocessProfile::qwen3_6();
+        let per_image_counts: Vec<usize> = data_uris
+            .iter()
+            .enumerate()
+            .map(|(i, uri)| {
+                let (h, w) =
+                    super::preprocess::resized_dims_for_uri(uri, &profile).map_err(|e| {
+                        InferenceError::Other(anyhow::anyhow!("resized_dims image #{i}: {e}"))
+                    })?;
+                Ok::<usize, InferenceError>((h as usize / factor) * (w as usize / factor))
+            })
+            .collect::<Result<Vec<_>, _>>()?;
         prompt_tokens = expand_image_pad_tokens(&prompt_tokens, image_token_id, &per_image_counts)
             .map_err(InferenceError::Other)?;
         Some((data_uris, image_token_id))
@@ -3917,10 +4140,12 @@ fn build_prompt_for_request(
 struct VisionMeta {
     has_vision: bool,
     image_token_id: Option<u32>,
-    /// LM-side tokens this model's vision tower emits per image at
-    /// the Stage B fixed `PreprocessProfile::qwen3_6()` resolution
-    /// (448×448). Equal to `(H/patch_size/spatial_merge_size)²`.
-    lm_tokens_per_image: Option<usize>,
+    /// `patch_size × spatial_merge_size` — the divisor that turns a
+    /// resized pixel dimension into an LM-grid dimension. An image of
+    /// resized `(h, w)` emits `(h/factor) × (w/factor)` LM tokens (#14
+    /// dynamic resolution; was a constant 196 at the old fixed 448²).
+    /// `None` for text-only models.
+    image_grid_factor: Option<usize>,
 }
 
 impl VisionMeta {
@@ -3949,22 +4174,18 @@ impl VisionMeta {
             .get("image_token_id")
             .and_then(|x| x.as_u64())
             .map(|n| n as u32);
-        // Compute LM tokens per image at the Stage B fixed resolution
-        // (PreprocessProfile::qwen3_6() → 448×448). One LM token per
-        // spatial-merge group of patches.
-        let target_h = super::preprocess::PreprocessProfile::qwen3_6().target_height as usize;
-        let target_w = super::preprocess::PreprocessProfile::qwen3_6().target_width as usize;
-        let lm_tokens_per_image = if patch_size > 0 && spatial_merge_size > 0 {
-            let gh = target_h / patch_size / spatial_merge_size;
-            let gw = target_w / patch_size / spatial_merge_size;
-            Some(gh * gw)
+        // The pixel→LM-grid divisor. An image resized to (h, w) emits
+        // (h/factor) × (w/factor) LM tokens — computed per image at
+        // request time now that resolution is dynamic (#14).
+        let image_grid_factor = if patch_size > 0 && spatial_merge_size > 0 {
+            Some(patch_size * spatial_merge_size)
         } else {
             None
         };
         Self {
             has_vision: true,
             image_token_id,
-            lm_tokens_per_image,
+            image_grid_factor,
         }
     }
 }
@@ -4011,13 +4232,13 @@ fn extract_images_from_request(
                     .and_then(|v| v.get("url"))
                     .and_then(|v| v.as_str())
                     .ok_or_else(|| anyhow::anyhow!("image_url part missing url field"))?;
-                let pixels = super::preprocess::preprocess_data_uri(url, profile)
+                let (pixels, h, w) = super::preprocess::preprocess_data_uri(url, profile)
                     .with_context(|| format!("preprocess image #{}", out.len()))?;
                 out.push(super::device_worker::jobs::ImageInput {
                     pixels,
                     c: 3,
-                    h: profile.target_height as usize,
-                    w: profile.target_width as usize,
+                    h: h as usize,
+                    w: w as usize,
                 });
             }
         }

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

@@ -201,6 +201,16 @@ pub(crate) fn run(device_index: u32, rx: Receiver<Job>, poisoned: Arc<AtomicBool
                 let _ = reply.send(resp);
             }
             #[cfg(feature = "cuda")]
+            Job::GetLeaderComm { reply } => {
+                // Clone the leader's Arc<Comm> out for the async-side
+                // watchdog. `None` before NcclInit. (#17 Stage 2)
+                let comm = state
+                    .nccl
+                    .comm()
+                    .map(crate::harness::tp::nccl_state::SendComm);
+                let _ = reply.send(comm);
+            }
+            #[cfg(feature = "cuda")]
             Job::TpLoadShard {
                 model_id,
                 config_json,
@@ -779,19 +789,17 @@ fn tp_forward_logits_with_images(
         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.
+    // Preprocess every image into a device-resident (C, H, W) tensor at
+    // its native-aspect resized dims (#14). Same `smart_resize` + decode
+    // path the subprocess workers run, so the encoded embeddings — and
+    // the per-image grids derived from these dims — 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)
+        let (px, h, w) = preprocess_data_uri(uri, &profile)
             .with_context(|| format!("preprocess image[{idx}] (TP leader)"))?;
-        let t = Tensor::from_vec(px, (3, h, w), &state.device)?;
+        let t = Tensor::from_vec(px, (3, h as usize, w as usize), &state.device)?;
         pixels.push(t);
     }
 
@@ -877,9 +885,17 @@ fn forward_logits_with_images(
         anyhow::anyhow!("ForwardLogitsWithImages: no model for handle {}", handle.0)
     })?;
 
+    // pixel→LM-grid divisor (patch×merge) for this tower; each image's
+    // LM grid is (h/factor, w/factor) (#14 dynamic resolution).
+    let factor = arch.vision_grid_factor().ok_or_else(|| {
+        anyhow::anyhow!("ForwardLogitsWithImages: loaded model has no vision tower")
+    })?;
+
     // Encode every image on the worker's device, collecting per-image
-    // post-merger embeddings as device-resident tensors.
+    // post-merger embeddings as device-resident tensors plus their LM
+    // grids (for the interleaved-M-RoPE position ids).
     let mut per_image: Vec<Tensor> = Vec::with_capacity(images.len());
+    let mut grids: Vec<(usize, usize)> = Vec::with_capacity(images.len());
     for (idx, img) in images.into_iter().enumerate() {
         anyhow::ensure!(
             img.pixels.len() == img.c * img.h * img.w,
@@ -889,6 +905,7 @@ fn forward_logits_with_images(
             img.h,
             img.w,
         );
+        grids.push((img.h / factor, img.w / factor));
         let image = Tensor::from_vec(img.pixels, (img.c, img.h, img.w), &state.device)?;
         let embed = arch
             .encode_image(&image)
@@ -901,7 +918,7 @@ fn forward_logits_with_images(
     let image_embeds = Tensor::cat(&per_image.iter().collect::<Vec<_>>(), 0)?;
 
     let input = Tensor::new(tokens, &state.device)?.unsqueeze(0)?;
-    let logits = arch.forward_with_vision(&input, offset, &image_embeds, image_token_id)?;
+    let logits = arch.forward_with_vision(&input, offset, &image_embeds, image_token_id, &grids)?;
     let values = logits
         .to_dtype(DType::F32)?
         .flatten_all()?
@@ -997,6 +1014,10 @@ fn drain_poisoned(job: Job, device_index: u32) {
                 message: format!("device worker {device_index} poisoned"),
             });
         }
+        #[cfg(feature = "cuda")]
+        Job::GetLeaderComm { reply } => {
+            let _ = reply.send(None);
+        }
         Job::NcclSanity { reply } => {
             let _ = reply.send(crate::harness::tp::rpc::WorkerResponse::Error {
                 kind: "device_worker_poisoned".into(),

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

@@ -36,8 +36,13 @@ pub struct TpHandle(pub u64);
 /// `Clone` so the vision-aware dispatch in `chat_completion` can
 /// match `&vision_route` (carrying borrowed images) and still hand
 /// owned `Vec<ImageInput>` to the worker job. The clone cost is one
-/// pixel-buffer memcpy per image — fine at fixed-resolution sizes
-/// (3 × 448 × 448 × 4 bytes = ~2.4 MiB per image).
+/// pixel-buffer memcpy per image — now variable with dynamic resolution
+/// (#14): `3 × h × w × 4` bytes, up to ~6.3 MiB at the default 1024²
+/// `max_pixels` budget.
+///
+/// `h`/`w` are the **resized** dims (factor-aligned), so the per-image LM
+/// grid is `(h/factor, w/factor)` — derived downstream for the splice
+/// and the interleaved-M-RoPE position ids.
 #[derive(Clone)]
 pub struct ImageInput {
     pub pixels: Vec<f32>,
@@ -187,6 +192,17 @@ pub enum Job {
     NcclSanity {
         reply: oneshot::Sender<crate::harness::tp::rpc::WorkerResponse>,
     },
+    /// Hand a clonable handle to the leader's NCCL `Comm` back to the
+    /// async side, so the TP step watchdog can call `ncclCommAbort` on
+    /// it from a *different* thread to unblock a wedged collective
+    /// (#17 Stage 2). Fetched once at init while the worker thread is
+    /// still responsive — a thread already wedged in a collective can't
+    /// service this job, which is exactly why the handle is cached
+    /// up front. Replies `None` before `NcclInit` has run.
+    #[cfg(feature = "cuda")]
+    GetLeaderComm {
+        reply: oneshot::Sender<Option<crate::harness::tp::nccl_state::SendComm>>,
+    },
     /// Load the leader's TP shard on the worker thread. The dispatch
     /// handler reads `state.nccl.comm()` directly (no cross-thread
     /// `Arc<Comm>` transfer, no `SendComm` wrapper) and builds the

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

@@ -161,6 +161,27 @@ impl DeviceWorkerHandle {
         }
     }
 
+    /// Fetch a clonable handle to the leader's NCCL `Comm` (#17 Stage 2).
+    /// The TP step watchdog caches this at init so it can call
+    /// `ncclCommAbort` from the async thread to unblock a wedged
+    /// collective. Returns `None` if uninitialised, poisoned, or gone —
+    /// the caller treats a missing handle as "can't abort" and logs it.
+    #[cfg(feature = "cuda")]
+    pub async fn get_leader_comm(&self) -> Option<crate::harness::tp::nccl_state::SendComm> {
+        if self.poisoned.load(Ordering::Acquire) {
+            return None;
+        }
+        let (reply_tx, reply_rx) = oneshot::channel();
+        if self
+            .tx
+            .send(Job::GetLeaderComm { reply: reply_tx })
+            .is_err()
+        {
+            return None;
+        }
+        reply_rx.await.ok().flatten()
+    }
+
     /// Load a GGUF (pre-quantized) single-GPU model on the worker
     /// thread. The hf-hub resolution happens on the async caller; the
     /// resolved local `gguf_path` plus the spec's model_id are sent

crates/neuron/src/harness/mod.rs

@@ -114,10 +114,8 @@ impl HarnessRegistry {
         for config in configs {
             match config.name.as_str() {
                 "candle" => {
-                    let harness = Arc::new(candle::CandleHarness::new(
-                        bind_url.to_string(),
-                        &settings.candle,
-                    ));
+                    let harness =
+                        candle::CandleHarness::new(bind_url.to_string(), &settings.candle);
                     registry.candle = Some(Arc::clone(&harness));
                     registry.harnesses.insert("candle".into(), harness);
                 }

crates/neuron/src/harness/preprocess.rs

@@ -2,11 +2,11 @@
 //!
 //! Decodes `data:image/...;base64,...` URIs from OpenAI-style
 //! `image_url` content parts into the patch tensors a candle vision
-//! tower expects. Stage A ships **fixed resolution** — every image
-//! is resized to the same target dimensions (default 448×448 for
-//! Qwen3.6, configurable per-call) so the patch count is constant
-//! per image. Variable resolution per [Qwen2VL convention] is tracked
-//! as issue #14.
+//! tower expects. Resolution is **dynamic** (#14): each image is
+//! resized to its native aspect via Qwen `smart_resize` — a
+//! factor-aligned `(h, w)` whose pixel count lands in the profile's
+//! `[min_pixels, max_pixels]` budget — so the LM token count varies per
+//! image (`(h/factor) × (w/factor)`).
 //!
 //! Spec reference: `doc/vision-qwen3_6-spec.md` — preprocessor
 //! section.
@@ -21,7 +21,7 @@
 //! Pipeline (per image):
 //!   1. data: URI → base64 decode → bytes
 //!   2. bytes → image::DynamicImage (PNG/JPEG/WebP/etc)
-//!   3. resize_exact to target H×W (pixel space)
+//!   3. smart_resize to a native-aspect, factor-aligned H×W (pixel space)
 //!   4. RGB→f32, normalise per mean/std
 //!   5. layout to (C, H, W) tensor
 //!
@@ -34,39 +34,126 @@ use base64::Engine;
 use image::DynamicImage;
 use image::imageops::FilterType;
 
-/// Preprocessing target. Captures the resize dimensions and the
-/// channel-wise normalisation constants from the model's
-/// `preprocessor_config.json`. Stage A ships a single `qwen3_6()`
-/// constructor for fixed-resolution Qwen3.6 preprocessing; other
-/// models can ship their own profile when added.
+/// Preprocessing target. Captures the resize policy (Qwen `smart_resize`
+/// factor + pixel budget) and the channel-wise normalisation constants
+/// from the model's `preprocessor_config.json`. Images are resized to
+/// their **native aspect** — a factor-aligned `(h, w)` whose pixel count
+/// lands in `[min_pixels, max_pixels]` — not a fixed square (#14).
 #[derive(Debug, Clone)]
 pub struct PreprocessProfile {
-    pub target_height: u32,
-    pub target_width: u32,
+    /// Both output dims are multiples of this. For Qwen3.6 it is
+    /// `patch_size(16) × spatial_merge_size(2) = 32`, so the post-merge
+    /// LM grid is exactly `(h/factor, w/factor)`.
+    pub factor: u32,
+    /// Lower pixel bound — tiny images are upscaled to at least this.
+    pub min_pixels: u32,
+    /// Upper pixel bound — large images are downscaled to at most this.
+    /// Caps per-image LM tokens (`max_pixels / factor²`) and the
+    /// O(patches²) ViT attention cost.
+    pub max_pixels: u32,
     pub image_mean: [f32; 3],
     pub image_std: [f32; 3],
 }
 
+/// The Qwen3.6 vision tower rejects any image whose **patch** count
+/// exceeds its learned pos-embed budget (`num_position_embeddings =
+/// 2304 = 48²`; see `vision.rs`). At `patch_size = 16` that is
+/// `2304 × 16² = 589_824` source pixels. `max_pixels` is hard-capped to
+/// this so `smart_resize` can never produce an over-budget grid — a
+/// per-rank "patch count exceeds pos_embed budget" error mid-TP-forward
+/// would otherwise poison the device context. The pos-embed grid is the
+/// resolution Qwen3.6 was trained at, so this cap is principled, not just
+/// defensive.
+const QWEN3_6_MAX_PIXELS_CAP: u32 = 2304 * 16 * 16; // 589_824 → ≤ 2304 patches → ≤ 576 LM tokens
+
+/// Default pixel budget for Qwen3.6: `256²` (64 LM tokens) up to the
+/// pos-embed cap (576 LM tokens). Generous for documents/OCR, bounded
+/// for serving. Operators lower it with `NEURON_VISION_MIN_PIXELS` /
+/// `NEURON_VISION_MAX_PIXELS` (the upper bound is still clamped to the
+/// cap above — raising it past the budget would poison the model).
+const QWEN3_6_MIN_PIXELS: u32 = 65_536;
+
+fn env_pixels(name: &str, default: u32) -> u32 {
+    std::env::var(name)
+        .ok()
+        .and_then(|v| v.trim().parse::<u32>().ok())
+        .unwrap_or(default)
+}
+
 impl PreprocessProfile {
-    /// Stage A profile for Qwen3.6. Resize to 448×448, normalise to
-    /// `[-1, 1]` via mean=std=0.5. Fits within the model's
-    /// `num_position_embeddings=2304` budget at 28×28 = 784 patches
-    /// before merging.
+    /// Profile for Qwen3.6. Native-aspect `smart_resize` (factor 32),
+    /// normalise to `[-1, 1]` via mean=std=0.5. Pixel budget defaults to
+    /// [`QWEN3_6_MIN_PIXELS`]…[`QWEN3_6_MAX_PIXELS_CAP`], overridable via
+    /// `NEURON_VISION_MIN_PIXELS` / `NEURON_VISION_MAX_PIXELS`. Clamped
+    /// sane: `factor² ≤ min ≤ max`, and `max ≤` the pos-embed cap (so the
+    /// vision tower never rejects a resized image and poisons the context).
     pub fn qwen3_6() -> Self {
+        let factor = 32u32;
+        let f2 = factor * factor;
+        let min_pixels = env_pixels("NEURON_VISION_MIN_PIXELS", QWEN3_6_MIN_PIXELS)
+            .max(f2)
+            .min(QWEN3_6_MAX_PIXELS_CAP);
+        let max_pixels = env_pixels("NEURON_VISION_MAX_PIXELS", QWEN3_6_MAX_PIXELS_CAP)
+            .min(QWEN3_6_MAX_PIXELS_CAP)
+            .max(min_pixels);
         Self {
-            target_height: 448,
-            target_width: 448,
+            factor,
+            min_pixels,
+            max_pixels,
             image_mean: [0.5, 0.5, 0.5],
             image_std: [0.5, 0.5, 0.5],
         }
     }
 
-    /// Per-channel CHW tensor length: 3 * H * W.
-    pub fn pixels_chw(&self) -> usize {
-        3 * (self.target_height as usize) * (self.target_width as usize)
+    /// The factor-aligned `(h, w)` this profile would resize a source
+    /// `src_h × src_w` image to. Pure integer policy — no pixel work.
+    pub fn resized_dims(&self, src_h: u32, src_w: u32) -> Result<(u32, u32)> {
+        smart_resize(src_h, src_w, self.factor, self.min_pixels, self.max_pixels)
     }
 }
 
+/// Qwen `smart_resize`: the smallest `factor`-aligned `(h_bar, w_bar)`
+/// that preserves aspect ratio as closely as possible while keeping the
+/// pixel count within `[min_pixels, max_pixels]`. Direct port of the
+/// canonical Qwen2-VL / Qwen3-VL image-processor function (so neuron's
+/// grid matches what the model was trained on).
+///
+/// Returns `(height, width)`. Errors if the aspect ratio exceeds 200:1
+/// (degenerate input — a 1-pixel-tall strip), matching upstream.
+pub fn smart_resize(
+    height: u32,
+    width: u32,
+    factor: u32,
+    min_pixels: u32,
+    max_pixels: u32,
+) -> Result<(u32, u32)> {
+    let h = height.max(1) as f64;
+    let w = width.max(1) as f64;
+    let ratio = h.max(w) / h.min(w);
+    if ratio > 200.0 {
+        anyhow::bail!(
+            "image aspect ratio {ratio:.1}:1 exceeds the 200:1 limit ({height}×{width}); \
+             refusing to resize"
+        );
+    }
+    let f = factor as f64;
+    let (minp, maxp) = (min_pixels as f64, max_pixels as f64);
+    // round-to-nearest-factor (may be 0 for sub-factor inputs; the
+    // min-pixels branch below grows it back up).
+    let mut h_bar = (h / f).round() * f;
+    let mut w_bar = (w / f).round() * f;
+    if h_bar * w_bar > maxp {
+        let beta = (h * w / maxp).sqrt();
+        h_bar = f.max((h / beta / f).floor() * f);
+        w_bar = f.max((w / beta / f).floor() * f);
+    } else if h_bar * w_bar < minp {
+        let beta = (minp / (h * w)).sqrt();
+        h_bar = (h * beta / f).ceil() * f;
+        w_bar = (w * beta / f).ceil() * f;
+    }
+    Ok((h_bar as u32, w_bar as u32))
+}
+
 /// Decode a `data:image/...;base64,...` URI into an in-memory image.
 ///
 /// Accepts the OpenAI Chat Completions `image_url` shape — a string
@@ -106,16 +193,13 @@ pub fn decode_data_uri(uri: &str) -> Result<DynamicImage> {
 /// faster on CPU. Quality difference is marginal for downstream
 /// vision-encoder consumption. The numerical-validation issue (#15)
 /// will quantify any discrepancy.
-pub fn preprocess(img: &DynamicImage, profile: &PreprocessProfile) -> Vec<f32> {
+pub fn preprocess(img: &DynamicImage, profile: &PreprocessProfile) -> Result<(Vec<f32>, u32, u32)> {
+    let (h_bar, w_bar) = profile.resized_dims(img.height(), img.width())?;
     let rgb = img
-        .resize_exact(
-            profile.target_width,
-            profile.target_height,
-            FilterType::Triangle,
-        )
+        .resize_exact(w_bar, h_bar, FilterType::Triangle)
         .to_rgb8();
-    let h = profile.target_height as usize;
-    let w = profile.target_width as usize;
+    let h = h_bar as usize;
+    let w = w_bar as usize;
     let mut out = vec![0.0_f32; 3 * h * w];
     // Row-major (C, H, W). Candle's Conv2d expects NCHW, so this is
     // the natural layout — the caller stacks `n` of these along the
@@ -131,16 +215,27 @@ pub fn preprocess(img: &DynamicImage, profile: &PreprocessProfile) -> Vec<f32> {
             }
         }
     }
-    out
+    Ok((out, h_bar, w_bar))
 }
 
-/// Combined helper: decode + preprocess in one call. Most call
-/// sites just want the final tensor; the two-step path exists for
-/// callers (tests, future video preprocessing) that need the
+/// Combined helper: decode + preprocess in one call. Returns the
+/// `(3, h, w)` row-major pixels plus the resized `(h, w)` — the caller
+/// needs the dims to build the tensor and to derive the LM token grid
+/// `(h/factor, w/factor)`. Most call sites use this; the two-step path
+/// exists for callers (tests, future video preprocessing) that need the
 /// intermediate `DynamicImage`.
-pub fn preprocess_data_uri(uri: &str, profile: &PreprocessProfile) -> Result<Vec<f32>> {
+pub fn preprocess_data_uri(uri: &str, profile: &PreprocessProfile) -> Result<(Vec<f32>, u32, u32)> {
     let img = decode_data_uri(uri)?;
-    Ok(preprocess(&img, profile))
+    preprocess(&img, profile)
+}
+
+/// Resized `(h, w)` for a data-URI image **without** running the pixel
+/// normalisation — decode header + `smart_resize` only. Lets a caller
+/// that just needs the LM token count (e.g. the TP leader expanding the
+/// prompt) avoid materialising the full pixel tensor twice.
+pub fn resized_dims_for_uri(uri: &str, profile: &PreprocessProfile) -> Result<(u32, u32)> {
+    let img = decode_data_uri(uri)?;
+    profile.resized_dims(img.height(), img.width())
 }
 
 #[cfg(test)]
@@ -205,13 +300,17 @@ mod tests {
         // decoding so this test isolates the resize+normalise path.
         let img: ImageBuffer<Rgb<u8>, Vec<u8>> = ImageBuffer::from_pixel(2, 2, Rgb([255, 0, 0]));
         let dyn_img = DynamicImage::ImageRgb8(img);
-        let out = preprocess(&dyn_img, &profile);
+        let (out, h_bar, w_bar) = preprocess(&dyn_img, &profile).expect("preprocess");
 
-        assert_eq!(out.len(), profile.pixels_chw());
+        let h = h_bar as usize;
+        let w = w_bar as usize;
+        assert_eq!(out.len(), 3 * h * w);
+        // Dims are factor-aligned and at least the min-pixel floor.
+        assert_eq!(h_bar % profile.factor, 0);
+        assert_eq!(w_bar % profile.factor, 0);
+        assert!(h * w >= profile.min_pixels as usize);
         // After mean=0.5, std=0.5: red channel (255/255=1.0) → (1.0 - 0.5)/0.5 = 1.0
         // green/blue (0.0) → (0.0 - 0.5)/0.5 = -1.0
-        let h = profile.target_height as usize;
-        let w = profile.target_width as usize;
         assert!(
             (out[0] - 1.0).abs() < 1e-5,
             "R[0] should be 1.0, got {}",
@@ -229,9 +328,12 @@ mod tests {
     #[test]
     fn preprocess_data_uri_end_to_end() {
         let profile = PreprocessProfile::qwen3_6();
-        let out = preprocess_data_uri(&red_png_uri(), &profile).expect("e2e preprocess");
-        assert_eq!(out.len(), profile.pixels_chw());
+        let (out, h, w) = preprocess_data_uri(&red_png_uri(), &profile).expect("e2e preprocess");
+        assert_eq!(out.len(), 3 * h as usize * w as usize);
         assert!(out.iter().all(|v| v.is_finite()));
+        // resized_dims_for_uri agrees with the full preprocess.
+        let (h2, w2) = resized_dims_for_uri(&red_png_uri(), &profile).expect("dims");
+        assert_eq!((h, w), (h2, w2));
     }
 
     #[test]
@@ -240,10 +342,10 @@ mod tests {
         // 1x1 grayscale = 200 → after conversion to RGB, all three
         // channels equal 200, normalised → (200/255 - 0.5)/0.5 ≈ 0.569
         let gray = DynamicImage::ImageLuma8(ImageBuffer::from_pixel(1, 1, image::Luma([200])));
-        let out = preprocess(&gray, &profile);
+        let (out, h_bar, w_bar) = preprocess(&gray, &profile).expect("preprocess");
         let expected = ((200.0 / 255.0) - 0.5) / 0.5;
-        let h = profile.target_height as usize;
-        let w = profile.target_width as usize;
+        let h = h_bar as usize;
+        let w = w_bar as usize;
         for c in 0..3 {
             let v = out[c * h * w];
             assert!(
@@ -252,4 +354,88 @@ mod tests {
             );
         }
     }
+
+    #[test]
+    fn smart_resize_keeps_factor_aligned_square_in_budget() {
+        // 448×448 sits inside [65536, 1048576] and is factor-aligned →
+        // unchanged. (Regression guard for the old fixed-res sweet spot.)
+        let (h, w) = smart_resize(448, 448, 32, 65_536, 1_048_576).unwrap();
+        assert_eq!((h, w), (448, 448));
+    }
+
+    #[test]
+    fn smart_resize_preserves_aspect_and_caps_at_max() {
+        // 3000×4000 (landscape) → downscaled under max_pixels, aspect kept.
+        let (h, w) = smart_resize(3000, 4000, 32, 65_536, 1_048_576).unwrap();
+        assert_eq!(h % 32, 0);
+        assert_eq!(w % 32, 0);
+        assert!(
+            (h as u64) * (w as u64) <= 1_048_576,
+            "must respect max_pixels"
+        );
+        assert!(w > h, "landscape orientation preserved");
+        // aspect ≈ 4000/3000 = 1.333; allow a factor-rounding tolerance.
+        let ar = w as f64 / h as f64;
+        assert!((ar - 4.0 / 3.0).abs() < 0.15, "aspect ~4:3, got {ar:.3}");
+    }
+
+    #[test]
+    fn smart_resize_floors_tiny_image_at_min() {
+        // 16×16 → upscaled to at least min_pixels, factor-aligned.
+        let (h, w) = smart_resize(16, 16, 32, 65_536, 1_048_576).unwrap();
+        assert_eq!(h % 32, 0);
+        assert_eq!(w % 32, 0);
+        assert!((h as u64) * (w as u64) >= 65_536, "must respect min_pixels");
+    }
+
+    #[test]
+    fn smart_resize_tall_nonsquare_stays_nonsquare() {
+        // A tall screenshot keeps portrait orientation.
+        let (h, w) = smart_resize(2000, 500, 32, 65_536, 1_048_576).unwrap();
+        assert!(h > w, "portrait orientation preserved");
+        assert_eq!(h % 32, 0);
+        assert_eq!(w % 32, 0);
+    }
+
+    #[test]
+    fn smart_resize_rejects_extreme_aspect() {
+        let err = smart_resize(1, 500, 32, 65_536, 1_048_576).unwrap_err();
+        assert!(format!("{err:#}").contains("200:1"));
+    }
+
+    #[test]
+    fn qwen3_6_never_exceeds_pos_embed_patch_budget() {
+        // The pos-embed cap must hold for huge, tall, wide, and extreme
+        // images — exceeding 2304 patches errors mid-tower and poisons
+        // the device context, so this invariant is load-bearing.
+        let p = PreprocessProfile::qwen3_6();
+        for (sh, sw) in [
+            (8000u32, 6000u32),
+            (808, 1600),
+            (4000, 400),
+            (1, 199),
+            (16, 16),
+        ] {
+            let (h, w) = p.resized_dims(sh, sw).unwrap();
+            let patches = (h / 16) * (w / 16);
+            assert!(
+                patches <= 2304,
+                "{sh}x{sw} → {h}x{w} = {patches} patches exceeds the 2304 budget"
+            );
+        }
+    }
+
+    #[test]
+    fn qwen3_6_default_budget_bounds_lm_tokens() {
+        // A huge source image caps at max_pixels → the per-image LM token
+        // count stays within budget (so it can't blow NEURON_MAX_PROMPT_TOKENS).
+        let p = PreprocessProfile::qwen3_6();
+        let (h, w) = p.resized_dims(8000, 6000).unwrap();
+        let lm_tokens = (h / p.factor) * (w / p.factor);
+        let budget = p.max_pixels / (p.factor * p.factor);
+        assert!(
+            lm_tokens <= budget,
+            "max-res image LM tokens {lm_tokens} must stay within budget {budget}"
+        );
+    }
 }

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

@@ -245,9 +245,67 @@ pub struct WorkerPool {
     /// Phase 4 the load itself moves onto the worker and that bridge
     /// goes away.
     pub(crate) leader_worker: std::sync::Arc<super::device_worker::DeviceWorkerHandle>,
+    /// Cached handle to the leader's NCCL `Comm`, fetched at `init_nccl`
+    /// while the worker thread is responsive. The TP step watchdog uses
+    /// it to `ncclCommAbort` a wedged collective from the async thread —
+    /// the one NCCL op allowed concurrently with an in-flight collective,
+    /// and the only way to unblock the in-process leader thread so
+    /// recovery's `unload` doesn't itself hang (#17 Stage 2). `None` if
+    /// init couldn't cache it; the watchdog then logs that it can't abort.
+    #[cfg(feature = "cuda")]
+    leader_comm: Option<nccl_state::SendComm>,
+}
+
+/// Per-step deadline for a TP forward (#17 Stage 2). A healthy decode
+/// step or chunked prefill completes in well under a second; a wedged
+/// NCCL collective never returns. Generous default so no legitimate step
+/// trips it; overridable via `NEURON_TP_STEP_TIMEOUT_S` (seconds).
+#[cfg(feature = "cuda")]
+fn tp_step_timeout() -> std::time::Duration {
+    let secs = std::env::var("NEURON_TP_STEP_TIMEOUT_S")
+        .ok()
+        .and_then(|v| v.trim().parse::<u64>().ok())
+        .filter(|&s| s > 0)
+        .unwrap_or(120);
+    std::time::Duration::from_secs(secs)
 }
 
 impl WorkerPool {
+    /// Abort the leader's NCCL comm to unblock a collective the watchdog
+    /// found wedged (#17 Stage 2). Logs the whole sequence loudly so a
+    /// real-world hang leaves a greppable forensic trail
+    /// (`tp watchdog:` / `ncclCommAbort`). Calling abort from this async
+    /// thread while the worker thread is blocked inside the collective is
+    /// the one concurrent NCCL op the library sanctions — it is how a
+    /// stuck/failed collective is unblocked.
+    #[cfg(feature = "cuda")]
+    fn watchdog_abort_leader_comm(&self, model_id: &str, secs: u64) {
+        tracing::error!(
+            model = %model_id,
+            timeout_s = secs,
+            "tp watchdog: leader forward exceeded deadline — NCCL collective wedged; \
+             aborting comm to unblock the leader thread for auto-recovery"
+        );
+        match &self.leader_comm {
+            Some(c) => match c.0.abort() {
+                Ok(()) => tracing::error!(
+                    model = %model_id,
+                    "tp watchdog: ncclCommAbort succeeded — wedged collective unblocked; \
+                     failing the step so the model auto-recovers (unload+reload)"
+                ),
+                Err(e) => tracing::error!(
+                    model = %model_id, error = ?e,
+                    "tp watchdog: ncclCommAbort failed — recovery may stall until a process restart"
+                ),
+            },
+            None => tracing::error!(
+                model = %model_id,
+                "tp watchdog: no cached leader comm handle — cannot abort; recovery will rely \
+                 on a process restart"
+            ),
+        }
+    }
+
     /// Spawn `world_size - 1` worker subprocesses. Rank 0 is the
     /// leader (in-process) and is *not* spawned here — the leader
     /// holds rank 0's NCCL Comm and shard in its own address space.
@@ -324,6 +382,8 @@ impl WorkerPool {
             workers,
             exe,
             leader_worker,
+            #[cfg(feature = "cuda")]
+            leader_comm: None,
         })
     }
 
@@ -404,6 +464,23 @@ impl WorkerPool {
             world_size = self.world_size,
             "NCCL communicator established across all ranks"
         );
+
+        // Cache the leader's Comm handle now, while the worker thread is
+        // responsive, so the TP step watchdog can abort a wedged
+        // collective later (it can't fetch it then — the thread is stuck).
+        // (#17 Stage 2.)
+        #[cfg(feature = "cuda")]
+        {
+            self.leader_comm = self.leader_worker.get_leader_comm().await;
+            if self.leader_comm.is_some() {
+                tracing::debug!("cached leader NCCL comm handle for the TP step watchdog");
+            } else {
+                tracing::warn!(
+                    "could not cache leader NCCL comm handle; the TP step watchdog will be \
+                     unable to abort a wedged collective (a hang would need a process restart)"
+                );
+            }
+        }
         Ok(())
     }
 
@@ -628,10 +705,27 @@ impl WorkerPool {
         //    that's the invariant the whole refactor exists to
         //    preserve.
         let leader_start = std::time::Instant::now();
-        let leader_result = self
+        let timeout = tp_step_timeout();
+        let leader_fut = self
             .leader_worker
-            .tp_forward_logits(leader_handle, tokens, offset)
-            .await;
+            .tp_forward_logits(leader_handle, tokens, offset);
+        let leader_result = match tokio::time::timeout(timeout, leader_fut).await {
+            Ok(r) => r,
+            Err(_elapsed) => {
+                // Watchdog (#17 Stage 2): the NCCL collective is wedged.
+                // Abort the leader comm to unblock its thread, then fail
+                // the step WITHOUT draining (the subprocess workers are
+                // wedged too; recovery's unload kills them). The error
+                // poisons the model → auto-recovery, which no longer hangs
+                // because the leader thread is now responsive.
+                self.watchdog_abort_leader_comm(model_id, timeout.as_secs());
+                anyhow::bail!(
+                    "tp watchdog: leader forward exceeded {}s deadline; aborted wedged NCCL \
+                     comm — model will auto-recover",
+                    timeout.as_secs()
+                );
+            }
+        };
         let leader_ok = leader_result.is_ok();
         let leader_ms = leader_start.elapsed().as_millis();
         // Surface the leader's own error at WARN before draining
@@ -767,17 +861,29 @@ impl WorkerPool {
         //    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,
-                chunk_size,
-            )
-            .await;
+        let timeout = tp_step_timeout();
+        let leader_fut = self.leader_worker.tp_forward_logits_with_images(
+            leader_handle,
+            tokens,
+            offset,
+            image_token_id,
+            image_data_uris,
+            chunk_size,
+        );
+        let leader_result = match tokio::time::timeout(timeout, leader_fut).await {
+            Ok(r) => r,
+            Err(_elapsed) => {
+                // Watchdog (#17 Stage 2) — see generate_step. Vision
+                // prefill is still well under the deadline on healthy
+                // hardware; a timeout means a wedged collective.
+                self.watchdog_abort_leader_comm(model_id, timeout.as_secs());
+                anyhow::bail!(
+                    "tp watchdog: leader image forward exceeded {}s deadline; aborted wedged \
+                     NCCL comm — model will auto-recover",
+                    timeout.as_secs()
+                );
+            }
+        };
         let leader_ok = leader_result.is_ok();
         let leader_ms = leader_start.elapsed().as_millis();
         if !leader_ok {

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

@@ -119,40 +119,25 @@ mod cuda_impl {
         }
     }
 
-    /// `Arc<Comm>` doesn't impl `Send` because `Comm` wraps a raw
-    /// `ncclComm_t` pointer. The NCCL contract is "operations against a
-    /// given comm must be serialised", not "the handle must stay on the
-    /// thread that created it" — so it's safe to move an `Arc<Comm>`
-    /// across threads as long as no concurrent ops are issued. The
-    /// pool's outer Mutex serialises us into `spawn_blocking`, so this
-    /// wrapper at the move boundary is the only thing missing.
+    /// Thin newtype over `Arc<Comm>`, kept for call-site clarity — it marks
+    /// the points where a comm handle is intentionally moved across threads
+    /// (e.g. cached async-side for the TP step watchdog's `ncclCommAbort`).
     ///
-    /// `Sync` is also marked safe because the `Arc<Comm>` clones held
-    /// by the row-parallel layers are only used from the
-    /// `spawn_blocking` thread driving the forward pass; concurrent
-    /// access from another thread would still be a bug.
+    /// `Send`/`Sync` are provided upstream by `cudarc`'s `Comm` (which
+    /// asserts the NCCL thread-safety invariant, including aborting from a
+    /// different thread than one inside a collective), so this type derives
+    /// them automatically — no manual `unsafe impl` here.
     pub struct SendComm(pub Arc<Comm>);
 
-    // SAFETY: see the doc-comment above; the invariant is enforced at
-    // the call site (pool Mutex + single spawn_blocking thread), not at
-    // the type level.
-    unsafe impl Send for SendComm {}
-    unsafe impl Sync for SendComm {}
-
     impl SendComm {
         pub fn into_inner(self) -> Arc<Comm> {
             self.0
         }
     }
 
-    // SAFETY: `cudarc::nccl::Comm` contains a raw `ncclComm_t` pointer
-    // (libnccl-allocated state). NCCL requires that operations against
-    // one Comm be issued one at a time; we serialise access by storing
-    // NcclState behind a Mutex in `WorkerPool`. The Comm itself is
-    // move-safe — NCCL doesn't track the calling OS thread, only the
-    // stream the operations are dispatched against.
-    unsafe impl Send for NcclState {}
-    unsafe impl Sync for NcclState {}
+    // `NcclState`'s `Send`/`Sync` are auto-derived: its `Arc<Comm>` and
+    // `Arc<CudaContext>` fields are now `Send`/`Sync` (cudarc asserts the
+    // comm thread-safety invariant), so no manual `unsafe impl` is needed.
 
     /// Generate a fresh NCCL `Id` and return it hex-encoded. Used by
     /// the leader to mint the shared communicator id which is then

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

@@ -1288,15 +1288,39 @@ impl TpQwen3_5ForCausalLM {
         let device = self.device().clone();
         let image_embeds = self.encode_images_concat(image_pixels)?;
 
+        // Each image's LM grid (lm_gh, lm_gw) = (h/factor, w/factor),
+        // factor = patch×merge. Recomputed per rank from this rank's own
+        // pixel tensors — deterministic, so every rank's grids (and hence
+        // M-RoPE positions) match without crossing the RPC (#14).
+        let factor = self
+            .vision
+            .as_ref()
+            .map(|v| {
+                let c = v.config();
+                c.patch_size * c.spatial_merge_size
+            })
+            .ok_or_else(|| {
+                candle_core::Error::Msg(
+                    "prefill_with_images_chunked: loaded without a vision tower".into(),
+                )
+            })?;
+        let grids: Vec<(usize, usize)> = image_pixels
+            .iter()
+            .map(|t| {
+                let (_, h, w) = t.dims3()?;
+                Ok::<(usize, usize), candle_core::Error>((h / factor, w / factor))
+            })
+            .collect::<candle_core::Result<Vec<_>>>()?;
+
         // Interleaved-M-RoPE 3D position ids for the whole prompt,
         // computed once and sliced per chunk so every rank assigns image
-        // tokens their 14×14 grid coordinates (and text after the image
-        // resumes from the compressed counter). `rope_delta` is stored on
-        // the base model for the decode that follows this prefill. Every
-        // chunk — text or image — uses the M-RoPE slice, because the image
-        // shifts the positions of the text around it.
+        // tokens their grid coordinates (and text after an image resumes
+        // from the compressed counter). `rope_delta` is stored on the base
+        // model for the decode that follows this prefill. Every chunk —
+        // text or image — uses the M-RoPE slice, because each image shifts
+        // the positions of the text around it.
         let (text, height, width, delta) =
-            crate::harness::arch::qwen3_5::rope::get_rope_index(tokens, image_token_id)
+            crate::harness::arch::qwen3_5::rope::get_rope_index(tokens, image_token_id, &grids)
                 .map_err(|e| candle_core::Error::Msg(format!("get_rope_index: {e}")))?;
         self.base.set_rope_delta(delta);
         let full_pos = crate::harness::arch::qwen3_5::rope::mrope_position_tensor(

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

@@ -494,16 +494,13 @@ impl WorkerState {
         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.
+        // embeddings — and thus the spliced hidden state and per-image
+        // grids — match across ranks. Native-aspect `smart_resize` (#14);
+        // deterministic, so each rank derives the same dims.
         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) {
+            let (px, h, w) = match preprocess_data_uri(uri, &profile) {
                 Ok(p) => p,
                 Err(e) => {
                     return WorkerResponse::Error {
@@ -512,7 +509,7 @@ impl WorkerState {
                     };
                 }
             };
-            match Tensor::from_vec(px, (3, h, w), &device) {
+            match Tensor::from_vec(px, (3, h as usize, w as usize), &device) {
                 Ok(t) => pixels.push(t),
                 Err(e) => {
                     return WorkerResponse::Error {