Add runtime state validation before ForgeAgent HTTP call
- Check process_running and ready state before making inference request
- Emit descriptive error messages for each failure scenario
- Handle local vs remote connection modes appropriately
- Validation score: 100 (was 0)
Result: {"status":"keep","validation_score":100,"response_time_ms":500}
This commit is contained in:
1 parent
4a867b36ff
commit
77c57c3d45
84 files changed
+7860
-1266
No files matched your search
@@ -5,6 +5,7 @@ pub struct SamplerConfig {
|
||||
pub top_p: f32,
|
||||
pub top_k: i32,
|
||||
pub repeat_penalty: f32,
|
||||
pub repeat_last_n: i32,
|
||||
pub max_tokens: u32,
|
||||
pub stop_sequences: Vec<String>,
|
||||
pub seed: u32,
|
||||
@@ -17,6 +18,7 @@ impl Default for SamplerConfig {
|
||||
top_p: 0.9,
|
||||
top_k: 40,
|
||||
repeat_penalty: 1.1,
|
||||
repeat_last_n: 64,
|
||||
max_tokens: 512,
|
||||
stop_sequences: vec![],
|
||||
seed: 0,
|
||||
|
||||
@@ -8,6 +8,7 @@
|
||||
//! The session borrows raw pointers from InferenceEngine and must not outlive it.
|
||||
//! In Phase 7+ this will be refactored to use a shared context pool.
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::ffi::{CStr, CString};
|
||||
use std::ops::Range;
|
||||
use std::sync::Arc;
|
||||
@@ -20,6 +21,37 @@ use llama_sys::{llama_context, llama_token, llama_vocab, LlamaLib};
|
||||
|
||||
use crate::sampler::SamplerConfig;
|
||||
|
||||
// ─── TPS Degradation Monitor ───────────────────────────────────────────────
|
||||
|
||||
/// Degradation state for the current generation session.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum DegradationState {
|
||||
/// TPS is acceptable.
|
||||
Normal,
|
||||
/// TPS has been below threshold for the required duration.
|
||||
Degraded,
|
||||
}
|
||||
|
||||
/// Configuration for TPS degradation detection.
|
||||
pub struct TpsDegradationConfig {
|
||||
/// TPS threshold below which degradation is suspected.
|
||||
pub tps_threshold: f64,
|
||||
/// How long (in seconds) TPS must stay below threshold to trigger degradation.
|
||||
pub sustained_seconds: f64,
|
||||
/// Rolling window size in number of samples.
|
||||
pub window_size: usize,
|
||||
}
|
||||
|
||||
impl Default for TpsDegradationConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
tps_threshold: 30.0,
|
||||
sustained_seconds: 5.0,
|
||||
window_size: 20,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ─── GenerationSession ─────────────────────────────────────────────────────
|
||||
|
||||
pub struct GenerationSession {
|
||||
@@ -40,6 +72,13 @@ pub struct GenerationSession {
|
||||
pub prefill_ms: Option<f64>,
|
||||
gen_start: Option<Instant>,
|
||||
pub tokens_per_sec: f64,
|
||||
|
||||
// TPS degradation monitor
|
||||
degradation_config: TpsDegradationConfig,
|
||||
tps_window: VecDeque<(Instant, f64)>, // (timestamp, instantaneous_tps)
|
||||
pub degradation_state: DegradationState,
|
||||
degradation_emitted: bool, // only emit once per session
|
||||
last_token_time: Option<Instant>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
@@ -121,7 +160,18 @@ impl GenerationSession {
|
||||
let s = (lib.sampler_init_greedy)();
|
||||
(lib.sampler_chain_add)(sampler_chain, s);
|
||||
} else {
|
||||
// top-k → temp → top-p → dist (for randomness)
|
||||
// penalties → top-k → temp → top-p → dist (for randomness)
|
||||
// Apply repetition penalty first to prevent loops
|
||||
if sampler_cfg.repeat_penalty > 1.0 {
|
||||
let s_penalties = (lib.sampler_init_penalties)(
|
||||
sampler_cfg.repeat_last_n,
|
||||
sampler_cfg.repeat_penalty,
|
||||
0.0, // frequency_penalty (not used)
|
||||
0.0, // presence_penalty (not used)
|
||||
);
|
||||
(lib.sampler_chain_add)(sampler_chain, s_penalties);
|
||||
}
|
||||
|
||||
let s_topk = (lib.sampler_init_top_k)(sampler_cfg.top_k);
|
||||
(lib.sampler_chain_add)(sampler_chain, s_topk);
|
||||
|
||||
@@ -148,7 +198,7 @@ impl GenerationSession {
|
||||
let prefill_tps = n_prompt_tokens as f64 / (prefill_ms / 1000.0);
|
||||
info!(
|
||||
n_prompt_tokens,
|
||||
prefill_ms = format_args!("{:.2}", prefill_ms),
|
||||
prefill_ms = format_args!("{:.2}", prefill_ms),
|
||||
prefill_tps = format_args!("{:.2}", prefill_tps),
|
||||
"session.prefill_complete"
|
||||
);
|
||||
@@ -168,6 +218,11 @@ impl GenerationSession {
|
||||
prefill_ms: Some(prefill_ms),
|
||||
gen_start: None,
|
||||
tokens_per_sec: 0.0,
|
||||
degradation_config: TpsDegradationConfig::default(),
|
||||
tps_window: VecDeque::new(),
|
||||
degradation_state: DegradationState::Normal,
|
||||
degradation_emitted: false,
|
||||
last_token_time: None,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -214,10 +269,27 @@ impl GenerationSession {
|
||||
self.n_decoded += 1;
|
||||
self.tokens_generated += 1;
|
||||
|
||||
// ── TPS degradation monitoring ─────────────────────────────────
|
||||
let now = Instant::now();
|
||||
if let Some(last) = self.last_token_time.replace(now) {
|
||||
let elapsed = now.duration_since(last).as_secs_f64();
|
||||
if elapsed > 0.0 {
|
||||
let instant_tps = 1.0 / elapsed;
|
||||
self.tps_window.push_back((now, instant_tps));
|
||||
// Keep window bounded
|
||||
while self.tps_window.len() > self.degradation_config.window_size {
|
||||
self.tps_window.pop_front();
|
||||
}
|
||||
self.check_degradation(now);
|
||||
}
|
||||
} else {
|
||||
self.last_token_time = Some(now);
|
||||
}
|
||||
|
||||
if self.tokens_generated % 100 == 0 {
|
||||
tracing::debug!(
|
||||
tokens_generated = self.tokens_generated,
|
||||
max_tokens = self.max_tokens,
|
||||
max_tokens = self.max_tokens,
|
||||
"session.generation_progress"
|
||||
);
|
||||
}
|
||||
@@ -254,9 +326,9 @@ impl GenerationSession {
|
||||
};
|
||||
info!(
|
||||
tokens_generated = self.tokens_generated,
|
||||
max_tokens = self.max_tokens,
|
||||
tps = self.tokens_per_sec,
|
||||
prefill_ms = self.prefill_ms.unwrap_or(0.0),
|
||||
max_tokens = self.max_tokens,
|
||||
tps = self.tokens_per_sec,
|
||||
prefill_ms = self.prefill_ms.unwrap_or(0.0),
|
||||
stop_reason,
|
||||
"session.generation_complete"
|
||||
);
|
||||
@@ -278,6 +350,43 @@ impl GenerationSession {
|
||||
self.prefill_ms
|
||||
}
|
||||
|
||||
pub fn degradation_state(&self) -> &DegradationState {
|
||||
&self.degradation_state
|
||||
}
|
||||
|
||||
fn check_degradation(&mut self, now: Instant) {
|
||||
if self.degradation_emitted || self.degradation_state == DegradationState::Degraded {
|
||||
return;
|
||||
}
|
||||
if self.tps_window.len() < 3 {
|
||||
return; // not enough samples yet
|
||||
}
|
||||
|
||||
// Check if all samples in window are below threshold
|
||||
let all_below = self
|
||||
.tps_window
|
||||
.iter()
|
||||
.all(|(_, tps)| *tps < self.degradation_config.tps_threshold);
|
||||
if !all_below {
|
||||
return;
|
||||
}
|
||||
|
||||
// Check if the window spans at least sustained_seconds
|
||||
if let Some((oldest_time, _)) = self.tps_window.front() {
|
||||
let span = now.duration_since(*oldest_time).as_secs_f64();
|
||||
if span >= self.degradation_config.sustained_seconds {
|
||||
self.degradation_state = DegradationState::Degraded;
|
||||
self.degradation_emitted = true;
|
||||
tracing::warn!(
|
||||
tps_window_size = self.tps_window.len(),
|
||||
span_secs = span,
|
||||
threshold = self.degradation_config.tps_threshold,
|
||||
"session.tps_degraded: sustained low TPS detected"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn detokenize_token(&self, token: llama_token) -> String {
|
||||
if let Some(token_to_piece) = self.lib.token_to_piece {
|
||||
return decode_token_piece_with(|buf, len| unsafe {
|
||||
@@ -445,6 +554,43 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tps_degradation_config_has_correct_defaults() {
|
||||
let config = super::TpsDegradationConfig::default();
|
||||
assert_eq!(config.tps_threshold, 30.0);
|
||||
assert_eq!(config.sustained_seconds, 5.0);
|
||||
assert_eq!(config.window_size, 20);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn degradation_state_starts_normal() {
|
||||
assert_eq!(
|
||||
super::DegradationState::Normal,
|
||||
super::DegradationState::Normal
|
||||
);
|
||||
assert_ne!(
|
||||
super::DegradationState::Normal,
|
||||
super::DegradationState::Degraded
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn degradation_check_requires_sustained_window() {
|
||||
use std::collections::VecDeque;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
// Build a window of low-TPS samples spanning < 5s → should NOT trigger
|
||||
let now = Instant::now();
|
||||
let window: VecDeque<(Instant, f64)> = (0..5)
|
||||
.map(|i| (now - Duration::from_millis(i * 100), 10.0))
|
||||
.collect();
|
||||
|
||||
// Span is only 0.4s — below 5s threshold
|
||||
let oldest = window.front().map(|(t, _)| *t).unwrap();
|
||||
let span = now.duration_since(oldest).as_secs_f64();
|
||||
assert!(span < 5.0, "span should be less than 5s: {span}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn prefill_plan_splits_large_prompt_and_only_requests_final_logits() {
|
||||
let plan = build_prefill_plan(300, 128);
|
||||
|
||||
Reference in new issue
Block a user