feat(mesh): expose model-bound tokenizer capability

crates/mesh-llm-host-runtime/src/inference/skippy/resolver/native_mtp_tests.rs

@@ -1,12 +1,91 @@
+use super::super::{SkippyModelHandle, SkippyModelLoadOptions};
 use super::test_support::*;
 use super::*;
 use crate::inference::skippy::SkippyTelemetryOptions;
+use anyhow::Result;
+use openai_frontend::OpenAiBackend;
 use skippy_protocol::LoadMode;
 use skippy_runtime::package::{
     PackageExtensionPolicyInfo, PackageGenerationInfo, PackageSpeculativeDecodingInfo,
     PackageSpeculativeProposerInfo, PackageSpeculativeStrategyInfo, PackageWindowPolicyInfo,
 };
-use std::collections::BTreeMap;
+use std::{
+    collections::BTreeMap,
+    path::Path,
+    sync::{
+        Arc, Mutex,
+        atomic::{AtomicUsize, Ordering},
+    },
+    time::Duration,
+};
+
+#[derive(Default)]
+struct RecordingNativeProposalIngress {
+    proposals: AtomicUsize,
+    reports: Mutex<Vec<skippy_server::LinearProposalReceipt>>,
+}
+
+impl skippy_server::LinearProposalIngress for RecordingNativeProposalIngress {
+    fn propose(
+        &self,
+        _query: skippy_server::LinearProposalQuery,
+    ) -> anyhow::Result<skippy_server::LinearProposalSourceResponse> {
+        self.proposals.fetch_add(1, Ordering::Relaxed);
+        let decision_id = skippy_server::OpaqueProposalDecisionId::new(vec![1])?;
+        Ok(skippy_server::LinearProposalSourceResponse::new(Some(
+            skippy_server::LinearProposal::new(decision_id, vec![0]),
+        )))
+    }
+
+    fn report(&self, receipt: &skippy_server::LinearProposalReceipt) -> anyhow::Result<()> {
+        self.reports.lock().unwrap().push(receipt.clone());
+        Ok(())
+    }
+}
+
+struct NoopGenerationReceiptSink;
+
+impl skippy_server::frontend::GenerationReceiptSink for NoopGenerationReceiptSink {
+    fn begin(&self, _start: &skippy_server::frontend::GenerationStart) -> Result<()> {
+        Ok(())
+    }
+
+    fn committed(&self, _commit: &skippy_server::frontend::GenerationCommit) -> Result<()> {
+        Ok(())
+    }
+
+    fn abort(&self, _abort: &skippy_server::frontend::GenerationAbort) -> Result<()> {
+        Ok(())
+    }
+
+    fn record(&self, _receipt: &skippy_server::frontend::GenerationReceipt) -> Result<()> {
+        Ok(())
+    }
+}
+
+struct RecordingNativeHooksFactory {
+    ingress: Arc<RecordingNativeProposalIngress>,
+}
+
+impl skippy_server::serving_hooks::ModelServingHooksFactory for RecordingNativeHooksFactory {
+    fn create(
+        &self,
+        _tokenizer: skippy_server::TokenizerCapability,
+    ) -> Result<skippy_server::serving_hooks::ModelServingHooks> {
+        let source: Arc<dyn skippy_server::LinearProposalIngress> = self.ingress.clone();
+        let ingress = skippy_server::frontend::LinearProposalIngressConfig::new(
+            source,
+            Duration::from_millis(25),
+            1,
+        )?;
+        Ok(skippy_server::serving_hooks::ModelServingHooks::new(
+            skippy_server::frontend::GenerationReceiptConfig::new(Arc::new(
+                NoopGenerationReceiptSink,
+            )),
+            ingress,
+        ))
+    }
+}
 
 fn native_mtp_generation() -> PackageGenerationInfo {
     let mut proposers = BTreeMap::new();
@@ -761,7 +840,7 @@ strategy = "ngram-suffix"
 }
 
 #[test]
