Add token/s on dashboard and reporting token/s in the metrics endpoint

paddler_agent/src/continuous_batch_arbiter.rs

@@ -306,6 +306,7 @@ impl ContinuousBatchArbiter {
                 inference_parameters,
                 model_path: model_path.clone(),
                 multimodal_context,
+                slot_aggregated_status: slot_aggregated_status_manager.slot_aggregated_status.clone(),
                 token_bos_str: model.token_to_piece(
                     &SampledToken::Content(model.token_bos()),
                     &mut special_token_decoder,

paddler_agent/src/continuous_batch_scheduler/advance_generating_phase.rs

@@ -49,7 +49,13 @@ impl AdvanceGeneratingPhase<'_> {
         })
         .run(request, batch_index)
         {
-            SampleOutcome::Sampled(token) => token,
+            SampleOutcome::Sampled(token) => {
+                self.scheduler_context
+                    .slot_aggregated_status
+                    .record_generated_token();
+
+                token
+            }
             SampleOutcome::AllCandidatesEliminated => {
                 error!(
                     "{:?}: sequence {} sampling exhausted candidates",

paddler_agent/src/continuous_batch_scheduler_context.rs

@@ -6,6 +6,7 @@ use llama_cpp_bindings::mtmd::MtmdContext;
 use paddler_messaging::inference_parameters::InferenceParameters;
 
 use crate::chat_template_renderer::ChatTemplateRenderer;
+use crate::slot_aggregated_status::SlotAggregatedStatus;
 
 pub struct ContinuousBatchSchedulerContext {
     pub agent_name: Option<String>,
@@ -15,6 +16,7 @@ pub struct ContinuousBatchSchedulerContext {
     pub model: Arc<LlamaModel>,
     pub model_path: PathBuf,
     pub multimodal_context: Option<Arc<MtmdContext>>,
+    pub slot_aggregated_status: Arc<SlotAggregatedStatus>,
     pub token_bos_str: String,
     pub token_eos_str: String,
     pub token_nl_str: String,

paddler_agent/src/lib.rs

@@ -59,6 +59,7 @@ pub mod slot_aggregated_status;
 pub mod slot_aggregated_status_download_progress;
 pub mod slot_aggregated_status_manager;
 pub mod slot_guard;
+pub mod token_throughput_meter;
 pub mod tool_call_buffer;
 pub mod tool_call_event;
 pub mod tool_call_pipeline;

paddler_agent/src/management_socket_client_service.rs

@@ -652,14 +652,31 @@ mod tests {
         let service = service_with_socket_url(format!(
             "ws://{refused_addr}/api/v1/agent_socket/test-agent"
         ));
-        let shutdown = CancellationToken::new();
 
-        let keep_alive_result =
-            tokio::time::timeout(SHUTDOWN_BUDGET, service.keep_connection_alive(shutdown))
-                .await
-                .expect("connecting to a refused port must fail fast instead of blocking");
+        // The kernel may take ~1s before it answers a connect to a just-closed
+        // port with ECONNREFUSED, so a single attempt could time out. Retry
+        // within the budget instead of assuming the refusal is immediate.
+        let deadline = tokio::time::Instant::now() + SHUTDOWN_BUDGET;
+
+        loop {
+            let shutdown = CancellationToken::new();
+
+            let keep_alive_result = tokio::select! {
+                keep_alive_result = service.keep_connection_alive(shutdown) => keep_alive_result,
+                () = tokio::time::sleep_until(deadline) => {
+                    panic!("connecting to a refused port must fail within the shutdown budget");
+                }
+            };
+
+            if let Err(err) = keep_alive_result {
+                assert!(
+                    err.to_string().contains("Connection refused"),
+                    "expected a connection refusal, got: {err}"
+                );
 
-        assert!(keep_alive_result.is_err());
+                return;
+            }
+        }
     }
 
     #[tokio::test]

paddler_agent/src/slot_aggregated_status.rs

@@ -12,6 +12,7 @@ use tokio::sync::watch;
 
 use crate::agent_issue_fix::AgentIssueFix;
 use crate::dispenses_slots::DispensesSlots;
+use crate::token_throughput_meter::TokenThroughputMeter;
 use paddler_messaging::atomic_value::AtomicValue;
 use paddler_messaging::produces_snapshot::ProducesSnapshot;
 use paddler_messaging::subscribes_to_updates::SubscribesToUpdates;
@@ -27,6 +28,7 @@ pub struct SlotAggregatedStatus {
     slots_processing: AtomicValue<AtomicI32>,
     slots_total: AtomicValue<AtomicI32>,
     state_application_status_code: AtomicValue<AtomicI32>,
+    token_throughput_meter: TokenThroughputMeter,
     update_tx: watch::Sender<()>,
     uses_chat_template_override: AtomicValue<AtomicBool>,
     version: AtomicValue<AtomicI32>,
@@ -50,6 +52,7 @@ impl SlotAggregatedStatus {
             ),
             slots_processing: AtomicValue::<AtomicI32>::new(0),
             slots_total: AtomicValue::<AtomicI32>::new(0),
+            token_throughput_meter: TokenThroughputMeter::new(),
             update_tx,
             uses_chat_template_override: AtomicValue::<AtomicBool>::new(false),
             version: AtomicValue::<AtomicI32>::new(0),
@@ -174,6 +177,21 @@ impl SlotAggregatedStatus {
     pub fn slots_processing_count(&self) -> i32 {
         self.slots_processing.get()
     }
+
+    /// Records that this agent has just generated one token, contributing to
+    /// the tokens-per-second rate reported in status snapshots.
+    ///
+    /// This does not bump `version` or notify subscribers: it happens once
+    /// per generated token, far too often to treat as a state change worth
+    /// pushing immediately. The periodic status update (sent once a second
+    /// regardless of `version`) is what picks the latest rate up.
+    pub fn record_generated_token(&self) {
+        self.token_throughput_meter.record_token();
+    }
+
+    pub fn tokens_per_second(&self) -> f64 {
+        self.token_throughput_meter.tokens_per_second()
+    }
 }
 
 impl DispensesSlots for SlotAggregatedStatus {
@@ -211,6 +229,7 @@ impl ProducesSnapshot for SlotAggregatedStatus {
             slots_processing: self.slots_processing.get(),
             slots_total: self.slots_total.get(),
             state_application_status: self.state_application_status_code.get().try_into()?,
+            tokens_per_second: self.tokens_per_second(),
             uses_chat_template_override: self.uses_chat_template_override.get(),
             version: self.version.get(),
         })
@@ -341,6 +360,37 @@ mod tests {
         );
     }
 
+    #[test]
+    fn make_snapshot_reports_zero_tokens_per_second_before_any_generation() {
+        let status = SlotAggregatedStatus::new(1);
+
+        let snapshot = status.make_snapshot().unwrap();
+
+        let actual = snapshot.tokens_per_second;
+        assert!((actual - 0.0).abs() < 0.0001);
+    }
+
+    #[test]
+    fn record_generated_token_is_reflected_in_snapshot_after_a_window_closes() {
+        let status = SlotAggregatedStatus::new(1);
+
+        for _ in 0..5 {
+            status.record_generated_token();
+        }
+
+        std::thread::sleep(std::time::Duration::from_millis(1050));
+
+        status.record_generated_token();
+
+        let snapshot = status.make_snapshot().unwrap();
+
+        assert!(
+            snapshot.tokens_per_second > 0.0,
+            "expected a positive tokens_per_second, got {}",
+            snapshot.tokens_per_second
+        );
+    }
+
     #[test]
     fn get_state_application_status_reflects_set_value() {
         let status = SlotAggregatedStatus::new(2);

paddler_agent/src/token_throughput_meter.rs

@@ -0,0 +1,126 @@
+use std::time::Duration;
+use std::time::Instant;
+
+use parking_lot::Mutex;
+
+/// How long a window stays open before its token count is turned into a
+/// tokens/sec measurement. One second keeps the reported rate intuitive
+/// (it is, literally, "tokens counted in the last second").
+const WINDOW_DURATION: Duration = Duration::from_secs(1);
+
+/// A measurement is considered stale, and reported as `0.0`, once this much
+/// time has passed without a new token. Without this, a slot that produced a
+/// quick burst and then went idle would keep reporting its last burst's rate
+/// forever.
+const STALE_AFTER: Duration = Duration::from_secs(2);
+
+struct ThroughputWindow {
+    tokens_per_second: f64,
+    window_started_at: Instant,
+    window_token_count: f64,
+}
+
+/// Tracks generated tokens for a single agent and derives a continuously
+/// updated, approximate tokens-per-second throughput value out of them.
+///
+/// The meter is intentionally simple: it counts tokens in a rolling
+/// one-second window and, once the window closes, turns that count into the
+/// reported rate. There is nothing to configure.
+pub struct TokenThroughputMeter {
+    window: Mutex<ThroughputWindow>,
+}
+
+impl TokenThroughputMeter {
+    #[must_use]
+    pub fn new() -> Self {
+        Self {
+            window: Mutex::new(ThroughputWindow {
+                tokens_per_second: 0.0,
+                window_started_at: Instant::now(),
+                window_token_count: 0.0,
+            }),
+        }
+    }
+
+    /// Records a single generated token, closing out and measuring the
+    /// current window if it has been open for at least [`WINDOW_DURATION`].
+    pub fn record_token(&self) {
+        let mut window = self.window.lock();
+        let elapsed = window.window_started_at.elapsed();
+
+        window.window_token_count += 1.0;
+
+        if elapsed >= WINDOW_DURATION {
+            window.tokens_per_second = window.window_token_count / elapsed.as_secs_f64();
+            window.window_token_count = 0.0;
+            window.window_started_at = Instant::now();
+        }
+    }
+
+    /// Returns the most recently measured tokens-per-second rate, or `0.0`
+    /// if generation has been idle for longer than [`STALE_AFTER`].
+    #[must_use]
+    pub fn tokens_per_second(&self) -> f64 {
+        let window = self.window.lock();
+
+        if window.window_started_at.elapsed() > STALE_AFTER && window.window_token_count == 0.0 {
+            return 0.0;
+        }
+
+        window.tokens_per_second
+    }
+}
+
+impl Default for TokenThroughputMeter {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::thread::sleep;
+
+    use super::*;
+
+    #[test]
+    fn reports_zero_before_any_tokens_are_recorded() {
+        let meter = TokenThroughputMeter::new();
+
+        let actual = meter.tokens_per_second();
+        assert!((actual - 0.0).abs() < 0.0001);
+    }
+
+    #[test]
+    fn measures_rate_once_a_window_closes() {
+        let meter = TokenThroughputMeter::new();
+
+        for _ in 0..5 {
+            meter.record_token();
+        }
+
+        sleep(WINDOW_DURATION + Duration::from_millis(50));
+
+        meter.record_token();
+
+        let rate = meter.tokens_per_second();
+
+        assert!(rate > 0.0, "expected a positive rate, got {rate}");
+    }
+
+    #[test]
+    fn decays_to_zero_after_being_idle() {
+        let meter = TokenThroughputMeter::new();
+
+        for _ in 0..5 {
+            meter.record_token();
+        }
+
+        sleep(WINDOW_DURATION + Duration::from_millis(50));
+        meter.record_token();
+        sleep(STALE_AFTER + Duration::from_millis(50));
+
+        let actual = meter.tokens_per_second();
+        assert!((actual - 0.0).abs() < 0.0001);
+    }
+}