agentos_runtime/
metrics.rs

1//! Fixed-cardinality runtime telemetry and bounded stderr fallback formatting.
2
3use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
4use std::sync::Arc;
5use std::time::Duration;
6
7use crate::supervision::{TaskClass, TaskTerminalReason};
8
9const ORDERING: Ordering = Ordering::Relaxed;
10pub const TASK_CLASS_COUNT: usize = 10;
11pub const TASK_TERMINAL_REASON_COUNT: usize = 4;
12pub const MAX_FALLBACK_MESSAGE_BYTES: usize = 512;
13
14pub const TASK_CLASSES: [TaskClass; TASK_CLASS_COUNT] = [
15    TaskClass::Runtime,
16    TaskClass::Dns,
17    TaskClass::Socket,
18    TaskClass::Listener,
19    TaskClass::Udp,
20    TaskClass::Tls,
21    TaskClass::Http2,
22    TaskClass::Timer,
23    TaskClass::Vm,
24    TaskClass::Plugin,
25];
26
27pub const TASK_TERMINAL_REASONS: [TaskTerminalReason; TASK_TERMINAL_REASON_COUNT] = [
28    TaskTerminalReason::Completed,
29    TaskTerminalReason::Cancelled,
30    TaskTerminalReason::Failed,
31    TaskTerminalReason::Panicked,
32];
33
34macro_rules! fixed_metric_enum {
35    ($name:ident, $count:ident, [$($variant:ident),+ $(,)?]) => {
36        #[repr(usize)]
37        #[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
38        pub enum $name {
39            $($variant),+
40        }
41
42        impl $name {
43            pub const ALL: [Self; $count] = [$(Self::$variant),+];
44
45            pub const fn index(self) -> usize {
46                self as usize
47            }
48        }
49    };
50}
51
52pub const RESOURCE_METRIC_CLASS_COUNT: usize = 13;
53fixed_metric_enum!(
54    ResourceMetricClass,
55    RESOURCE_METRIC_CLASS_COUNT,
56    [
57        ActiveVms,
58        Capabilities,
59        ReadyHandles,
60        Sockets,
61        Connections,
62        Timers,
63        Tasks,
64        BridgeCalls,
65        HandleCommands,
66        AsyncCompletions,
67        Datagrams,
68        Http2Connections,
69        Http2Streams,
70    ]
71);
72
73pub const BUFFER_METRIC_CLASS_COUNT: usize = 8;
74fixed_metric_enum!(
75    BufferMetricClass,
76    BUFFER_METRIC_CLASS_COUNT,
77    [Kernel, Native, Tls, Http2, Bridge, Datagram, Executor, Guest]
78);
79
80pub const WAKE_METRIC_COUNT: usize = 4;
81fixed_metric_enum!(
82    WakeMetric,
83    WAKE_METRIC_COUNT,
84    [Attempted, Coalesced, Delivered, Rearmed]
85);
86
87pub const FAIRNESS_LEVEL_COUNT: usize = 6;
88fixed_metric_enum!(
89    FairnessLevel,
90    FAIRNESS_LEVEL_COUNT,
91    [
92        Process,
93        Vm,
94        Capability,
95        Http2Stream,
96        BridgeCompletion,
97        Signal
98    ]
99);
100
101pub const COMPLETION_ANOMALY_COUNT: usize = 3;
102fixed_metric_enum!(
103    CompletionAnomaly,
104    COMPLETION_ANOMALY_COUNT,
105    [Stale, Duplicate, Late]
106);
107
108pub const CHANNEL_METRIC_CLASS_COUNT: usize = 10;
109fixed_metric_enum!(
110    ChannelMetricClass,
111    CHANNEL_METRIC_CLASS_COUNT,
112    [
113        ReadyWake,
114        HandleCommand,
115        BridgeResponse,
116        BridgeEvent,
117        AsyncCompletion,
118        Signal,
119        Datagram,
120        Http2,
121        StdioIngress,
122        StdioEgress,
123    ]
124);
125
126pub const EXECUTOR_METRIC_CLASS_COUNT: usize = 3;
127fixed_metric_enum!(
128    ExecutorMetricClass,
129    EXECUTOR_METRIC_CLASS_COUNT,
130    [Runtime, Vm, Blocking]
131);
132
133pub const WATCHDOG_METRIC_COUNT: usize = 4;
134fixed_metric_enum!(
135    WatchdogMetric,
136    WATCHDOG_METRIC_COUNT,
137    [
138        LongTaskPoll,
139        NonYieldingTask,
140        RuntimeWorkerStall,
141        ExecutorStall
142    ]
143);
144
145pub const FALLBACK_SEVERITY_COUNT: usize = 2;
146fixed_metric_enum!(TelemetrySeverity, FALLBACK_SEVERITY_COUNT, [Warning, Fatal]);
147
148#[derive(Clone, Copy, Debug, Eq, PartialEq)]
149pub enum TelemetryFallbackCode {
150    ResourceLimit,
151    Overloaded,
152    SupervisedTaskExit,
153    RuntimeWorkerStall,
154    TelemetryUnavailable,
155}
156
157impl TelemetryFallbackCode {
158    const fn as_str(self) -> &'static str {
159        match self {
160            Self::ResourceLimit => "ERR_AGENTOS_RESOURCE_LIMIT",
161            Self::Overloaded => "ERR_AGENTOS_OVERLOADED",
162            Self::SupervisedTaskExit => "ERR_AGENTOS_SUPERVISED_TASK_EXIT",
163            Self::RuntimeWorkerStall => "ERR_AGENTOS_RUNTIME_WORKER_STALL",
164            Self::TelemetryUnavailable => "ERR_AGENTOS_TELEMETRY_UNAVAILABLE",
165        }
166    }
167}
168
169#[derive(Clone, Copy, Debug, Eq, PartialEq)]
170pub enum TelemetrySubsystem {
171    Runtime,
172    Reactor,
173    Bridge,
174    Executor,
175    Telemetry,
176}
177
178impl TelemetrySubsystem {
179    const fn as_str(self) -> &'static str {
180        match self {
181            Self::Runtime => "runtime",
182            Self::Reactor => "reactor",
183            Self::Bridge => "bridge",
184            Self::Executor => "executor",
185            Self::Telemetry => "telemetry",
186        }
187    }
188}
189
190impl TelemetrySeverity {
191    const fn as_str(self) -> &'static str {
192        match self {
193            Self::Warning => "warning",
194            Self::Fatal => "fatal",
195        }
196    }
197}
198
199#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
200pub struct GaugeSnapshot {
201    pub current: usize,
202    pub high_water: usize,
203}
204
205#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
206pub struct TaskMetricSnapshot {
207    pub active: usize,
208    pub terminal: [u64; TASK_TERMINAL_REASON_COUNT],
209}
210
211impl TaskMetricSnapshot {
212    pub fn terminal_count(self, reason: TaskTerminalReason) -> u64 {
213        self.terminal[task_terminal_reason_index(reason)]
214    }
215}
216
217#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
218pub struct ReadinessMetricSnapshot {
219    pub current_size: usize,
220    pub size_high_water: usize,
221    pub age_samples: u64,
222    pub total_oldest_age_micros: u64,
223    pub max_oldest_age_micros: u64,
224}
225
226#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
227pub struct ChannelMetricSnapshot {
228    pub count_high_water: usize,
229    pub byte_high_water: usize,
230}
231
232#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
233pub struct ExecutorMetricSnapshot {
234    pub active: GaugeSnapshot,
235    pub queued: GaugeSnapshot,
236}
237
238#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
239pub struct WatchdogMetricSnapshot {
240    pub events: u64,
241    pub total_stall_micros: u64,
242    pub max_stall_micros: u64,
243}
244
245#[derive(Clone, Debug, Eq, PartialEq)]
246pub struct RuntimeMetricsSnapshot {
247    pub resources: [GaugeSnapshot; RESOURCE_METRIC_CLASS_COUNT],
248    pub buffers: [GaugeSnapshot; BUFFER_METRIC_CLASS_COUNT],
249    pub tasks: [TaskMetricSnapshot; TASK_CLASS_COUNT],
250    pub wakes: [u64; WAKE_METRIC_COUNT],
251    pub readiness: ReadinessMetricSnapshot,
252    pub fairness_yields: [u64; FAIRNESS_LEVEL_COUNT],
253    pub completion_anomalies: [u64; COMPLETION_ANOMALY_COUNT],
254    pub channels: [ChannelMetricSnapshot; CHANNEL_METRIC_CLASS_COUNT],
255    pub executors: [ExecutorMetricSnapshot; EXECUTOR_METRIC_CLASS_COUNT],
256    pub watchdogs: [WatchdogMetricSnapshot; WATCHDOG_METRIC_COUNT],
257    pub stderr_fallbacks: [u64; FALLBACK_SEVERITY_COUNT],
258}
259
260impl RuntimeMetricsSnapshot {
261    pub fn task(&self, class: TaskClass) -> TaskMetricSnapshot {
262        self.tasks[task_class_index(class)]
263    }
264}
265
266#[derive(Debug)]
267struct AtomicGauge {
268    current: AtomicUsize,
269    high_water: AtomicUsize,
270}
271
272impl AtomicGauge {
273    fn new() -> Self {
274        Self {
275            current: AtomicUsize::new(0),
276            high_water: AtomicUsize::new(0),
277        }
278    }
279
280    fn observe(&self, current: usize) {
281        self.current.store(current, ORDERING);
282        saturating_fetch_max_usize(&self.high_water, current);
283    }
284
285    fn snapshot(&self) -> GaugeSnapshot {
286        GaugeSnapshot {
287            current: self.current.load(ORDERING),
288            high_water: self.high_water.load(ORDERING),
289        }
290    }
291}
292
293#[derive(Debug)]
294struct AtomicTaskMetric {
295    active: AtomicUsize,
296    terminal: [AtomicU64; TASK_TERMINAL_REASON_COUNT],
297}
298
299impl AtomicTaskMetric {
300    fn new() -> Self {
301        Self {
302            active: AtomicUsize::new(0),
303            terminal: std::array::from_fn(|_| AtomicU64::new(0)),
304        }
305    }
306}
307
308#[derive(Debug)]
309struct AtomicReadinessMetric {
310    size: AtomicGauge,
311    age_samples: AtomicU64,
312    total_oldest_age_micros: AtomicU64,
313    max_oldest_age_micros: AtomicU64,
314}
315
316impl AtomicReadinessMetric {
317    fn new() -> Self {
318        Self {
319            size: AtomicGauge::new(),
320            age_samples: AtomicU64::new(0),
321            total_oldest_age_micros: AtomicU64::new(0),
322            max_oldest_age_micros: AtomicU64::new(0),
323        }
324    }
325}
326
327#[derive(Debug)]
328struct AtomicChannelMetric {
329    count_high_water: AtomicUsize,
330    byte_high_water: AtomicUsize,
331}
332
333impl AtomicChannelMetric {
334    fn new() -> Self {
335        Self {
336            count_high_water: AtomicUsize::new(0),
337            byte_high_water: AtomicUsize::new(0),
338        }
339    }
340}
341
342#[derive(Debug)]
343struct AtomicExecutorMetric {
344    active: AtomicGauge,
345    queued: AtomicGauge,
346}
347
348impl AtomicExecutorMetric {
349    fn new() -> Self {
350        Self {
351            active: AtomicGauge::new(),
352            queued: AtomicGauge::new(),
353        }
354    }
355}
356
357#[derive(Debug)]
358struct AtomicWatchdogMetric {
359    events: AtomicU64,
360    total_stall_micros: AtomicU64,
361    max_stall_micros: AtomicU64,
362}
363
364impl AtomicWatchdogMetric {
365    fn new() -> Self {
366        Self {
367            events: AtomicU64::new(0),
368            total_stall_micros: AtomicU64::new(0),
369            max_stall_micros: AtomicU64::new(0),
370        }
371    }
372}
373
374#[derive(Debug)]
375struct MetricsInner {
376    resources: [AtomicGauge; RESOURCE_METRIC_CLASS_COUNT],
377    buffers: [AtomicGauge; BUFFER_METRIC_CLASS_COUNT],
378    tasks: [AtomicTaskMetric; TASK_CLASS_COUNT],
379    wakes: [AtomicU64; WAKE_METRIC_COUNT],
380    readiness: AtomicReadinessMetric,
381    fairness_yields: [AtomicU64; FAIRNESS_LEVEL_COUNT],
382    completion_anomalies: [AtomicU64; COMPLETION_ANOMALY_COUNT],
383    channels: [AtomicChannelMetric; CHANNEL_METRIC_CLASS_COUNT],
384    executors: [AtomicExecutorMetric; EXECUTOR_METRIC_CLASS_COUNT],
385    watchdogs: [AtomicWatchdogMetric; WATCHDOG_METRIC_COUNT],
386    stderr_fallbacks: [AtomicU64; FALLBACK_SEVERITY_COUNT],
387}
388
389impl MetricsInner {
390    fn new() -> Self {
391        Self {
392            resources: std::array::from_fn(|_| AtomicGauge::new()),
393            buffers: std::array::from_fn(|_| AtomicGauge::new()),
394            tasks: std::array::from_fn(|_| AtomicTaskMetric::new()),
395            wakes: std::array::from_fn(|_| AtomicU64::new(0)),
396            readiness: AtomicReadinessMetric::new(),
397            fairness_yields: std::array::from_fn(|_| AtomicU64::new(0)),
398            completion_anomalies: std::array::from_fn(|_| AtomicU64::new(0)),
399            channels: std::array::from_fn(|_| AtomicChannelMetric::new()),
400            executors: std::array::from_fn(|_| AtomicExecutorMetric::new()),
401            watchdogs: std::array::from_fn(|_| AtomicWatchdogMetric::new()),
402            stderr_fallbacks: std::array::from_fn(|_| AtomicU64::new(0)),
403        }
404    }
405}
406
407/// Cloneable process-wide metrics handle. Its storage cardinality is fixed at construction.
408#[derive(Clone, Debug)]
409pub struct RuntimeMetrics {
410    inner: Arc<MetricsInner>,
411}
412
413impl Default for RuntimeMetrics {
414    fn default() -> Self {
415        Self::new()
416    }
417}
418
419impl RuntimeMetrics {
420    pub fn new() -> Self {
421        Self {
422            inner: Arc::new(MetricsInner::new()),
423        }
424    }
425
426    pub fn observe_resource(&self, class: ResourceMetricClass, current: usize) {
427        self.inner.resources[class.index()].observe(current);
428    }
429
430    pub fn observe_buffer(&self, class: BufferMetricClass, current_bytes: usize) {
431        self.inner.buffers[class.index()].observe(current_bytes);
432    }
433
434    pub fn task_started(&self, class: TaskClass) {
435        saturating_add_usize(&self.inner.tasks[task_class_index(class)].active, 1);
436    }
437
438    pub fn task_finished(&self, class: TaskClass, reason: TaskTerminalReason) {
439        saturating_sub_usize(&self.inner.tasks[task_class_index(class)].active, 1);
440        saturating_add_u64(
441            &self.inner.tasks[task_class_index(class)].terminal[task_terminal_reason_index(reason)],
442            1,
443        );
444    }
445
446    pub fn record_wake(&self, metric: WakeMetric) {
447        saturating_add_u64(&self.inner.wakes[metric.index()], 1);
448    }
449
450    pub fn observe_readiness(&self, ready_size: usize, oldest_age: Duration) {
451        self.inner.readiness.size.observe(ready_size);
452        let age_micros = duration_micros(oldest_age);
453        saturating_add_u64(&self.inner.readiness.age_samples, 1);
454        saturating_add_u64(&self.inner.readiness.total_oldest_age_micros, age_micros);
455        saturating_fetch_max_u64(&self.inner.readiness.max_oldest_age_micros, age_micros);
456    }
457
458    pub fn record_fairness_yield(&self, level: FairnessLevel) {
459        saturating_add_u64(&self.inner.fairness_yields[level.index()], 1);
460    }
461
462    pub fn record_completion_anomaly(&self, anomaly: CompletionAnomaly) {
463        saturating_add_u64(&self.inner.completion_anomalies[anomaly.index()], 1);
464    }
465
466    pub fn observe_channel(
467        &self,
468        class: ChannelMetricClass,
469        current_count: usize,
470        current_bytes: usize,
471    ) {
472        let channel = &self.inner.channels[class.index()];
473        saturating_fetch_max_usize(&channel.count_high_water, current_count);
474        saturating_fetch_max_usize(&channel.byte_high_water, current_bytes);
475    }
476
477    pub fn observe_executor(&self, class: ExecutorMetricClass, active: usize, queued: usize) {
478        let executor = &self.inner.executors[class.index()];
479        executor.active.observe(active);
480        executor.queued.observe(queued);
481    }
482
483    pub fn record_watchdog(&self, metric: WatchdogMetric, stall: Duration) {
484        let watchdog = &self.inner.watchdogs[metric.index()];
485        let micros = duration_micros(stall);
486        saturating_add_u64(&watchdog.events, 1);
487        saturating_add_u64(&watchdog.total_stall_micros, micros);
488        saturating_fetch_max_u64(&watchdog.max_stall_micros, micros);
489    }
490
491    pub fn record_stderr_fallback(&self, severity: TelemetrySeverity) {
492        saturating_add_u64(&self.inner.stderr_fallbacks[severity.index()], 1);
493    }
494
495    pub fn emit_stderr_fallback(&self, fallback: TelemetryFallback<'_>) {
496        self.record_stderr_fallback(fallback.severity);
497        emit_stderr_fallback(fallback);
498    }
499
500    /// Atomics make this snapshot race-safe but intentionally not transactional.
501    pub fn snapshot(&self) -> RuntimeMetricsSnapshot {
502        RuntimeMetricsSnapshot {
503            resources: std::array::from_fn(|index| self.inner.resources[index].snapshot()),
504            buffers: std::array::from_fn(|index| self.inner.buffers[index].snapshot()),
505            tasks: std::array::from_fn(|index| TaskMetricSnapshot {
506                active: self.inner.tasks[index].active.load(ORDERING),
507                terminal: std::array::from_fn(|reason| {
508                    self.inner.tasks[index].terminal[reason].load(ORDERING)
509                }),
510            }),
511            wakes: std::array::from_fn(|index| self.inner.wakes[index].load(ORDERING)),
512            readiness: ReadinessMetricSnapshot {
513                current_size: self.inner.readiness.size.current.load(ORDERING),
514                size_high_water: self.inner.readiness.size.high_water.load(ORDERING),
515                age_samples: self.inner.readiness.age_samples.load(ORDERING),
516                total_oldest_age_micros: self
517                    .inner
518                    .readiness
519                    .total_oldest_age_micros
520                    .load(ORDERING),
521                max_oldest_age_micros: self.inner.readiness.max_oldest_age_micros.load(ORDERING),
522            },
523            fairness_yields: std::array::from_fn(|index| {
524                self.inner.fairness_yields[index].load(ORDERING)
525            }),
526            completion_anomalies: std::array::from_fn(|index| {
527                self.inner.completion_anomalies[index].load(ORDERING)
528            }),
529            channels: std::array::from_fn(|index| ChannelMetricSnapshot {
530                count_high_water: self.inner.channels[index].count_high_water.load(ORDERING),
531                byte_high_water: self.inner.channels[index].byte_high_water.load(ORDERING),
532            }),
533            executors: std::array::from_fn(|index| ExecutorMetricSnapshot {
534                active: self.inner.executors[index].active.snapshot(),
535                queued: self.inner.executors[index].queued.snapshot(),
536            }),
537            watchdogs: std::array::from_fn(|index| WatchdogMetricSnapshot {
538                events: self.inner.watchdogs[index].events.load(ORDERING),
539                total_stall_micros: self.inner.watchdogs[index]
540                    .total_stall_micros
541                    .load(ORDERING),
542                max_stall_micros: self.inner.watchdogs[index].max_stall_micros.load(ORDERING),
543            }),
544            stderr_fallbacks: std::array::from_fn(|index| {
545                self.inner.stderr_fallbacks[index].load(ORDERING)
546            }),
547        }
548    }
549}
550
551#[derive(Clone, Copy, Debug, Eq, PartialEq)]
552pub struct TelemetryFallback<'a> {
553    pub severity: TelemetrySeverity,
554    pub code: TelemetryFallbackCode,
555    pub subsystem: TelemetrySubsystem,
556    pub message: &'a str,
557}
558
559pub fn format_stderr_fallback(fallback: TelemetryFallback<'_>) -> String {
560    let message_end = floor_char_boundary(fallback.message, MAX_FALLBACK_MESSAGE_BYTES);
561    let truncated = message_end < fallback.message.len();
562    let mut message = String::with_capacity(message_end);
563    for character in fallback.message[..message_end].chars() {
564        message.push(match character {
565            '"' => '\'',
566            '\\' => '/',
567            character if character.is_control() => ' ',
568            character => character,
569        });
570    }
571    format!(
572        "AGENTOS_TELEMETRY_FALLBACK severity={} code={} subsystem={} message=\"{}\" truncated={truncated}",
573        fallback.severity.as_str(),
574        fallback.code.as_str(),
575        fallback.subsystem.as_str(),
576        message,
577    )
578}
579
580pub fn emit_stderr_fallback(fallback: TelemetryFallback<'_>) {
581    eprintln!("{}", format_stderr_fallback(fallback));
582}
583
584fn task_class_index(class: TaskClass) -> usize {
585    match class {
586        TaskClass::Runtime => 0,
587        TaskClass::Dns => 1,
588        TaskClass::Socket => 2,
589        TaskClass::Listener => 3,
590        TaskClass::Udp => 4,
591        TaskClass::Tls => 5,
592        TaskClass::Http2 => 6,
593        TaskClass::Timer => 7,
594        TaskClass::Vm => 8,
595        TaskClass::Plugin => 9,
596    }
597}
598
599fn task_terminal_reason_index(reason: TaskTerminalReason) -> usize {
600    match reason {
601        TaskTerminalReason::Completed => 0,
602        TaskTerminalReason::Cancelled => 1,
603        TaskTerminalReason::Failed => 2,
604        TaskTerminalReason::Panicked => 3,
605    }
606}
607
608fn duration_micros(duration: Duration) -> u64 {
609    duration.as_micros().min(u128::from(u64::MAX)) as u64
610}
611
612fn floor_char_boundary(value: &str, maximum_bytes: usize) -> usize {
613    let mut end = value.len().min(maximum_bytes);
614    while !value.is_char_boundary(end) {
615        end -= 1;
616    }
617    end
618}
619
620fn saturating_add_u64(value: &AtomicU64, amount: u64) {
621    let mut current = value.load(ORDERING);
622    loop {
623        let next = current.saturating_add(amount);
624        if next == current {
625            return;
626        }
627        match value.compare_exchange_weak(current, next, ORDERING, ORDERING) {
628            Ok(_) => return,
629            Err(observed) => current = observed,
630        }
631    }
632}
633
634fn saturating_add_usize(value: &AtomicUsize, amount: usize) {
635    let mut current = value.load(ORDERING);
636    loop {
637        let next = current.saturating_add(amount);
638        if next == current {
639            return;
640        }
641        match value.compare_exchange_weak(current, next, ORDERING, ORDERING) {
642            Ok(_) => return,
643            Err(observed) => current = observed,
644        }
645    }
646}
647
648fn saturating_sub_usize(value: &AtomicUsize, amount: usize) {
649    let mut current = value.load(ORDERING);
650    loop {
651        let next = current.saturating_sub(amount);
652        if next == current {
653            return;
654        }
655        match value.compare_exchange_weak(current, next, ORDERING, ORDERING) {
656            Ok(_) => return,
657            Err(observed) => current = observed,
658        }
659    }
660}
661
662fn saturating_fetch_max_u64(value: &AtomicU64, observed: u64) {
663    let mut current = value.load(ORDERING);
664    while observed > current {
665        match value.compare_exchange_weak(current, observed, ORDERING, ORDERING) {
666            Ok(_) => return,
667            Err(actual) => current = actual,
668        }
669    }
670}
671
672fn saturating_fetch_max_usize(value: &AtomicUsize, observed: usize) {
673    let mut current = value.load(ORDERING);
674    while observed > current {
675        match value.compare_exchange_weak(current, observed, ORDERING, ORDERING) {
676            Ok(_) => return,
677            Err(actual) => current = actual,
678        }
679    }
680}
681
682#[cfg(test)]
683mod tests {
684    use super::*;
685
686    #[test]
687    fn snapshot_cardinality_is_fixed_by_enums() {
688        let metrics = RuntimeMetrics::new();
689        for class in ResourceMetricClass::ALL {
690            metrics.observe_resource(class, class.index() + 1);
691        }
692        for class in BufferMetricClass::ALL {
693            metrics.observe_buffer(class, class.index() + 1);
694        }
695        for class in TASK_CLASSES {
696            metrics.task_started(class);
697            for reason in TASK_TERMINAL_REASONS {
698                metrics.task_finished(class, reason);
699            }
700        }
701        for metric in WakeMetric::ALL {
702            metrics.record_wake(metric);
703        }
704        for level in FairnessLevel::ALL {
705            metrics.record_fairness_yield(level);
706        }
707        for anomaly in CompletionAnomaly::ALL {
708            metrics.record_completion_anomaly(anomaly);
709        }
710        for class in ChannelMetricClass::ALL {
711            metrics.observe_channel(class, 1, 1);
712        }
713        for class in ExecutorMetricClass::ALL {
714            metrics.observe_executor(class, 1, 1);
715        }
716        for metric in WatchdogMetric::ALL {
717            metrics.record_watchdog(metric, Duration::from_micros(1));
718        }
719
720        let snapshot = metrics.snapshot();
721        assert_eq!(snapshot.resources.len(), RESOURCE_METRIC_CLASS_COUNT);
722        assert_eq!(snapshot.buffers.len(), BUFFER_METRIC_CLASS_COUNT);
723        assert_eq!(snapshot.tasks.len(), TASK_CLASS_COUNT);
724        assert!(snapshot
725            .tasks
726            .iter()
727            .all(|task| task.terminal.len() == TASK_TERMINAL_REASON_COUNT));
728        assert_eq!(snapshot.wakes.len(), WAKE_METRIC_COUNT);
729        assert_eq!(snapshot.fairness_yields.len(), FAIRNESS_LEVEL_COUNT);
730        assert_eq!(
731            snapshot.completion_anomalies.len(),
732            COMPLETION_ANOMALY_COUNT
733        );
734        assert_eq!(snapshot.channels.len(), CHANNEL_METRIC_CLASS_COUNT);
735        assert_eq!(snapshot.executors.len(), EXECUTOR_METRIC_CLASS_COUNT);
736        assert_eq!(snapshot.watchdogs.len(), WATCHDOG_METRIC_COUNT);
737        assert_eq!(snapshot.stderr_fallbacks.len(), FALLBACK_SEVERITY_COUNT);
738    }
739
740    #[test]
741    fn high_water_marks_do_not_fall_and_dimensions_are_independent() {
742        let metrics = RuntimeMetrics::new();
743        metrics.observe_channel(ChannelMetricClass::BridgeResponse, 5, 100);
744        metrics.observe_channel(ChannelMetricClass::BridgeResponse, 3, 200);
745        metrics.observe_channel(ChannelMetricClass::BridgeResponse, 7, 150);
746        metrics.observe_executor(ExecutorMetricClass::Vm, 4, 9);
747        metrics.observe_executor(ExecutorMetricClass::Vm, 2, 3);
748        metrics.observe_readiness(8, Duration::from_micros(20));
749        metrics.observe_readiness(3, Duration::from_micros(50));
750
751        let snapshot = metrics.snapshot();
752        assert_eq!(
753            snapshot.channels[ChannelMetricClass::BridgeResponse.index()],
754            ChannelMetricSnapshot {
755                count_high_water: 7,
756                byte_high_water: 200,
757            }
758        );
759        assert_eq!(
760            snapshot.executors[ExecutorMetricClass::Vm.index()],
761            ExecutorMetricSnapshot {
762                active: GaugeSnapshot {
763                    current: 2,
764                    high_water: 4,
765                },
766                queued: GaugeSnapshot {
767                    current: 3,
768                    high_water: 9,
769                },
770            }
771        );
772        assert_eq!(snapshot.readiness.current_size, 3);
773        assert_eq!(snapshot.readiness.size_high_water, 8);
774        assert_eq!(snapshot.readiness.age_samples, 2);
775        assert_eq!(snapshot.readiness.total_oldest_age_micros, 70);
776        assert_eq!(snapshot.readiness.max_oldest_age_micros, 50);
777    }
778
779    #[test]
780    fn counters_saturate_instead_of_wrapping() {
781        let metrics = RuntimeMetrics::new();
782        metrics.inner.wakes[WakeMetric::Attempted.index()].store(u64::MAX - 1, ORDERING);
783        metrics.record_wake(WakeMetric::Attempted);
784        metrics.record_wake(WakeMetric::Attempted);
785        metrics.inner.tasks[task_class_index(TaskClass::Runtime)]
786            .active
787            .store(usize::MAX, ORDERING);
788        metrics.task_started(TaskClass::Runtime);
789        metrics
790            .inner
791            .readiness
792            .total_oldest_age_micros
793            .store(u64::MAX - 1, ORDERING);
794        metrics.observe_readiness(0, Duration::from_micros(2));
795
796        let snapshot = metrics.snapshot();
797        assert_eq!(snapshot.wakes[WakeMetric::Attempted.index()], u64::MAX);
798        assert_eq!(
799            snapshot.tasks[task_class_index(TaskClass::Runtime)].active,
800            usize::MAX
801        );
802        assert_eq!(snapshot.readiness.total_oldest_age_micros, u64::MAX);
803    }
804
805    #[test]
806    fn fallback_format_is_compact_bounded_and_single_line() {
807        let formatted = format_stderr_fallback(TelemetryFallback {
808            severity: TelemetrySeverity::Warning,
809            code: TelemetryFallbackCode::ResourceLimit,
810            subsystem: TelemetrySubsystem::Reactor,
811            message: "queue \"full\"\nretry",
812        });
813        assert_eq!(
814            formatted,
815            "AGENTOS_TELEMETRY_FALLBACK severity=warning code=ERR_AGENTOS_RESOURCE_LIMIT subsystem=reactor message=\"queue 'full' retry\" truncated=false"
816        );
817        assert!(!formatted.contains('\n'));
818
819        let oversized = "x".repeat(MAX_FALLBACK_MESSAGE_BYTES + 100);
820        let truncated = format_stderr_fallback(TelemetryFallback {
821            severity: TelemetrySeverity::Fatal,
822            code: TelemetryFallbackCode::RuntimeWorkerStall,
823            subsystem: TelemetrySubsystem::Runtime,
824            message: &oversized,
825        });
826        assert!(truncated.ends_with("truncated=true"));
827        assert!(truncated.len() < MAX_FALLBACK_MESSAGE_BYTES + 200);
828    }
829}
agentos_runtime/metrics.rs

agentos_runtime/
metrics.rs
