Run-ahead verify-window admission, stale-tail discard, and a jitter-capable wire model

crates/mesh-llm-config/src/model.rs

@@ -603,6 +603,7 @@ pub struct SpeculativeConfig {
     pub verify_window_min_tokens: Option<u32>,
     pub verify_window_max_tokens: Option<u32>,
     pub verify_window_pipeline_depth: Option<u32>,
+    pub verify_window_runahead_tokens: Option<u32>,
     pub spec_default: Option<BoolOrAuto>,
     pub(crate) legacy_draft_model_path_used: bool,
 }
@@ -654,6 +655,7 @@ impl SpeculativeConfig {
             verify_window_min_tokens: pick!(verify_window_min_tokens),
             verify_window_max_tokens: pick!(verify_window_max_tokens),
             verify_window_pipeline_depth: pick!(verify_window_pipeline_depth),
+            verify_window_runahead_tokens: pick!(verify_window_runahead_tokens),
             spec_default: pick!(spec_default),
             legacy_draft_model_path_used: overrides
                 .filter(|config| config.draft_model.is_some())
@@ -727,6 +729,8 @@ struct SpeculativeConfigRaw {
     #[serde(default)]
     verify_window_pipeline_depth: Option<u32>,
     #[serde(default)]
+    verify_window_runahead_tokens: Option<u32>,
+    #[serde(default)]
     spec_default: Option<BoolOrAuto>,
 }
 
@@ -771,6 +775,7 @@ impl<'de> Deserialize<'de> for SpeculativeConfig {
             verify_window_min_tokens: raw.verify_window_min_tokens,
             verify_window_max_tokens: raw.verify_window_max_tokens,
             verify_window_pipeline_depth: raw.verify_window_pipeline_depth,
+            verify_window_runahead_tokens: raw.verify_window_runahead_tokens,
             spec_default: raw.spec_default,
             legacy_draft_model_path_used: legacy_used,
         })
@@ -833,6 +838,10 @@ impl Serialize for SpeculativeConfig {
             "verify_window_pipeline_depth",
             &self.verify_window_pipeline_depth,
         )?;
+        map.serialize_entry(
+            "verify_window_runahead_tokens",
+            &self.verify_window_runahead_tokens,
+        )?;
         map.serialize_entry("spec_default", &self.spec_default)?;
         map.end()
     }

crates/mesh-llm-config/src/model/built_in_schema/control_behavior/speculative.rs

@@ -73,7 +73,8 @@ pub(super) fn apply_speculative_behavior(
         | "native_mtp_suppress_cooldown_draft_limit"
         | "verify_window_min_tokens"
         | "verify_window_max_tokens"
-        | "verify_window_pipeline_depth" => {}
+        | "verify_window_pipeline_depth"
+        | "verify_window_runahead_tokens" => {}
         _ => {}
     }
 }

crates/mesh-llm-config/src/model/built_in_schema/declarations.rs

@@ -659,6 +659,10 @@ fn speculative_settings(prefix: &str) -> Vec<ConfigSettingSchema> {
             &format!("{prefix}.verify_window_pipeline_depth"),
             ConfigValueSchema::Integer,
         ),
+        basic_setting(
+            &format!("{prefix}.verify_window_runahead_tokens"),
+            ConfigValueSchema::Integer,
+        ),
         basic_setting(&format!("{prefix}.spec_default"), bool_or_auto_schema()),
     ]
 }

crates/mesh-llm-config/src/model_validation.rs

@@ -9,7 +9,7 @@ use crate::model::{
     SkippyConfig, SpeculativeConfig, StringOrStringList, merge_hardware, merge_model_fit,
     merge_multimodal, merge_throughput,
 };
