fix(discovery): stop a Qwen3-VL worker from serving video with another worker's contract

lib/llm/src/discovery/watcher.rs

@@ -59,6 +59,7 @@ use crate::{
         tensor::{NvCreateTensorRequest, NvCreateTensorResponse},
     },
     types::generic::realtime::{RealtimeClientEvent, RealtimeServerEvent},
+    utils::canonicalize_json,
     worker_type::WorkerType,
 };
 
@@ -1328,25 +1329,6 @@ fn lora_projection_fingerprint(card: &ModelDeploymentCard) -> anyhow::Result<Str
     Ok(blake3::hash(&serde_json::to_vec(&value)?).to_string())
 }
 
-fn canonicalize_json(value: &mut serde_json::Value) {
-    match value {
-        serde_json::Value::Object(object) => {
-            let mut entries = std::mem::take(object).into_iter().collect::<Vec<_>>();
-            entries.sort_by(|left, right| left.0.cmp(&right.0));
-            for (key, mut value) in entries {
-                canonicalize_json(&mut value);
-                object.insert(key, value);
-            }
-        }
-        serde_json::Value::Array(values) => {
-            for value in values {
-                canonicalize_json(value);
-            }
-        }
-        _ => {}
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

lib/llm/src/model_card.rs

@@ -1222,6 +1222,17 @@ impl ModelDeploymentCard {
                     bytes_to_hash.extend_from_slice(b"\0vllm_enable_tower_connector_lora\0true");
                 }
 
+                // Workers derive this contract from their own packages, so hash
+                // its content. Canonicalize: serde_json preserves key order.
+                if let Some(contract) = self.runtime_config.runtime_data.get(
+                    crate::local_model::runtime_config::VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY,
+                ) {
+                    let mut contract = contract.clone();
+                    crate::utils::canonicalize_json(&mut contract);
+                    bytes_to_hash.extend_from_slice(b"\0vllm_qwen_video_processor_contract\0");
+                    bytes_to_hash.extend(blake3::hash(contract.to_string().as_bytes()).as_bytes());
+                }
+
                 // TODO: Do we want any other user_data or runtime_config?
 
                 blake3::hash(&bytes_to_hash).to_string()
@@ -3200,6 +3211,75 @@ mod ownership_tests {
         assert_eq!(missing.mdcsum(), disabled.mdcsum());
         assert_ne!(missing.mdcsum(), enabled.mdcsum());
     }
+
+    /// The group builds one video-routing processor from a single
+    /// representative card, so every member of a WorkerSet is served with the
+    /// representative's Qwen video contract. Cards whose contract differs -
+    /// including a card that withholds it because engine-level
+    /// `--mm-processor-kwargs` may change the video token layout - must
+    /// therefore hash apart.
+    ///
+    /// Note: `mdcsum()` caches its result on first call via `OnceLock`, so each
+    /// case builds a fresh card rather than mutating one and re-hashing.
+    #[test]
+    fn qwen_video_processor_contract_isolates_worker_sets() {
+        use crate::local_model::runtime_config::VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY;
+
+        fn card_with_contract(contract: serde_json::Value) -> ModelDeploymentCard {
+            let mut card = ModelDeploymentCard::with_name_only("model");
+            card.runtime_config.runtime_data.insert(
+                VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY.to_string(),
+                contract,
+            );
+            card
+        }
+
+        fn legacy_ceil() -> serde_json::Value {
+            serde_json::json!({
+                "placeholder_target": "bare_video_token",
+                "resize_mode": "legacy_ceil",
+            })
+        }
+
+        let withheld = ModelDeploymentCard::with_name_only("model");
+        assert_ne!(
+            withheld.mdcsum(),
+            card_with_contract(legacy_ceil()).mdcsum(),
+            "a worker that never published the contract must not join the WorkerSet of one that did"
+        );
+
+        assert_ne!(
+            card_with_contract(legacy_ceil()).mdcsum(),
+            card_with_contract(serde_json::json!({
+                "placeholder_target": "bare_video_token",
+                "resize_mode": "smart_resize",
+            }))
+            .mdcsum(),
+            "two workers publishing different contracts must not share a WorkerSet"
+        );
+
+        // serde_json preserves key order here, so the same contract written in
+        // a different key order must still hash the same.
+        assert_eq!(
+            card_with_contract(legacy_ceil()).mdcsum(),
+            card_with_contract(serde_json::json!({
+                "resize_mode": "legacy_ceil",
+                "placeholder_target": "bare_video_token",
+            }))
+            .mdcsum(),
+            "key order within the contract must not change the checksum"
+        );
+
+        // Negative control: a deployment that never carries this key is
+        // unaffected, so its cache directory and reported checksum do not move.
+        let bare = ModelDeploymentCard::with_name_only("model");
+        let mut unrelated = ModelDeploymentCard::with_name_only("model");
+        unrelated
+            .runtime_config
+            .runtime_data
+            .insert("some_unrelated_runtime_key".to_string(), true.into());
+        assert_eq!(bare.mdcsum(), unrelated.mdcsum());
+    }
 }
 
 #[cfg(test)]

lib/llm/src/utils/mod.rs

@@ -5,3 +5,28 @@ pub mod lora;
 pub mod zmq;
 
 pub use lora::lora_name_to_id;
+
+/// Sort every JSON object in `value` by key, recursively.
+///
+/// `serde_json` runs with `preserve_order` in this workspace, so an object
+/// keeps the key order it was parsed in. Anything that hashes JSON as an
+/// identity has to canonicalize first, or two logically identical values
+/// hash apart.
+pub(crate) fn canonicalize_json(value: &mut serde_json::Value) {
+    match value {
+        serde_json::Value::Object(object) => {
+            let mut entries = std::mem::take(object).into_iter().collect::<Vec<_>>();
+            entries.sort_by(|left, right| left.0.cmp(&right.0));
+            for (key, mut value) in entries {
+                canonicalize_json(&mut value);
+                object.insert(key, value);
+            }
+        }
+        serde_json::Value::Array(values) => {
+            for value in values {
+                canonicalize_json(value);
+            }
+        }
+        _ => {}
+    }
+}