feat(topology): add a performance-aware planner and simulator

Cargo.toml

@@ -43,6 +43,7 @@ members = [
     "crates/skippy-tokenizer",
     "crates/skippy-protocol",
     "crates/skippy-coordinator",
+    "crates/skippy-topology-sim",
     "crates/skippy-topology",
     "crates/skippy-cache",
     "crates/skippy-metrics",

crates/mesh-llm-host-runtime/src/api/tests/node_state.rs

@@ -87,6 +87,7 @@ fn make_test_state_peer(seed: u8, role: mesh::NodeRole) -> mesh::PeerInfo {
         models: vec![],
         vram_bytes: 0,
         rtt_ms: None,
+        rtt_observation_window: None,
         model_source: None,
         admitted: true,
         serving_models: vec![],
@@ -125,6 +126,7 @@ fn make_test_state_peer(seed: u8, role: mesh::NodeRole) -> mesh::PeerInfo {
 
         display_rtt: None,
         selected_path: None,
+        observed_large_frame: None,
         propagated_latency: None,
         inference_admission_state: None,
     }

crates/mesh-llm-host-runtime/src/api/tests/support.rs

@@ -617,6 +617,7 @@ fn make_test_peer(
         models: Vec::new(),
         vram_bytes: 24_000_000_000,
         rtt_ms: None,
+        rtt_observation_window: None,
         model_source: None,
         admitted: true,
         serving_models: serving_models.into_iter().map(str::to_string).collect(),
@@ -654,6 +655,7 @@ fn make_test_peer(
 
         display_rtt: None,
         selected_path: None,
+        observed_large_frame: None,
         propagated_latency: None,
         inference_admission_state: None,
     }

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

@@ -380,9 +380,33 @@ impl Node {
             serving_models.clear();
             hosted_models.clear();
         }
-        let advertised_model_throughput = self
+        let mut advertised_model_throughput = self
             .routing_metrics
             .advertisable_model_throughput(&hosted_models);
+        for timing in skippy_server::stage_decode_timing_hints() {
+            if !hosted_models.iter().any(|model| model == &timing.model_id) {
+                continue;
+            }
+            if let Some(existing) = advertised_model_throughput
+                .iter_mut()
+                .find(|hint| hint.model_name == timing.model_id)
+            {
+                existing.observed_stage_us_per_layer = Some(timing.observed_us_per_layer);
+                existing.stage_timing_samples = Some(timing.sample_count);
+                existing.stage_timing_age_ms = Some(timing.sample_age_ms);
+            } else {
+                advertised_model_throughput.push(crate::network::metrics::ModelThroughputHint {
+                    model_name: timing.model_id,
+                    avg_tokens_per_second_milli: 0,
+                    throughput_samples: 0,
+                    observed_stage_us_per_layer: Some(timing.observed_us_per_layer),
+                    stage_timing_samples: Some(timing.sample_count),
+                    stage_timing_age_ms: Some(timing.sample_age_ms),
+                });
+            }
+        }
+        let advertised_model_throughput =
+            crate::network::metrics::sanitize_model_throughput_hints(advertised_model_throughput);
         let now_unix_ms = current_time_unix_ms();
         let cache_affinity = cache_affinity_gossip::local_advertisement(
             &self.cache_affinity_inventory,

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

@@ -163,7 +163,7 @@ pub use node::{
 pub(crate) use node::{PeerDownReport, peer_down_endpoint_id};
 pub(crate) use peer_state::{
     ControlListenerLifecycle, DEAD_PEER_TTL, MeshState, PEER_DOWN_REPORTER_COOLDOWN_SECS,
-    PEER_STALE_SECS, resolve_peer_leaving,
+    PEER_STALE_SECS, RttObservationAges, resolve_peer_leaving,
 };
 #[expect(
     unused_imports,

crates/mesh-llm-host-runtime/src/mesh/node.rs

@@ -1547,6 +1547,21 @@ impl Node {
         let updated_peer = {
             let mut state = self.state.lock().await;
             if let Some(peer) = state.peers.get_mut(&id) {
+                let observed_at = std::time::Instant::now();
+                match peer.rtt_observation_window.as_mut() {
+                    Some(window) => {
+                        window.sample_count = window.sample_count.saturating_add(1);
+                        window.last_observed_at = observed_at;
+                    }
+                    None => {
+                        peer.rtt_observation_window =
+                            Some(crate::mesh::peer_state::RttObservationWindow {
+                                sample_count: 1,
+                                first_observed_at: observed_at,
+                                last_observed_at: observed_at,
+                            });
+                    }
+                }
                 let prev = peer.rtt_ms;
                 // Only accept equal-or-lower RTT for planner preference and display.
                 // Gossip round-trip timing can inflate the value when routed via
@@ -1556,14 +1571,14 @@ impl Node {
                     // Store display_rtt regardless (for UI refresh), but don't update best RTT.
                     peer.display_rtt = Some(DirectLatencyObservation {
                         rtt_ms,
-                        observed_at: std::time::Instant::now(),
+                        observed_at,
                     });
                     return;
                 }
                 peer.rtt_ms = Some(rtt_ms);
                 peer.display_rtt = Some(DirectLatencyObservation {
                     rtt_ms,
-                    observed_at: std::time::Instant::now(),
+                    observed_at,
                 });
                 Some(peer.clone())
             } else {
@@ -1581,6 +1596,40 @@ impl Node {
         }
     }
 
+    /// Record a passive large-frame throughput observation for a peer link,
+    /// measured from a real bulk transfer (artifact download or upload).
+    /// Samples below the minimum duration/size floor are ignored — they
+    /// measure handshake jitter, not sustained throughput. Latest-wins.
+    pub(crate) async fn record_peer_large_frame_observation(
+        &self,
+        id: EndpointId,
+        bytes: u64,
+        elapsed: std::time::Duration,
+    ) {
+        const MIN_SAMPLE_BYTES: u64 = 512 * 1024;
+        const MIN_SAMPLE_DURATION: std::time::Duration = std::time::Duration::from_millis(100);
+        if bytes < MIN_SAMPLE_BYTES || elapsed < MIN_SAMPLE_DURATION {
+            return;
+        }
+        let elapsed_micros = elapsed.as_micros().max(1);
+        let mib_per_s = ((u128::from(bytes) * 1_000_000 / elapsed_micros) / 1_048_576) as u32;
+        if mib_per_s == 0 {
+            return;
+        }
+        let mut state = self.state.lock().await;
+        if let Some(peer) = state.peers.get_mut(&id) {
+            peer.observed_large_frame = Some(LargeFrameObservation {
+                mib_per_s,
+                observed_at: std::time::Instant::now(),
+            });
+            tracing::debug!(
+                "Peer {} large-frame: {mib_per_s} MiB/s over {} bytes in {elapsed:?}",
+                id.fmt_short(),
+                bytes
+            );
+        }
+    }
+
     pub(crate) async fn update_peer_selected_path(
         &self,
         id: EndpointId,

crates/mesh-llm-host-runtime/src/mesh/peer_state.rs

@@ -175,6 +175,20 @@ pub struct DirectLatencyObservation {
     pub observed_at: std::time::Instant,
 }
 
+/// A large-frame throughput observation on a peer link, measured passively
+/// from a real artifact transfer (bulk bytes over the same QUIC transport
+/// the split pipeline uses). Latest-wins; `observed_at` lets consumers
+/// discard stale samples.
+#[derive(Debug, Clone)]
+pub struct LargeFrameObservation {
+    pub mib_per_s: u32,
+    pub observed_at: std::time::Instant,
+}
+
+/// How long a passive large-frame observation stays planner-relevant.
+pub const LARGE_FRAME_OBSERVATION_MAX_AGE: std::time::Duration =
+    std::time::Duration::from_secs(30 * 60);
+
 /// Latency propagated via transitive gossip (not measured directly).
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct PropagatedLatencyObservation {
@@ -201,6 +215,26 @@ pub struct DisplayLatency {
     pub observer_id: Option<EndpointId>,
 }
 
+/// Confidence metadata retained behind the planner's best-seen RTT floor.
+///
+/// This intentionally records only observation count and timing. It is not a
+/// variance or jitter estimate: iroh owns path-quality selection, while split
+/// placement only needs to distinguish a one-off early sample from a floor
+/// corroborated across the connection-settle window.
+#[derive(Clone, Copy, Debug)]
+pub(crate) struct RttObservationWindow {
+    pub(crate) sample_count: u32,
+    pub(crate) first_observed_at: std::time::Instant,
+    pub(crate) last_observed_at: std::time::Instant,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) struct RttObservationAges {
+    pub(crate) sample_count: u32,
+    pub(crate) first_sample_age_ms: u64,
+    pub(crate) last_sample_age_ms: u64,
+}
+
 #[derive(Debug, Clone)]
 pub struct PeerInfo {
     pub id: EndpointId,
@@ -213,6 +247,9 @@ pub struct PeerInfo {
     pub models: Vec<String>,
     pub vram_bytes: u64,
     pub rtt_ms: Option<u32>,
+    /// Observation confidence for `rtt_ms`, whose value remains the minimum
+    /// accepted sample. Higher samples still advance this window.
+    pub(crate) rtt_observation_window: Option<RttObservationWindow>,
     pub model_source: Option<String>,
     pub admitted: bool,
     /// All models assigned to this peer, even if not yet healthy.
@@ -269,6 +306,11 @@ pub struct PeerInfo {
     pub display_rtt: Option<DirectLatencyObservation>,
     /// Last selected path observed on the mesh control connection to this peer.
     pub(crate) selected_path: Option<SelectedPathObservation>,
+    /// Latest large-frame throughput observed on this peer's link, measured
+    /// passively from real artifact transfers (either direction). Used as the
+    /// edge bandwidth signal for topology planning; stale observations decay
+    /// to `None` (latency-only edge).
+    pub(crate) observed_large_frame: Option<LargeFrameObservation>,
     /// Latency propagated via transitive gossip.
     pub propagated_latency: Option<PropagatedLatencyObservation>,
     pub owner_summary: OwnershipSummary,
@@ -320,6 +362,7 @@ impl PeerInfo {
             models: ann.models.clone(),
             vram_bytes: ann.vram_bytes,
             rtt_ms: None,
+            rtt_observation_window: None,
             model_source: ann.model_source.clone(),
             admitted: false,
             serving_models: ann.serving_models.clone(),
@@ -358,6 +401,7 @@ impl PeerInfo {
             cache_affinity: ann.cache_affinity.clone(),
             display_rtt: None,
             selected_path: None,
+            observed_large_frame: None,
             propagated_latency: None,
             owner_summary,
             inference_admission_state: ann.inference_admission_state,
@@ -373,6 +417,26 @@ impl PeerInfo {
         self.display_rtt.as_ref().map(|d| d.rtt_ms).or(self.rtt_ms)
     }
 
+    pub(crate) fn rtt_observation_ages(&self) -> Option<RttObservationAges> {
+        let window = self.rtt_observation_window?;
+        Some(RttObservationAges {
+            sample_count: window.sample_count,
+            first_sample_age_ms: super::elapsed_ms_u64(window.first_observed_at.elapsed()),
+            last_sample_age_ms: super::elapsed_ms_u64(window.last_observed_at.elapsed()),
+        })
+    }
+
+    /// Sustained large-frame throughput for this peer link from a recent
+    /// passive artifact-transfer observation. `None` when never measured or
+    /// the observation has aged out — the planner then treats the edge as
+    /// latency-only, which is exactly the pre-probing behavior.
+    pub fn large_frame_mib_per_s(&self) -> Option<u32> {
+        self.observed_large_frame
+            .as_ref()
+            .filter(|observed| observed.observed_at.elapsed() <= LARGE_FRAME_OBSERVATION_MAX_AGE)
+            .map(|observed| observed.mib_per_s)
+    }
+
     pub(crate) fn split_stage_path_fallback(&self) -> Option<SelectedPathObservation> {
         let observation = self.selected_path?;
         if observation.path_type != "direct" {
@@ -893,6 +957,28 @@ impl Node {
         self.vram_bytes
     }
 
+    /// Measured sustained node performance for split planning: summed
+    /// memory bandwidth in MiB/s and fp16 compute in GFLOP/s across GPUs.
+    /// `(None, None)` until gpu-bench measurements populate the metrics —
+    /// the planner treats unreported nodes as capacity-only.
+    pub async fn sustained_perf_signals(&self) -> (Option<u32>, Option<u32>) {
+        let bandwidth = {
+            let metrics = self.gpu_mem_bandwidth_gbps.lock().await;
+            metrics.as_ref().map(|values| values.iter().sum::<f64>())
+        };
+        let compute = {
+            let metrics = self.gpu_compute_tflops_fp16.lock().await;
+            metrics.as_ref().map(|values| values.iter().sum::<f64>())
+        };
+        let bandwidth_mib = bandwidth
+            .map(|gbps| gbps * 1_000_000_000.0 / 1_048_576.0)
+            .and_then(|mib| u32::try_from(mib.max(0.0) as u64).ok());
+        let compute_gflops = compute
+            .map(|tflops| tflops * 1_000.0)
+            .and_then(|gflops| u32::try_from(gflops.max(0.0) as u64).ok());
+        (bandwidth_mib, compute_gflops)
+    }
+
     /// Local model-fit budget, including supported CPU offload memory.
     pub fn local_runtime_capacity_bytes(&self) -> u64 {
         self.local_runtime_capacity_bytes

crates/mesh-llm-host-runtime/src/mesh/stage_artifacts.rs

@@ -691,6 +691,7 @@ impl Node {
         );
 
         let transfer_result = async {
+            let transfer_started = std::time::Instant::now();
             append_artifact_transfer_body(
                 &mut recv,
                 &temp_path,
@@ -700,6 +701,15 @@ impl Node {
                 ARTIFACT_TRANSFER_READ_IDLE_TIMEOUT,
             )
             .await?;
+            // Passive edge-bandwidth measurement: this transfer moved real
+            // bulk bytes over the same QUIC transport the split pipeline
+            // uses. Record it as the peer link's large-frame observation.
+            self.record_peer_large_frame_observation(
+                peer_id,
+                response.total_size.saturating_sub(offset),
+                transfer_started.elapsed(),
+            )
+            .await;
 
             let actual_size = tokio::fs::metadata(&temp_path)
                 .await
@@ -967,6 +977,7 @@ impl Node {
             .context("seek artifact for transfer")?;
         let mut buffer = vec![0u8; ARTIFACT_TRANSFER_BUFFER_BYTES];
         let mut remaining = artifact.size.saturating_sub(request.offset);
+        let upload_started = std::time::Instant::now();
         while remaining > 0 {
             let limit = buffer.len().min(remaining as usize);
             let read = file
@@ -982,6 +993,14 @@ impl Node {
             .await?;
             remaining -= read as u64;
         }
+        // Passive edge-bandwidth measurement, upload direction: see
+        // fetch_artifact_from_peer for the rationale.
+        self.record_peer_large_frame_observation(
+            remote,
+            artifact.size.saturating_sub(request.offset),
+            upload_started.elapsed(),
+        )
+        .await;
         let _ = send.finish();
         Ok(())
     }