feat(skippy): expose a model-bound tokenizer facade

.github/workflows/docker-precheck.yml

@@ -96,6 +96,7 @@ jobs:
             crates/model-artifact/ \
             crates/model-hf/ \
             crates/skippy-protocol/ \
+            crates/skippy-tokenizer/ \
             crates/skippy-topology/ \
             crates/skippy-ffi/ \
             crates/skippy-runtime/ \

Cargo.toml

@@ -39,6 +39,7 @@ members = [
     "crates/model-artifact",
     "crates/model-hf",
     "crates/model-resolver",
+    "crates/skippy-tokenizer",
     "crates/skippy-protocol",
     "crates/skippy-coordinator",
     "crates/skippy-topology",

crates/mesh-llm-host-runtime/src/runtime/local_split/tests.rs

@@ -1018,8 +1018,26 @@ fn split_recovery_candidate_participants_excludes_unavailable_stage_nodes() {
     );
 }
 
-#[tokio::test]
-async fn load_split_runtime_generation_stops_candidate_stages_after_partial_load_failure() {
+#[test]
+fn load_split_runtime_generation_stops_candidate_stages_after_partial_load_failure() {
+    std::thread::Builder::new()
+        .name("local-split-test".to_owned())
+        .stack_size(8 * 1024 * 1024)
+        .spawn(|| {
+            tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .expect("build local split test runtime")
+                .block_on(
+                    load_split_runtime_generation_stops_candidate_stages_after_partial_load_failure_inner(),
+                );
+        })
+        .expect("spawn local split test thread")
+        .join()
+        .expect("local split test thread panicked");
+}
+
+async fn load_split_runtime_generation_stops_candidate_stages_after_partial_load_failure_inner() {
     let node = mesh::Node::new_for_tests(NodeRole::Host { http_port: 9337 })
         .await
         .unwrap();

crates/skippy-protocol/Cargo.toml

@@ -10,6 +10,7 @@ homepage = "https://github.com/Mesh-LLM/mesh-llm"
 [dependencies]
 prost = "0.14"
 serde.workspace = true
+skippy-tokenizer = { path = "../skippy-tokenizer", version = "0.72.1" }
 
 [dev-dependencies]
 serde_json.workspace = true

crates/skippy-protocol/src/lib.rs

@@ -1,6 +1,13 @@
 use serde::{Deserialize, Serialize};
 
-pub mod tokenizer;
+/// Compatibility namespace for tokenizer contracts.
+///
+/// New consumers should depend on `skippy-tokenizer` directly. Keeping this
+/// re-export avoids breaking older protocol users while the contract moves
+/// out of the wire-protocol crate.
+pub mod tokenizer {
+    pub use skippy_tokenizer::*;
+}
 
 pub mod binary;
 pub mod proto {

crates/skippy-runtime/src/native.rs

@@ -325,6 +325,21 @@ impl StageModel {
     }
 
     pub fn tokenize(&self, text: &str, add_special: bool) -> Result<Vec<i32>> {
+        self.tokenize_bounded(text, add_special, usize::MAX)?
+            .ok_or_else(|| anyhow!("tokenizer output exceeds the requested limit"))
+    }
+
+    /// Tokenize without allocating a token buffer larger than `max_tokens`.
+    ///
+    /// The native ABI reports the required count during its sizing call. When
+    /// that count exceeds the bound, this returns `Ok(None)` before allocating
+    /// the output vector.
+    pub fn tokenize_bounded(
+        &self,
+        text: &str,
+        add_special: bool,
+        max_tokens: usize,
+    ) -> Result<Option<Vec<i32>>> {
         let text = CString::new(text).context("text contains an interior NUL byte")?;
         let mut count = 0usize;
         let mut error = ptr::null_mut();
@@ -345,6 +360,10 @@ impl StageModel {
             free_error(error);
         }
 
+        if count > max_tokens {
+            return Ok(None);
+        }
+
         let mut tokens = vec![0_i32; count];
         let mut error = ptr::null_mut();
         let status = unsafe {
@@ -358,9 +377,13 @@ impl StageModel {
                 &mut error,
             )
         };
+        if status == Status::BufferTooSmall {
+            free_error(error);
+            return Ok(None);
+        }
         ensure_ok(status, error)?;
         tokens.truncate(count);
-        Ok(tokens)
+        Ok(Some(tokens))
     }
 
     pub fn detokenize(&self, tokens: &[i32]) -> Result<String> {

crates/skippy-server/Cargo.toml

@@ -25,6 +25,7 @@ blake3.workspace = true
 clap.workspace = true
 futures-util = "0.3"
 skippy-runtime = { path = "../skippy-runtime", version = "0.72.1"  }
+skippy-tokenizer = { path = "../skippy-tokenizer", version = "0.72.1" }
 model-artifact = { path = "../model-artifact", version = "0.72.1" }
 mesh-native-serving-plugin-api = { path = "../mesh-native-serving-plugin-api", version = "0.72.1" }
 skippy-protocol = { path = "../skippy-protocol", version = "0.72.1"  }

crates/skippy-server/README.md

@@ -77,6 +77,20 @@ topology, builds the stage configs, loads/starts handles, watches readiness and
 status, withdraws routes before shutdown, and then calls handle shutdown during
 unload or replan.
 
+### In-process tokenizer capability
+
+`SkippyRuntimeHandle::tokenizer_capability()` returns a model-bound
+`skippy_tokenizer::Tokenizer` backed by the already-loaded stage-zero runtime.
+Consumers can call `tokenize_batch` for bounded, ordered results without an HTTP
+round trip or a second model load. Every request supplies the expected
+`TokenizerIdentity`; mismatches are returned as per-item errors. The identity
+includes the model, source digest, tokenizer id, stage, and serving profile.
+
+The `/v1/tokenize` route is retained only as an explicit compatibility and
+out-of-band adapter. It accepts the legacy `add_special` field as well as the
+facade's `special_tokens` policy and is not part of generation or proposal
+deadline handling.
+
 ## Notes
 
 - `serve-binary` is the tuned binary stage-to-stage path.

crates/skippy-server/src/embedded.rs

@@ -1,6 +1,7 @@
 use std::{
     net::SocketAddr,
-    sync::{Arc, Mutex, TryLockError},
+    sync::atomic::{AtomicBool, Ordering},
+    sync::{Arc, Mutex, OnceLock, TryLockError},
 };
 
 use anyhow::{Context, Result};
@@ -84,6 +85,8 @@ pub struct SkippyRuntimeHandle {
     runtime: Arc<Mutex<RuntimeState>>,
     telemetry: Telemetry,
     status: Arc<Mutex<RuntimeHandleState>>,
+    tokenizer_active: Arc<AtomicBool>,
+    tokenizer_capability: OnceLock<Result<TokenizerCapability, TokenizerCapabilityError>>,
     /// Last session stats read out of [`Self::runtime`], and when.
     ///
     /// A native call (long prefill, decode batch) holds the runtime lock while
@@ -140,6 +143,8 @@ impl SkippyRuntimeHandle {
                 stopped_at_unix_nanos: None,
                 last_error: None,
             })),
+            tokenizer_active: Arc::new(AtomicBool::new(true)),
+            tokenizer_capability: OnceLock::new(),
             last_session_stats: Arc::new(Mutex::new(initial_session_stats)),
         }
     }
@@ -241,7 +246,15 @@ impl SkippyRuntimeHandle {
     /// Returns the stateless tokenizer capability backed by this already-loaded
     /// stage-zero runtime. This never opens a second model.
     pub fn tokenizer_capability(&self) -> Result<TokenizerCapability, TokenizerCapabilityError> {
-        TokenizerCapability::from_stage_zero(&self.config, self.runtime.clone())
+        self.tokenizer_capability
+            .get_or_init(|| {
+                TokenizerCapability::from_stage_zero_with_lifecycle(
+                    &self.config,
+                    self.runtime.clone(),
+                    self.tokenizer_active.clone(),
+                )
+            })
+            .clone()
     }
 
     pub fn status(&self) -> EmbeddedRuntimeStatus {
@@ -270,6 +283,9 @@ impl SkippyRuntimeHandle {
     }
 
     pub fn shutdown(&self) {
+        self.tokenizer_active.store(false, Ordering::Release);
+        let runtime = self.runtime.lock().expect("runtime lock poisoned");
+        drop(runtime);
         let mut status = self.status.lock().expect("runtime status lock poisoned");
         if status.state == EmbeddedState::Stopped {
             return;
@@ -689,6 +705,28 @@ mod tests {
         assert!(status.runtime_loaded);
     }
 
+    #[test]
+    fn shutdown_waits_for_runtime_lock_after_invalidating_tokenizer() {
+        let handle = Arc::new(test_handle(1));
+        let held = handle.runtime.lock().expect("runtime lock");
+        let (shutdown_tx, shutdown_rx) = mpsc::channel();
+        let shutdown_handle = Arc::clone(&handle);
+        thread::spawn(move || {
+            shutdown_handle.shutdown();
+            shutdown_tx.send(()).expect("send shutdown result");
+        });
+
+        assert!(
+            shutdown_rx.recv_timeout(Duration::from_millis(50)).is_err(),
+            "shutdown should synchronize with an in-flight runtime operation"
+        );
+        drop(held);
+        shutdown_rx
+            .recv_timeout(Duration::from_secs(1))
+            .expect("shutdown should complete after the runtime lock is released");
+        assert_eq!(handle.status().state, EmbeddedState::Stopped);
+    }
+
     fn empty_cache(value: u32) -> Mutex<Captured<u32>> {
         Mutex::new(Captured {
             value,