use core::fmt;
use core::time::Duration;
use crate::admission::StopReason;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Seam {
TalkerLogits,
MicrodecoderLogits,
CodecOutput,
Pcm,
}
impl Seam {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::TalkerLogits => "talker_logits",
Self::MicrodecoderLogits => "microdecoder_logits",
Self::CodecOutput => "codec_output",
Self::Pcm => "pcm",
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum SeamPolicy {
Off,
Sampled { every: u32 },
All,
}
impl SeamPolicy {
#[must_use]
pub const fn is_checking(self) -> bool {
!matches!(self, Self::Off)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum HealthViolation {
NonFinite {
seam: Seam,
index: usize,
is_nan: bool,
},
NoProgress {
frames_emitted: u64,
stalled_millis: u64,
},
StopInconsistent {
claimed: StopReason,
frames_emitted: u64,
frame_cap: u64,
},
RepetitionRunaway { token: u32, repeats: u32 },
OutputSilent { silent_millis: u64 },
KernelDemoted { from: KernelTier, to: KernelTier },
ThermalDegraded {
percent_below_baseline: u32,
},
}
impl HealthViolation {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::NonFinite { .. } => "non_finite",
Self::NoProgress { .. } => "no_progress",
Self::StopInconsistent { .. } => "stop_inconsistent",
Self::RepetitionRunaway { .. } => "repetition_runaway",
Self::OutputSilent { .. } => "output_silent",
Self::KernelDemoted { .. } => "kernel_demoted",
Self::ThermalDegraded { .. } => "thermal_degraded",
}
}
#[must_use]
pub const fn invalidates_output(self) -> bool {
!matches!(
self,
Self::KernelDemoted { .. } | Self::ThermalDegraded { .. }
)
}
#[must_use]
pub const fn remedy(self) -> &'static str {
match self {
Self::NonFinite { .. } => {
"a non-finite value reached this seam: rerun with FTTS_MATH_MODE=strict; if it \
persists there, the fault is in the kernel rather than a fast-math approximation"
}
Self::NoProgress { .. } => {
"generation stopped advancing: cancel and retry; if reproducible, capture the \
prompt — a stalled decode loop is a bug, not a capacity problem"
}
Self::StopInconsistent { .. } => {
"the stop reason disagrees with the frame counters; treat this result as \
untrusted and report it — one of the two is lying about whether audio was cut off"
}
Self::RepetitionRunaway { .. } => {
"the model entered a repetition loop: raise the repetition penalty or shorten the \
input; the audio to this point is usable, everything after the loop began is not"
}
Self::OutputSilent { .. } => {
"output was silent past the allowed window: check the voice pack and reference \
audio; a silent result is a failure even though it produced bytes"
}
Self::KernelDemoted { .. } => {
"an optimised kernel failed its selftest and the certified scalar path took over: \
results stay correct and slower; report the ISA and CPU"
}
Self::ThermalDegraded { .. } => {
"sustained throughput fell below the opening window: expected under thermal load; \
do not quote this run's rate as a steady-state number"
}
}
}
}
impl fmt::Display for HealthViolation {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::NonFinite {
seam,
index,
is_nan,
} => write!(
formatter,
"{} at {} index {index}",
if *is_nan { "NaN" } else { "infinity" },
seam.as_str()
),
Self::NoProgress {
frames_emitted,
stalled_millis,
} => write!(
formatter,
"no frame progress for {stalled_millis} ms after {frames_emitted} frame(s)"
),
Self::StopInconsistent {
claimed,
frames_emitted,
frame_cap,
} => write!(
formatter,
"stop reason {} contradicts {frames_emitted} frame(s) against a cap of {frame_cap}",
claimed.as_str()
),
Self::RepetitionRunaway { token, repeats } => {
write!(formatter, "token {token} repeated {repeats} times")
}
Self::OutputSilent { silent_millis } => {
write!(formatter, "output silent for {silent_millis} ms")
}
Self::KernelDemoted { from, to } => write!(
formatter,
"kernel demoted from {} to {}",
from.as_str(),
to.as_str()
),
Self::ThermalDegraded {
percent_below_baseline,
} => write!(
formatter,
"throughput {percent_below_baseline}% below the opening window"
),
}
}
}
#[derive(Clone, Debug)]
pub struct NumericGuard {
policy: SeamPolicy,
calls: u32,
}
impl NumericGuard {
#[must_use]
pub const fn new(policy: SeamPolicy) -> Self {
Self { policy, calls: 0 }
}
#[must_use]
pub const fn policy(&self) -> SeamPolicy {
self.policy
}
pub fn check(&mut self, seam: Seam, values: &[f32]) -> Result<(), HealthViolation> {
if !self.should_check() {
return Ok(());
}
for (index, value) in values.iter().enumerate() {
if !value.is_finite() {
return Err(HealthViolation::NonFinite {
seam,
index,
is_nan: value.is_nan(),
});
}
}
Ok(())
}
fn should_check(&mut self) -> bool {
match self.policy {
SeamPolicy::Off => false,
SeamPolicy::All => true,
SeamPolicy::Sampled { every } => {
if every <= 1 {
return true;
}
let due = self.calls.is_multiple_of(every);
self.calls = self.calls.wrapping_add(1);
due
}
}
}
}
#[derive(Clone, Debug)]
pub struct ProgressWatchdog<T> {
timeout: Duration,
last_progress: T,
frames_emitted: u64,
}
impl<T: Copy + core::ops::Sub<T, Output = Duration>> ProgressWatchdog<T> {
#[must_use]
pub const fn new(timeout: Duration, started: T) -> Self {
Self {
timeout,
last_progress: started,
frames_emitted: 0,
}
}
pub fn record_frame(&mut self, now: T) {
self.frames_emitted += 1;
self.last_progress = now;
}
#[must_use]
pub const fn frames_emitted(&self) -> u64 {
self.frames_emitted
}
pub fn check(&self, now: T) -> Result<(), HealthViolation> {
let stalled = now - self.last_progress;
if stalled > self.timeout {
return Err(HealthViolation::NoProgress {
frames_emitted: self.frames_emitted,
stalled_millis: u64::try_from(stalled.as_millis()).unwrap_or(u64::MAX),
});
}
Ok(())
}
}
pub fn check_stop_consistency(
claimed: StopReason,
frames_emitted: u64,
frame_cap: u64,
) -> Result<(), HealthViolation> {
let inconsistent = match claimed {
StopReason::EndOfSpeech => frames_emitted >= frame_cap,
StopReason::FrameCapReached => frames_emitted < frame_cap,
StopReason::DurationLimitReached | StopReason::Cancelled => false,
};
if inconsistent {
return Err(HealthViolation::StopInconsistent {
claimed,
frames_emitted,
frame_cap,
});
}
Ok(())
}
#[derive(Clone, Debug)]
pub struct RunawayDetector {
max_consecutive: u32,
max_cycle_repeats: u32,
last: Option<u32>,
consecutive: u32,
recent: [u32; Self::CYCLE_WINDOW],
filled: usize,
cycle_repeats: u32,
}
impl RunawayDetector {
const CYCLE_WINDOW: usize = 8;
#[must_use]
pub const fn new(max_consecutive: u32, max_cycle_repeats: u32) -> Self {
Self {
max_consecutive,
max_cycle_repeats,
last: None,
consecutive: 0,
recent: [u32::MAX; Self::CYCLE_WINDOW],
filled: 0,
cycle_repeats: 0,
}
}
pub fn observe(&mut self, token: u32) -> Result<(), HealthViolation> {
if self.last == Some(token) {
self.consecutive += 1;
} else {
self.consecutive = 1;
self.last = Some(token);
}
if self.consecutive > self.max_consecutive {
return Err(HealthViolation::RepetitionRunaway {
token,
repeats: self.consecutive,
});
}
if self.filled >= 2 && self.recent[(self.filled - 2) % Self::CYCLE_WINDOW] == token {
self.cycle_repeats += 1;
if self.cycle_repeats > self.max_cycle_repeats {
return Err(HealthViolation::RepetitionRunaway {
token,
repeats: self.cycle_repeats,
});
}
} else {
self.cycle_repeats = 0;
}
self.recent[self.filled % Self::CYCLE_WINDOW] = token;
self.filled += 1;
Ok(())
}
}
#[derive(Clone, Debug)]
pub struct SilenceDetector {
floor: i16,
max_silent_samples: u64,
sample_rate: u32,
silent_samples: u64,
}
impl SilenceDetector {
#[must_use]
pub const fn new(floor: i16, max_silent: Duration, sample_rate: u32) -> Self {
Self {
floor,
max_silent_samples: (max_silent.as_millis() as u64) * (sample_rate as u64) / 1000,
sample_rate,
silent_samples: 0,
}
}
pub fn observe(&mut self, samples: &[i16]) -> Result<(), HealthViolation> {
for sample in samples {
if sample.saturating_abs() > self.floor {
self.silent_samples = 0;
} else {
self.silent_samples += 1;
}
}
if self.sample_rate == 0 {
return Ok(());
}
if self.silent_samples > self.max_silent_samples {
return Err(HealthViolation::OutputSilent {
silent_millis: self.silent_millis(),
});
}
Ok(())
}
#[must_use]
pub const fn silent_millis(&self) -> u64 {
if self.sample_rate == 0 {
return 0;
}
self.silent_samples * 1000 / self.sample_rate as u64
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum KernelTier {
Optimized(&'static str),
Scalar,
}
impl KernelTier {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Optimized(name) => name,
Self::Scalar => "scalar",
}
}
}
#[derive(Clone, Debug)]
pub struct KernelSelector {
preferred: KernelTier,
active: KernelTier,
demoted: bool,
}
impl KernelSelector {
#[must_use]
pub const fn new(preferred: KernelTier) -> Self {
Self {
preferred,
active: preferred,
demoted: false,
}
}
#[must_use]
pub const fn active(&self) -> KernelTier {
self.active
}
#[must_use]
pub const fn demoted(&self) -> bool {
self.demoted
}
pub fn on_selftest_failure(&mut self) -> Option<HealthViolation> {
if self.demoted || matches!(self.active, KernelTier::Scalar) {
self.demoted = true;
self.active = KernelTier::Scalar;
return None;
}
let from = self.active;
self.active = KernelTier::Scalar;
self.demoted = true;
Some(HealthViolation::KernelDemoted {
from,
to: KernelTier::Scalar,
})
}
#[must_use]
pub const fn preferred(&self) -> KernelTier {
self.preferred
}
}
#[derive(Clone, Debug)]
pub struct ThermalReporter {
baseline: Option<f64>,
latest: Option<f64>,
report_below_percent: u32,
}
impl ThermalReporter {
#[must_use]
pub const fn new(report_below_percent: u32) -> Self {
Self {
baseline: None,
latest: None,
report_below_percent,
}
}
pub fn observe(&mut self, throughput: f64) -> Option<HealthViolation> {
if !throughput.is_finite() || throughput <= 0.0 {
return None;
}
self.latest = Some(throughput);
let baseline = *self.baseline.get_or_insert(throughput);
if throughput >= baseline {
return None;
}
let percent = ((baseline - throughput) / baseline * 100.0).floor();
let percent = percent.clamp(0.0, f64::from(u32::MAX)) as u32;
if percent >= self.report_below_percent {
return Some(HealthViolation::ThermalDegraded {
percent_below_baseline: percent,
});
}
None
}
#[must_use]
pub const fn baseline(&self) -> Option<f64> {
self.baseline
}
#[must_use]
pub const fn latest(&self) -> Option<f64> {
self.latest
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Instant;
#[test]
fn a_nan_is_located_at_its_first_index() {
let mut guard = NumericGuard::new(SeamPolicy::All);
let values = [1.0, 2.0, f32::NAN, f32::NAN];
let violation = guard
.check(Seam::TalkerLogits, &values)
.expect_err("NaN must be caught");
assert_eq!(
violation,
HealthViolation::NonFinite {
seam: Seam::TalkerLogits,
index: 2,
is_nan: true,
}
);
assert!(violation.invalidates_output());
}
#[test]
fn an_infinity_is_distinguished_from_a_nan() {
let mut guard = NumericGuard::new(SeamPolicy::All);
let violation = guard
.check(Seam::CodecOutput, &[f32::INFINITY])
.expect_err("infinity must be caught");
assert!(matches!(
violation,
HealthViolation::NonFinite { is_nan: false, .. }
));
}
#[test]
fn policy_off_never_looks_and_says_so() {
let mut guard = NumericGuard::new(SeamPolicy::Off);
assert!(guard.check(Seam::Pcm, &[f32::NAN]).is_ok());
assert!(!guard.policy().is_checking());
}
#[test]
fn a_zero_sampling_interval_checks_rather_than_disables() {
let mut guard = NumericGuard::new(SeamPolicy::Sampled { every: 0 });
assert!(guard.check(Seam::Pcm, &[f32::NAN]).is_err());
}
#[test]
fn sampling_checks_periodically() {
let mut guard = NumericGuard::new(SeamPolicy::Sampled { every: 3 });
let bad = [f32::NAN];
assert!(guard.check(Seam::Pcm, &bad).is_err(), "first call checks");
assert!(guard.check(Seam::Pcm, &bad).is_ok(), "second is skipped");
assert!(guard.check(Seam::Pcm, &bad).is_ok(), "third is skipped");
assert!(guard.check(Seam::Pcm, &bad).is_err(), "fourth checks again");
}
#[test]
fn the_watchdog_fires_only_after_the_timeout() {
let start = Instant::now();
let watchdog = ProgressWatchdog::new(Duration::from_millis(500), start);
assert!(watchdog.check(start + Duration::from_millis(499)).is_ok());
let violation = watchdog
.check(start + Duration::from_millis(501))
.expect_err("a stall past the timeout must fire");
assert!(matches!(
violation,
HealthViolation::NoProgress {
frames_emitted: 0,
..
}
));
}
#[test]
fn recording_a_frame_resets_the_stall_timer() {
let start = Instant::now();
let mut watchdog = ProgressWatchdog::new(Duration::from_millis(100), start);
let later = start + Duration::from_millis(90);
watchdog.record_frame(later);
assert!(watchdog.check(later + Duration::from_millis(90)).is_ok());
assert_eq!(watchdog.frames_emitted(), 1);
}
#[test]
fn end_of_speech_on_the_cap_is_reported_as_inconsistent() {
let violation = check_stop_consistency(StopReason::EndOfSpeech, 2048, 2048)
.expect_err("EOS exactly on the cap must be challenged");
assert!(matches!(
violation,
HealthViolation::StopInconsistent { .. }
));
assert!(violation.invalidates_output());
}
#[test]
fn a_cap_stop_short_of_the_cap_is_inconsistent() {
assert!(check_stop_consistency(StopReason::FrameCapReached, 100, 2048).is_err());
}
#[test]
fn consistent_outcomes_pass() {
assert!(check_stop_consistency(StopReason::EndOfSpeech, 100, 2048).is_ok());
assert!(check_stop_consistency(StopReason::FrameCapReached, 2048, 2048).is_ok());
assert!(check_stop_consistency(StopReason::Cancelled, 7, 2048).is_ok());
assert!(check_stop_consistency(StopReason::DurationLimitReached, 7, 2048).is_ok());
}
#[test]
fn a_stuck_token_trips_the_runaway_detector() {
let mut detector = RunawayDetector::new(4, 8);
for _ in 0..4 {
detector.observe(42).expect("within threshold");
}
let violation = detector
.observe(42)
.expect_err("the fifth repeat must trip");
assert!(matches!(
violation,
HealthViolation::RepetitionRunaway {
token: 42,
repeats: 5
}
));
}
#[test]
fn a_two_token_cycle_trips_even_though_nothing_repeats_consecutively() {
let mut detector = RunawayDetector::new(100, 3);
let mut result = Ok(());
for index in 0..12 {
result = detector.observe(if index % 2 == 0 { 7 } else { 9 });
if result.is_err() {
break;
}
}
assert!(result.is_err(), "a ping-pong cycle must be detected");
}
#[test]
fn ordinary_variety_does_not_trip_the_detector() {
let mut detector = RunawayDetector::new(4, 3);
for token in 0..64u32 {
detector.observe(token).expect("varied tokens are healthy");
}
}
#[test]
fn silence_past_the_window_is_a_violation() {
let mut detector = SilenceDetector::new(4, Duration::from_millis(100), 24_000);
let silent = vec![0i16; 2_401];
let violation = detector
.observe(&silent)
.expect_err("silence past the window must fire");
assert!(matches!(violation, HealthViolation::OutputSilent { .. }));
}
#[test]
fn any_audible_sample_resets_the_silence_run() {
let mut detector = SilenceDetector::new(4, Duration::from_millis(100), 24_000);
detector.observe(&vec![0i16; 2_000]).expect("under window");
detector.observe(&[9_000]).expect("audible sample resets");
assert_eq!(detector.silent_millis(), 0);
detector
.observe(&vec![0i16; 2_000])
.expect("run restarted, so still under the window");
}
#[test]
fn demotion_is_one_way_and_reported_once() {
let mut selector = KernelSelector::new(KernelTier::Optimized("i8mm"));
assert_eq!(selector.active(), KernelTier::Optimized("i8mm"));
let violation = selector
.on_selftest_failure()
.expect("the first demotion is reported");
assert_eq!(
violation,
HealthViolation::KernelDemoted {
from: KernelTier::Optimized("i8mm"),
to: KernelTier::Scalar,
}
);
assert!(!violation.invalidates_output());
assert_eq!(selector.active(), KernelTier::Scalar);
assert!(selector.on_selftest_failure().is_none());
assert_eq!(selector.active(), KernelTier::Scalar);
assert!(selector.demoted());
assert_eq!(selector.preferred(), KernelTier::Optimized("i8mm"));
}
#[test]
fn thermal_decline_is_reported_against_the_opening_window() {
let mut reporter = ThermalReporter::new(10);
assert!(
reporter.observe(20.0).is_none(),
"the first sample is the baseline"
);
assert!(
reporter.observe(19.0).is_none(),
"5% is under the threshold"
);
let violation = reporter
.observe(17.0)
.expect("15% below baseline must be reported");
assert_eq!(
violation,
HealthViolation::ThermalDegraded {
percent_below_baseline: 15
}
);
assert!(!violation.invalidates_output());
assert_eq!(reporter.baseline(), Some(20.0));
assert_eq!(reporter.latest(), Some(17.0));
}
#[test]
fn a_nonsense_throughput_sample_is_ignored_rather_than_becoming_the_baseline() {
let mut reporter = ThermalReporter::new(10);
assert!(reporter.observe(0.0).is_none());
assert!(reporter.observe(f64::NAN).is_none());
assert_eq!(reporter.baseline(), None, "no baseline was established");
assert!(reporter.observe(10.0).is_none());
assert_eq!(reporter.baseline(), Some(10.0));
}
#[test]
fn every_violation_carries_a_remedy_and_a_wire_name() {
let violations = [
HealthViolation::NonFinite {
seam: Seam::Pcm,
index: 0,
is_nan: true,
},
HealthViolation::NoProgress {
frames_emitted: 1,
stalled_millis: 2,
},
HealthViolation::StopInconsistent {
claimed: StopReason::EndOfSpeech,
frames_emitted: 1,
frame_cap: 1,
},
HealthViolation::RepetitionRunaway {
token: 1,
repeats: 2,
},
HealthViolation::OutputSilent { silent_millis: 1 },
HealthViolation::KernelDemoted {
from: KernelTier::Optimized("i8mm"),
to: KernelTier::Scalar,
},
HealthViolation::ThermalDegraded {
percent_below_baseline: 11,
},
];
for violation in violations {
assert!(!violation.as_str().is_empty());
assert!(
violation.remedy().len() > 40,
"{}: a remedy must tell the caller what to do",
violation.as_str()
);
assert!(!violation.to_string().is_empty());
}
}
}