-use skippy_protocol::MAX_VERIFY_WINDOW_PIPELINE_DEPTH;
+use skippy_protocol::{MAX_VERIFY_WINDOW_PIPELINE_DEPTH, MAX_VERIFY_WINDOW_RUNAHEAD_TOKENS};
 
 use crate::validation_support::{
     looks_like_model_identifier, validate_allowed, validate_bool_or_auto, validate_hf_pair,
@@ -633,6 +633,14 @@ fn validate_verify_window_controls(
         &format!("{base_path}.verify_window_pipeline_depth"),
         1,
         u32::try_from(MAX_VERIFY_WINDOW_PIPELINE_DEPTH).expect("verify depth limit fits u32"),
+    )?;
+    validate_optional_u32_range(
+        config.verify_window_runahead_tokens,
+        &format!("{base_path}.verify_window_runahead_tokens"),
+        // Zero is the documented fixed-depth sentinel, so a model-level block
+        // can switch run-ahead back off when the global defaults enable it.
+        0,
+        u32::try_from(MAX_VERIFY_WINDOW_RUNAHEAD_TOKENS).expect("runahead limit fits u32"),
     )
 }
 

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

@@ -79,26 +79,36 @@ pub(crate) use stage::{
 pub(crate) use topology::{StageTopologyParticipant, plan_package_identity_topology};
 
 const BENCH_DOWNSTREAM_WIRE_DELAY_MS_ENV: &str = "MESH_LLM_BENCH_DOWNSTREAM_WIRE_DELAY_MS";
+const BENCH_DOWNSTREAM_WIRE_JITTER_MS_ENV: &str = "MESH_LLM_BENCH_DOWNSTREAM_WIRE_JITTER_MS";
+const BENCH_DOWNSTREAM_WIRE_STALL_MS_ENV: &str = "MESH_LLM_BENCH_DOWNSTREAM_WIRE_STALL_MS";
+const BENCH_DOWNSTREAM_WIRE_STALL_P_ENV: &str = "MESH_LLM_BENCH_DOWNSTREAM_WIRE_STALL_P";
 
 fn benchmark_downstream_wire_condition() -> Result<WireCondition> {
-    let delay_ms = match env::var(BENCH_DOWNSTREAM_WIRE_DELAY_MS_ENV) {
-        Ok(value) => parse_benchmark_downstream_wire_delay_ms(&value)?,
-        Err(env::VarError::NotPresent) => 0.0,
+    let delay_ms = parse_benchmark_wire_env(BENCH_DOWNSTREAM_WIRE_DELAY_MS_ENV)?;
+    let jitter_ms = parse_benchmark_wire_env(BENCH_DOWNSTREAM_WIRE_JITTER_MS_ENV)?;
+    let stall_ms = parse_benchmark_wire_env(BENCH_DOWNSTREAM_WIRE_STALL_MS_ENV)?;
+    let stall_p = parse_benchmark_wire_env(BENCH_DOWNSTREAM_WIRE_STALL_P_ENV)?;
+    WireCondition::with_jitter(delay_ms, None, jitter_ms, stall_ms, stall_p)
+}
+
+fn parse_benchmark_wire_env(name: &'static str) -> Result<f64> {
+    match env::var(name) {
+        Ok(value) => parse_benchmark_downstream_wire_value(name, &value),
+        Err(env::VarError::NotPresent) => Ok(0.0),
         Err(env::VarError::NotUnicode(_)) => {
-            anyhow::bail!("{BENCH_DOWNSTREAM_WIRE_DELAY_MS_ENV} must be valid UTF-8")
+            anyhow::bail!("{name} must be valid UTF-8")
         }
-    };
-    WireCondition::new(delay_ms, None)
+    }
 }
 
-fn parse_benchmark_downstream_wire_delay_ms(value: &str) -> Result<f64> {
-    let delay_ms = value.parse::<f64>().with_context(|| {
-        format!("{BENCH_DOWNSTREAM_WIRE_DELAY_MS_ENV} must be a finite non-negative number")
-    })?;
-    if !delay_ms.is_finite() || delay_ms < 0.0 {
-        anyhow::bail!("{BENCH_DOWNSTREAM_WIRE_DELAY_MS_ENV} must be a finite non-negative number");
+fn parse_benchmark_downstream_wire_value(name: &str, value: &str) -> Result<f64> {
+    let parsed = value
+        .parse::<f64>()
+        .with_context(|| format!("{name} must be a finite non-negative number"))?;
+    if !parsed.is_finite() || parsed < 0.0 {
+        anyhow::bail!("{name} must be a finite non-negative number");
     }
-    Ok(delay_ms)
+    Ok(parsed)
 }
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -1345,17 +1355,20 @@ mod tests {
 
     #[test]
     fn benchmark_wire_delay_accepts_finite_non_negative_values() {
-        assert_eq!(parse_benchmark_downstream_wire_delay_ms("0").unwrap(), 0.0);
         assert_eq!(
-            parse_benchmark_downstream_wire_delay_ms("25.5").unwrap(),
+            parse_benchmark_downstream_wire_value("test", "0").unwrap(),
+            0.0
+        );
+        assert_eq!(
+            parse_benchmark_downstream_wire_value("test", "25.5").unwrap(),
             25.5
         );
     }
 
     #[test]
     fn benchmark_wire_delay_rejects_invalid_values() {
         for value in ["-1", "NaN", "inf", "not-a-number"] {
-            assert!(parse_benchmark_downstream_wire_delay_ms(value).is_err());
+            assert!(parse_benchmark_downstream_wire_value("test", value).is_err());
         }
     }
 

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

@@ -397,6 +397,17 @@ fn resolve_decode_config(input: DecodeResolutionInput<'_>) -> Result<Speculative
             .and_then(|config| config.verify_window_pipeline_depth),
     )
     .map_or(config.verify_window.pipeline_depth, |value| value as usize);
+    config.verify_window.runahead_max_tokens = pick_optional_u32(
+        input
+            .model_config
+            .and_then(|config| config.verify_window_runahead_tokens),
+        input
+            .global_config
+            .and_then(|config| config.verify_window_runahead_tokens),
+    )
+    .map_or(config.verify_window.runahead_max_tokens, |value| {
+        value as usize
+    });
     if config.verify_window.min_tokens > config.verify_window.max_tokens {
         bail!("skippy speculative verify window requires min_tokens <= max_tokens");
     }
@@ -479,6 +490,7 @@ fn package_decode_config(
             min_tokens: 1,
             max_tokens: 4,
             pipeline_depth: 1,
+            runahead_max_tokens: 0,
         });
     let effective_strategy = match (native_mtp.enabled, ngram.as_ref().map(|value| value.kind)) {
         (true, Some(NgramProposerKind::Cache)) => "native-mtp+ngram-cache",
@@ -564,6 +576,7 @@ fn verify_window_config(policy: &PackageWindowPolicyInfo) -> VerifyWindowConfig
         min_tokens: policy.min_window as usize,
         max_tokens: policy.max_window as usize,
         pipeline_depth: policy.pipeline_depth.unwrap_or(1) as usize,
+        runahead_max_tokens: 0,
     }
 }
 

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

@@ -1592,3 +1592,46 @@ verify_window_pipeline_depth = 2
     assert_eq!(translated.max_proposal_tokens, 48);
     assert_eq!(args.speculative.verify_window.pipeline_depth, 2);
 }
+
+#[test]
+fn model_level_zero_runahead_overrides_a_positive_global_default() {
+    use crate::plugin::SpeculativeConfig;
+    let global: SpeculativeConfig = toml::from_str(
+        r#"
+strategy = "ngram-suffix"
+ngram_proposer = "suffix"
+ngram_min = 5
+ngram_max = 32
+verify_window_pipeline_depth = 2
+verify_window_runahead_tokens = 256
+"#,
+    )
+    .expect("parse global speculative config");
+    let model: SpeculativeConfig = toml::from_str(
+        r#"
+verify_window_runahead_tokens = 0
+"#,
+    )
+    .expect("parse model speculative config");
+    let inherited = super::speculative::resolve_speculative_config(
+        None,
+        Some(&global),
+        "meshllm/test-model",
+        std::path::Path::new("/nonexistent/test-model.gguf"),
+        None,
+    )
+    .expect("global run-ahead must resolve");
+    assert_eq!(inherited.decode.verify_window.runahead_max_tokens, 256);
+    let overridden = super::speculative::resolve_speculative_config(
+        Some(&model),
+        Some(&global),
+        "meshllm/test-model",
+        std::path::Path::new("/nonexistent/test-model.gguf"),
+        None,
+    )
+    .expect("model-level zero must resolve to fixed-depth mode");
+    assert_eq!(
+        overridden.decode.verify_window.runahead_max_tokens, 0,
+        "Some(0) at the model level must win over the inherited positive default"
+    );
+}

crates/mesh-llm-host-runtime/src/protocol/convert.rs

@@ -17,7 +17,7 @@ fn skippy_stage_subprotocols(
     let mut features = vec![skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STAGE_CONTROL.to_string()];
     if stage_protocol_generation_supported {
         features.push(
-            skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STAGE_PROTOCOL_GENERATION_V4.to_string(),
+            skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STAGE_PROTOCOL_GENERATION_V5.to_string(),
         );
     }
     if artifact_transfer_supported {
@@ -50,7 +50,7 @@ fn supports_skippy_status_list(subprotocols: &[crate::proto::node::MeshSubprotoc
 fn supports_skippy_stage_generation(subprotocols: &[crate::proto::node::MeshSubprotocol]) -> bool {
     supports_skippy_stage_feature(
         subprotocols,
-        skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STAGE_PROTOCOL_GENERATION_V4,
+        skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STAGE_PROTOCOL_GENERATION_V5,
     ) && supports_skippy_stage_feature(
         subprotocols,
         skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STAGE_CONTROL,

crates/mesh-llm-host-runtime/src/protocol/tests/announcements.rs

@@ -82,7 +82,7 @@ fn owner_fields_roundtrip_through_proto_announcement() {
             .any(|feature| feature == skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STATUS_LIST)
     );
     assert!(skippy.features.iter().any(|feature| feature
-        == skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STAGE_PROTOCOL_GENERATION_V4));
+        == skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STAGE_PROTOCOL_GENERATION_V5));
     assert_eq!(
         proto_pa
             .owner_attestation
@@ -310,7 +310,7 @@ fn proto_announcement_without_stage_control_is_not_stage_compatible() {
             name: skippy_protocol::STAGE_SUBPROTOCOL_NAME.to_string(),
             major: skippy_protocol::STAGE_SUBPROTOCOL_MAJOR,
             features: vec![
-                skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STAGE_PROTOCOL_GENERATION_V4.to_string(),
+                skippy_protocol::STAGE_SUBPROTOCOL_FEATURE_STAGE_PROTOCOL_GENERATION_V5.to_string(),
             ],
         }],
         ..Default::default()

crates/mesh-llm-host-runtime/tests/fixtures/config_schema_defaults_ui_reference.json

@@ -924,6 +924,13 @@
         "kind": "built_in"
       }
     },
+    {
+      "canonical_path": "defaults.speculative.verify_window_runahead_tokens",
+      "support": "supported",
+      "source": {
+        "kind": "built_in"
+      }
+    },
     {
       "canonical_path": "defaults.throughput.continuous_batching",
       "support": "supported",
@@ -995,4 +1002,4 @@
       }
     }
   ]
-}
+}