-fn standalone_ngram_rejected_for_single_stage_serving() {
+fn standalone_ngram_uses_native_plugin_verify_path_for_single_stage_serving() {
     let mesh_config = parse_config(
         r#"
 [defaults.speculative]
@@ -789,17 +868,102 @@ ngram_max_proposal_tokens = 48
         .to_embedded_openai_args(4096, true)
         .expect("staged serving should build OpenAI args");
 
-    // Single-stage/direct serving has no N-gram verify path, so it must reject
-    // rather than silently run target-only while reporting a proposer.
-    let error = resolved
+    // Single-stage/direct serving uses the native plugin ingress, whose
+    // proposal is verified and repaired by the local target runtime.
+    let openai = resolved
         .to_embedded_openai_args(0, false)
-        .expect_err("single-stage standalone N-gram must be rejected");
+        .expect("single-stage standalone N-gram should use the native plugin verify path");
+    assert_eq!(openai.speculative_window, 48);
+    assert_eq!(
+        openai.speculative.ngram.as_ref().map(|ngram| ngram.kind),
+        Some(skippy_server::NgramProposerKind::Suffix)
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn direct_single_stage_serving_delivers_target_authoritative_native_receipts() -> Result<()> {
+    let Some(model_path) = std::env::var_os("SKIPPY_NATIVE_PLUGIN_MODEL") else {
+        eprintln!("skipping: SKIPPY_NATIVE_PLUGIN_MODEL is not set");
+        return Ok(());
+    };
+    let model_path = Path::new(&model_path);
+    let model_bytes = std::fs::metadata(model_path)?.len();
+    let model_id = "Qwen/Qwen3-0.6B:Q4_K_M";
+    let mesh_config = parse_config(
+        r#"
+[defaults.speculative]
+strategy = "ngram-suffix"
+ngram_min = 2
+ngram_max = 8
+ngram_max_proposal_tokens = 1
+"#,
+    );
+    let resolved = resolve_skippy_config(SkippyConfigResolveRequest {
+        mesh_config: &mesh_config,
+        model_id,
+        model_path,
+        model_bytes,
+        allocatable_memory_bytes: None,
+        request_defaults: None,
+        package_generation: None,
+    })?;
+    let embedded_openai = resolved.to_embedded_openai_args(0, false)?;
+    let ingress = Arc::new(RecordingNativeProposalIngress::default());
+    let factory: skippy_server::serving_hooks::SharedModelServingHooksFactory =
+        Arc::new(RecordingNativeHooksFactory {
+            ingress: Arc::clone(&ingress),
+        });
+    let mut options = SkippyModelLoadOptions::for_direct_gguf(model_id, model_path)
+        .with_ctx_size(256)
+        .with_embedded_openai(embedded_openai)
+        .with_serving_hooks_factory(Some(factory));
+    options.n_gpu_layers = 0;
+
+    let handle = SkippyModelHandle::load_with_hooks(
+        options,
+        None,
+        crate::runtime::survey::SurveyTelemetry::disabled(),
+    )?;
+    let request = serde_json::from_value(serde_json::json!({
+        "model": model_id,
+        "messages": [{"role": "user", "content": "Say hello."}],
+        "max_tokens": 2,
+        "temperature": 0.0
+    }))?;
+    let response = handle.chat_completion(request).await;
+    handle.shutdown();
+    response?;
+
+    for _ in 0..100 {
+        if !ingress.reports.lock().unwrap().is_empty() {
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+    assert!(ingress.proposals.load(Ordering::Relaxed) > 0);
+    let reports = ingress.reports.lock().unwrap();
     assert!(
-        error
-            .to_string()
-            .contains("requires multi-stage split serving"),
-        "{error}"
+        !reports.is_empty(),
+        "native proposal report was not delivered"
     );
+    for receipt in reports.iter() {
+        assert_eq!(receipt.proposal_token_count, 1);
+        assert!(receipt.verification_rows > 0);
+        assert!(receipt.accepted_proposal_tokens <= receipt.proposal_token_count);
+        assert!(!receipt.committed_tokens.is_empty());
+        assert_eq!(
+            receipt.canonical_prediction_count,
+            receipt.committed_tokens.len()
+        );
+        assert_eq!(
+            receipt.verification_rows,
+            receipt.verification_row_predictions.len()
+        );
+        assert!(receipt.canonical_prediction_count <= receipt.verification_rows);
+        assert!(receipt.canonical_position >= receipt.base_position);
+        assert!(receipt.canonical_position <= receipt.position_after_verification);
+    }
+    Ok(())
 }
 
 #[test]

crates/mesh-llm-host-runtime/src/inference/skippy/resolver/translation.rs

@@ -280,27 +280,14 @@ impl ResolvedSkippyConfig {
             if self.skippy.prefill_controls_explicit {
                 bail!("skippy prefill chunk controls require staged serving");
             }
-            // A stage with no downstream delegates to generate_local_tokens, which
-            // has no N-gram verification path. Reject a standalone N-gram plan here
-            // rather than silently running target-only while reporting a proposer.
-            if self.speculative.decode.ngram.is_some()
-                && !self.speculative.decode.native_mtp.enabled
-            {
-                bail!(
-                    "standalone N-gram speculation ({}) requires multi-stage split serving; \
-                     single-stage and direct GGUF requests have no N-gram verification path",
-                    self.speculative.decode.effective_strategy
-                );
-            }
         }
         Ok(())
     }
 
-    fn speculative_mode_for_embedded(&self, staged: bool) -> &'static str {
+    fn speculative_mode_for_embedded(&self, _staged: bool) -> &'static str {
         if self.speculative.mode == "draft" && self.speculative.draft_model_path.is_some() {
             "draft"
-        } else if staged
-            && self.speculative.decode.ngram.is_some()
+        } else if self.speculative.decode.ngram.is_some()
             && !self.speculative.decode.native_mtp.enabled
         {
             "ngram"

crates/mesh-native-serving-plugin-api/README.md

@@ -14,9 +14,27 @@ layout of the prefix understood by an older plugin. Removing or reordering
 fields, changing a field's type, or changing a callback signature requires a
 new versioned ABI table and continued support for the previous table.
 
-The V1 activation contract lends the complete native tokenizer inventory through
-`ActivationContext`. A plugin copies or transforms the inventory before
-returning from `activate`.
+The V2 activation contract lends a model-bound tokenizer capability through
+`ActivationContext`. The capability includes the opaque native tokenizer
+inventory, the model identity and binding digest, explicit input/output
+limits, and a bounded encode callback for tagged ordinary-byte and opaque
+control pieces. Mesh does not know or own Rosetta vocabulary: a consumer
+builds any translation from the inventory and calls the capability outside the
+proposal deadline. Unsupported or non-lossless input is rejected; it is never
+decoded with replacement semantics. A plugin copies or transforms the
+inventory before returning from `activate` and must not call the encode
+callback from proposal callbacks.
+
+The V2 table is a breaking replacement for V1. Plugins must resolve
+`mesh_native_serving_plugin_v2` and validate the V2 ABI/version and structure
+sizes before use.
+
+For `TokenizerCapability::encode`, a non-null `output_length` is required. The
+host initializes it to zero before validating the remaining arguments, so every
+non-OK return other than `OUTPUT_TOO_SMALL` reports zero tokens; on
+`OUTPUT_TOO_SMALL` it reports the required capacity, and on `OK` it reports the
+number of written tokens. A null `output_length` cannot be written and therefore
+returns `INVALID_ARGUMENT`.
 
 Each proposal query carries the capacity Skippy can verify at that exact decode
 position. The native host adapter also applies a bounded implementation cap so