feat(self-host): frontend MDC verify-and-cache pipeline (gh-8749)

components/src/dynamo/frontend/sglang_processor.py

@@ -622,6 +622,9 @@ async def chat_engine_factory(
             )
         loop = asyncio.get_running_loop()
 
+        # TODO(gh-8749): consume mdc.model_info.path()'s parent (slug_dir)
+        # instead of re-running fetch_model. The MDC cache already has
+        # blake3-verified copies; this path duplicates the download.
         source_path = mdc.source_path()
         if not os.path.exists(source_path):
             await fetch_model(source_path, ignore_weights=True)

components/src/dynamo/frontend/vllm_processor.py

@@ -760,6 +760,9 @@ async def chat_engine_factory(
             )
         loop = asyncio.get_running_loop()
 
+        # TODO(gh-8749): consume mdc.model_info.path()'s parent (slug_dir)
+        # instead of re-running fetch_model. The MDC cache already has
+        # blake3-verified copies; this path duplicates the download.
         source_path = mdc.source_path()
         if not os.path.exists(source_path):
             await fetch_model(source_path, ignore_weights=True)

lib/bindings/c/src/lib.rs

@@ -1498,7 +1498,7 @@ async fn fetch_preprocessor_from_discovery(
     );
 
     // Download config (tokenizer files) if not local
-    card.download_config().await?;
+    card.download_config(None).await?;
 
     // Create preprocessor
     let preprocessor = OpenAIPreprocessor::new(card.clone())?;

lib/llm/Cargo.toml

@@ -77,6 +77,7 @@ futures = { workspace = true }
 futures-util = { workspace = true }
 hf-hub = { workspace = true }
 ipnet = { workspace = true }
+libc = { workspace = true }
 lru = { workspace = true }
 rand = { workspace = true }
 oneshot = { workspace = true }

lib/llm/src/common/checked_file.rs

@@ -235,6 +235,27 @@ impl Checksum {
             algorithm,
         }
     }
+
+    pub fn hash(&self) -> &str {
+        &self.hash
+    }
+}
+
+/// Worker-controlled wire field used as a path component downstream;
+/// reject anything that doesn't look like the algorithm's canonical
+/// digest. Called from both `Deserialize` and `TryFrom<&str>` so both
+/// wire entry points stay in lockstep.
+fn validate_checksum_hash(algorithm: CryptographicHashMethods, hash: &str) -> Result<(), String> {
+    match algorithm {
+        CryptographicHashMethods::BLAKE3 => {
+            if hash.len() != 64 || !hash.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')) {
+                return Err(format!(
+                    "expected 64-char lowercase hex blake3 digest, got {hash:?}"
+                ));
+            }
+        }
+    }
+    Ok(())
 }
 
 impl Serialize for Checksum {
@@ -273,6 +294,9 @@ impl<'de> Deserialize<'de> for Checksum {
                 let algorithm = parts[0].parse().map_err(|_| {
                     de::Error::invalid_value(de::Unexpected::Str(parts[0]), &"invalid algorithm")
                 })?;
+                validate_checksum_hash(algorithm, parts[1]).map_err(|msg| {
+                    de::Error::invalid_value(de::Unexpected::Str(parts[1]), &msg.as_str())
+                })?;
 
                 Ok(Checksum::new(parts[1], algorithm))
             }
@@ -297,6 +321,7 @@ impl TryFrom<&str> for Checksum {
                 anyhow::bail!("Unsupported cryptographic hash method: {}", parts[0]);
             }
         };
+        validate_checksum_hash(algo, parts[1]).map_err(|msg| anyhow::anyhow!("{msg}"))?;
 
         Ok(Checksum::new(parts[1], algo))
     }
@@ -348,26 +373,32 @@ mod tests {
         assert_eq!(serialized.trim(), "\"blake3:a12c3d4\"");
     }
 
+    const HEX64: &str = "62bc124be974d3a25db05bedc99422660c26715e5bbda0b37d14bd84a0c65ab2";
+
     #[test]
     fn test_deserialization_blake3() {
-        let s = "\"blake3:abcd1234\"";
-        let deserialized: Checksum = serde_json::from_str(s).unwrap();
-
+        let s = format!("\"blake3:{HEX64}\"");
+        let deserialized: Checksum = serde_json::from_str(&s).unwrap();
         assert_eq!(deserialized.algorithm, CryptographicHashMethods::BLAKE3);
-        assert_eq!(deserialized.hash, "abcd1234");
+        assert_eq!(deserialized.hash, HEX64);
     }
 
     #[test]
     fn test_deserialization_invalid_format() {
-        let s = "\"invalidformat\"";
-        let result: Result<Checksum, _> = serde_json::from_str(s);
-
-        assert!(result.is_err());
-
-        let s = "\"blake3:invalid:format\"";
-        let result: Result<Checksum, _> = serde_json::from_str(s);
-
-        assert!(result.is_err());
+        for bad in [
+            "\"invalidformat\"",
+            "\"blake3:invalid:format\"",
+            // Worker-controlled wire field shouldn't survive as a path
+            // component; reject anything that's not a 64-char lowercase
+            // hex blake3 digest.
+            "\"blake3:abcd1234\"",         // too short
+            "\"blake3:../../etc/passwd\"", // path traversal
+            "\"blake3:62BC124BE974D3A25DB05BEDC99422660C26715E5BBDA0B37D14BD84A0C65AB2\"", // uppercase
+            &format!("\"blake3:{HEX64}x\""), // 65 chars
+        ] {
+            let result: Result<Checksum, _> = serde_json::from_str(bad);
+            assert!(result.is_err(), "expected reject: {bad}");
+        }
     }
 
     #[test]