crates/skippy-protocol/src/binary/types.rs

@@ -58,6 +58,7 @@ pub enum WireMessageKind {
     DecodeLightCtx = 9,
     VerifyWindow = 21,
     RetireVerifyWindow = 22,
+    DiscardStaleWindows = 23,
     StateExport = 13,
     ConfigureGeneration = 14,
     ProbePrefill = 15,
@@ -101,6 +102,14 @@ impl WireMessageKind {
         matches!(self, Self::RetireVerifyWindow)
     }
 
+    /// Control message invalidating a contiguous range of not-yet-executed
+    /// verify windows after the driver detected divergence. Recorded at
+    /// message-receive time so buffered stale windows are skipped instead of
+    /// executed.
+    pub fn is_stale_window_discard(self) -> bool {
+        matches!(self, Self::DiscardStaleWindows)
+    }
+
     pub fn is_generation_control(self) -> bool {
         matches!(self, Self::ConfigureGeneration)
     }
@@ -147,6 +156,7 @@ impl TryFrom<i32> for WireMessageKind {
             20 => Ok(Self::PredictionReturnOpen),
             21 => Ok(Self::VerifyWindow),
             22 => Ok(Self::RetireVerifyWindow),
+            23 => Ok(Self::DiscardStaleWindows),
             _ => Err(invalid_data("unknown stage message kind")),
         }
     }
@@ -329,6 +339,7 @@ impl StageStateHeader {
             WireMessageKind::StateImport | WireMessageKind::StateExport
         ) || kind.is_session_control()
             || kind.is_verify_retirement()
+            || kind.is_stale_window_discard()
             || kind.is_generation_control()
         {
             return true;

crates/skippy-protocol/src/lib.rs

@@ -29,11 +29,11 @@ pub use messages::{
     StateImportMessage, StopMessage, TokenReplyMessage,
 };
 pub use validation::{
-    MAX_STAGE_FRAME_BYTES, MAX_VERIFY_WINDOW_PIPELINE_DEPTH, SCHEMA_VERSION, STAGE_ALPN_V2,
-    STAGE_PROTOCOL_GENERATION, STAGE_STREAM_ARTIFACT_TRANSFER, STAGE_STREAM_CONTROL,
-    STAGE_STREAM_TRANSPORT, STAGE_SUBPROTOCOL_FEATURE_ARTIFACT_TRANSFER,
+    MAX_STAGE_FRAME_BYTES, MAX_VERIFY_WINDOW_PIPELINE_DEPTH, MAX_VERIFY_WINDOW_RUNAHEAD_TOKENS,
+    SCHEMA_VERSION, STAGE_ALPN_V2, STAGE_PROTOCOL_GENERATION, STAGE_STREAM_ARTIFACT_TRANSFER,
+    STAGE_STREAM_CONTROL, STAGE_STREAM_TRANSPORT, STAGE_SUBPROTOCOL_FEATURE_ARTIFACT_TRANSFER,
     STAGE_SUBPROTOCOL_FEATURE_STAGE_CONTROL, STAGE_SUBPROTOCOL_FEATURE_STAGE_GENERATION,
-    STAGE_SUBPROTOCOL_FEATURE_STAGE_PROTOCOL_GENERATION_V4, STAGE_SUBPROTOCOL_FEATURE_STATUS_LIST,
+    STAGE_SUBPROTOCOL_FEATURE_STAGE_PROTOCOL_GENERATION_V5, STAGE_SUBPROTOCOL_FEATURE_STATUS_LIST,
     STAGE_SUBPROTOCOL_MAJOR, STAGE_SUBPROTOCOL_NAME, StageFrameError,
     validate_stage_artifact_transfer_request, validate_stage_artifact_transfer_response,
     validate_stage_control_request, validate_stage_control_response, validate_stage_transport_open,
@@ -53,7 +53,7 @@ mod tests {
         stage_control_response,
     };
     use super::{
-        STAGE_PROTOCOL_GENERATION, STAGE_SUBPROTOCOL_FEATURE_STAGE_PROTOCOL_GENERATION_V4,
+        STAGE_PROTOCOL_GENERATION, STAGE_SUBPROTOCOL_FEATURE_STAGE_PROTOCOL_GENERATION_V5,
         StageFrameError, validate_stage_artifact_transfer_request,
         validate_stage_artifact_transfer_response, validate_stage_control_request,
         validate_stage_control_response, validate_stage_transport_open,
@@ -62,7 +62,7 @@ mod tests {
     #[test]
     fn stage_protocol_generation_feature_names_current_generation() {
         assert_eq!(
-            STAGE_SUBPROTOCOL_FEATURE_STAGE_PROTOCOL_GENERATION_V4,
+            STAGE_SUBPROTOCOL_FEATURE_STAGE_PROTOCOL_GENERATION_V5,
             format!("stage-generation-{STAGE_PROTOCOL_GENERATION}")
         );
     }