@@ -378,5 +409,38 @@ mod tests {
         let expected =
             Checksum::blake3("62bc124be974d3a25db05bedc99422660c26715e5bbda0b37d14bd84a0c65ab2");
         assert_eq!(expected, *cf.checksum());
+        assert_eq!(Some(560), cf.size(), "TinyLlama config.json is 560 bytes");
+        assert_eq!(
+            "62bc124be974d3a25db05bedc99422660c26715e5bbda0b37d14bd84a0c65ab2",
+            cf.checksum().hash()
+        );
+    }
+
+    /// Old MDC bytes (no `size` field) deserialize cleanly with
+    /// `size = None`, preserving cross-version compat between a
+    /// pre-size worker and a post-size frontend.
+    #[test]
+    fn test_legacy_wire_format_omits_size() {
+        let legacy = format!(r#"{{"path":"/x/config.json","checksum":"blake3:{HEX64}"}}"#);
+        let cf: CheckedFile = serde_json::from_str(&legacy).unwrap();
+        assert_eq!(None, cf.size());
+        assert_eq!("/x/config.json", cf.path().unwrap().to_str().unwrap());
+    }
+
+    /// New-format CheckedFiles round-trip with the size field.
+    #[test]
+    fn test_new_wire_format_carries_size() {
+        let cf = CheckedFile {
+            path: Either::Left(PathBuf::from("/x/tokenizer.json")),
+            checksum: Checksum::blake3(HEX64),
+            size: Some(12345),
+        };
+        let json = serde_json::to_string(&cf).unwrap();
+        assert!(
+            json.contains("\"size\":12345"),
+            "expected size field in serialization, got: {json}"
+        );
+        let back: CheckedFile = serde_json::from_str(&json).unwrap();
+        assert_eq!(Some(12345), back.size());
     }
 }

lib/llm/src/discovery/watcher.rs

@@ -1,6 +1,7 @@
 // SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::Notify;
@@ -88,6 +89,10 @@ pub struct ModelWatcher {
     /// Tracks in-flight `handle_put` tasks by instance path so that `handle_delete`
     /// can await a racing put before proceeding with cleanup.
     pending_puts: DashMap<String, JoinHandle<()>>,
+    /// Frontend's `--model-path`. Threaded into `download_config` so
+    /// `file://` slots can fall back here when the worker's path is
+    /// unreachable on this host.
+    local_model_path: Option<PathBuf>,
 }
 
 const ALL_MODEL_TYPES: &[ModelType] = &[
@@ -167,13 +172,18 @@ impl ModelWatcher {
             registering_worker_sets: DashSet::new(),
             registration_notify: Notify::new(),
             pending_puts: DashMap::new(),
+            local_model_path: None,
         }
     }
 
     pub fn set_notify_on_model_update(&mut self, tx: Sender<ModelUpdate>) {
         self.model_update_tx = Some(tx);
     }
 
+    pub fn set_local_model_path(&mut self, path: Option<PathBuf>) {
+        self.local_model_path = path;
+    }
+
     /// Wait until we have at least one chat completions model and return it's name.
     pub async fn wait_for_chat_model(&self) -> String {
         // Loop in case it gets added and immediately deleted
@@ -673,7 +683,8 @@ impl ModelWatcher {
         mcid: &ModelCardInstanceId,
         card: &mut ModelDeploymentCard,
     ) -> anyhow::Result<()> {
-        card.download_config().await?;
+        card.download_config(self.local_model_path.as_deref())
+            .await?;
 
         // Use per-worker-set router config if the worker provided one in its MDC,
         // otherwise fall back to the frontend-level global config.

lib/llm/src/entrypoint/input/common.rs

@@ -101,7 +101,7 @@ pub async fn prepare_engine(
             let model_manager = Arc::new(ModelManager::new());
             // Create metrics for migration tracking (not exposed via /metrics in Dynamic engine mode)
             let metrics = Arc::new(Metrics::new());
-            let watch_obj = Arc::new(ModelWatcher::new(
+            let mut watcher = ModelWatcher::new(
                 distributed_runtime.clone(),
                 model_manager.clone(),
                 RouterConfig::default(),
@@ -110,7 +110,11 @@ pub async fn prepare_engine(
                 None,
                 prefill_load_estimator,
                 metrics,
-            ));
+            );
+            if !local_model.path().as_os_str().is_empty() {
+                watcher.set_local_model_path(Some(local_model.path().to_path_buf()));
+            }
+            let watch_obj = Arc::new(watcher);
             let discovery = distributed_runtime.discovery();
             let discovery_stream = discovery
                 .list_and_watch(

lib/llm/src/entrypoint/input/grpc.rs

@@ -1,6 +1,7 @@
 // SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+use std::path::PathBuf;
 use std::sync::Arc;
 
 use crate::{
@@ -47,6 +48,8 @@ pub async fn run(
                 model.namespace(),
                 model.namespace_prefix(),
             );
+            let local_model_path =
+                (!model.path().as_os_str().is_empty()).then(|| model.path().to_path_buf());
             run_watcher(
                 distributed_runtime.clone(),
                 grpc_service.state().manager_clone(),
@@ -55,6 +58,7 @@ pub async fn run(
                 migration_max_seq_len,
                 namespace_filter,
                 prefill_load_estimator.clone(),
+                local_model_path,
             )
             .await?;
             grpc_service
@@ -112,6 +116,7 @@ pub async fn run(
 
 /// Spawns a task that watches for new models in store,
 /// and registers them with the ModelManager so that the HTTP service can use them.
+#[allow(clippy::too_many_arguments)]
 async fn run_watcher(
     runtime: DistributedRuntime,
     model_manager: Arc<ModelManager>,
@@ -120,10 +125,11 @@ async fn run_watcher(
     migration_max_seq_len: Option<u32>,
     namespace_filter: NamespaceFilter,
     prefill_load_estimator: Option<Arc<dyn dynamo_kv_router::PrefillLoadEstimator>>,
+    local_model_path: Option<PathBuf>,
 ) -> anyhow::Result<()> {
     // Create metrics for migration tracking (not exposed via /metrics in gRPC mode)
     let metrics = Arc::new(Metrics::new());
-    let watch_obj = ModelWatcher::new(
+    let mut watch_obj = ModelWatcher::new(
         runtime.clone(),
         model_manager,
         router_config,
@@ -133,6 +139,7 @@ async fn run_watcher(
         prefill_load_estimator,
         metrics,
     );
+    watch_obj.set_local_model_path(local_model_path);
     tracing::debug!("Waiting for remote model");
     let discovery = runtime.discovery();
     let discovery_stream = discovery

lib/llm/src/entrypoint/input/http.rs

@@ -1,6 +1,7 @@
 // SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+use std::path::PathBuf;
 use std::sync::Arc;
 
 use crate::{
@@ -83,6 +84,8 @@ pub async fn run(
                 model.namespace(),
                 model.namespace_prefix(),
             );
+            let local_model_path =
+                (!model.path().as_os_str().is_empty()).then(|| model.path().to_path_buf());
             run_watcher(
                 distributed_runtime.clone(),
                 http_service.state().manager_clone(),
@@ -94,6 +97,7 @@ pub async fn run(
                 http_service.state().metrics_clone(),
                 chat_engine_factory.clone(),
                 prefill_load_estimator.clone(),
+                local_model_path,
             )
             .await?;
             http_service
@@ -173,6 +177,7 @@ async fn run_watcher(
     metrics: Arc<crate::http::service::metrics::Metrics>,
     chat_engine_factory: Option<ChatEngineFactoryCallback>,
     prefill_load_estimator: Option<Arc<dyn dynamo_kv_router::PrefillLoadEstimator>>,
+    local_model_path: Option<PathBuf>,
 ) -> anyhow::Result<()> {
     let mut watch_obj = ModelWatcher::new(
         runtime.clone(),
@@ -184,6 +189,7 @@ async fn run_watcher(
         prefill_load_estimator,
         metrics.clone(),
     );
+    watch_obj.set_local_model_path(local_model_path);
     tracing::debug!("Waiting for remote model");
     let discovery = runtime.discovery();
     let discovery_stream = discovery

lib/llm/src/local_model.rs

@@ -564,24 +564,20 @@ impl LocalModel {
             let Some(local_path) = cf.path().map(Path::to_path_buf) else {
                 continue;
             };
-            // Filename from the original path — HF cache symlinks would
-            // otherwise canonicalize to the LFS SHA1 and break downstream
-            // lookups by literal name (e.g. `parent.join("generation_config.json")`).
             let Some(filename) = local_path
                 .file_name()
                 .and_then(|f| f.to_str())
                 .map(|s| s.to_string())
             else {
                 continue;
             };
-            // Canonicalize so the handler serves the resolved HF blob, not the symlink.
-            let absolute = match fs::canonicalize(&local_path) {
+            let absolute = match std::path::absolute(&local_path) {
                 Ok(p) => p,
                 Err(err) => {
                     tracing::warn!(
                         path = %local_path.display(),
                         %err,
-                        "failed to canonicalize self-host metadata path; skipping",
+                        "failed to absolutize self-host metadata path; skipping",
                     );
                     continue;
                 }