agentos_bridge/
queue_tracker.rs

1//! Centralized bounded-queue usage tracker.
2//!
3//! secure-exec streams guest output through a *chain* of bounded queues: the
4//! V8 -> host event channel, the sidecar stdout/stdin frame queues, and so on.
5//! Each queue applies backpressure when full (it parks the producer until the
6//! consumer drains) rather than crashing, but backpressure is invisible: a slow
7//! host consumer silently stalls a session with nothing in the logs.
8//!
9//! This module gives that whole chain a single, inspectable home:
10//!
11//! * Every bounded queue registers a [`QueueGauge`] (with a stable name and its
12//!   capacity) in a process-global [`QueueRegistry`].
13//! * Producers report depth as they enqueue (either by an exact count for
14//!   manually-tracked queues via [`TrackedSyncSender`], or by sampling the live
15//!   depth of a Tokio channel via [`QueueGauge::observe_depth`]).
16//! * When a queue crosses [`WARN_FILL_PERCENT`] of capacity the gauge emits a
17//!   single `warn!`, so "the consumer is falling behind" shows up *before* the
18//!   queue saturates and backpressure stalls the session. It re-arms once the
19//!   queue drains back below [`REARM_FILL_PERCENT`].
20//! * [`queue_snapshot`] returns the live depth / high-water / capacity of every
21//!   registered queue for debugging or a status endpoint.
22
23use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
24use std::sync::mpsc::{Receiver, RecvError, SendError, SyncSender, TrySendError};
25use std::sync::{Arc, Mutex, OnceLock, Weak};
26
27/// Fill fraction (percent of capacity) at or above which a queue is considered
28/// "near full" and emits a warning. Edge-triggered so a steadily-full queue logs
29/// once, not on every enqueue.
30pub const WARN_FILL_PERCENT: usize = 80;
31
32/// Fill fraction a near-full queue must drain back below before it will warn
33/// again. The gap to [`WARN_FILL_PERCENT`] provides hysteresis so a queue
34/// hovering at the threshold does not flap.
35pub const REARM_FILL_PERCENT: usize = 50;
36
37/// What class of bounded resource a gauge tracks. Lets a snapshot / a host hook
38/// group and reason about limits beyond just queues.
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40pub enum LimitCategory {
41    /// A bounded channel/buffer with enqueue/dequeue flow (the default).
42    Queue,
43    /// A saturating resource counter (fds, processes, sockets, bytes in use).
44    Resource,
45    /// A memory/heap envelope.
46    Memory,
47    /// A CPU or wall-clock execution budget.
48    Cpu,
49}
50
51impl LimitCategory {
52    /// Stable lowercase tag for logs and snapshots.
53    pub fn as_str(self) -> &'static str {
54        match self {
55            LimitCategory::Queue => "queue",
56            LimitCategory::Resource => "resource",
57            LimitCategory::Memory => "memory",
58            LimitCategory::Cpu => "cpu",
59        }
60    }
61}
62
63/// Stable catalog of tracked limits that may emit near-capacity or exhaustion
64/// warnings. Keep `website/src/content/docs/docs/features/resource-limits.mdx`
65/// in sync when adding, removing, or renaming variants so host-visible warning
66/// names and the documented constants do not drift.
67#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
68pub enum TrackedLimit {
69    JavascriptEventChannel,
70    V8SessionFrames,
71    SidecarStdinFrames,
72    SidecarStdoutFrames,
73    CompletedSidecarResponses,
74    PendingProcessEvents,
75    PendingSidecarResponses,
76    OutboundSidecarRequests,
77    VmProcesses,
78    VmOpenFds,
79    VmPipes,
80    VmPtys,
81    VmSockets,
82    VmConnections,
83    VmSocketBufferedBytes,
84    VmSocketDatagramQueueLen,
85    VmFilesystemBytes,
86    VmInodes,
87    VmRecursiveFsDepth,
88    VmRecursiveFsEntries,
89    V8HeapBytes,
90    V8CpuTimeMs,
91    V8WallClockMs,
92    WasmFuelMs,
93    WasmMemoryBytes,
94}
95
96impl TrackedLimit {
97    /// Stable lowercase tag emitted in logs, snapshots, and host
98    /// `limit_warning` events.
99    pub fn as_str(self) -> &'static str {
100        match self {
101            TrackedLimit::JavascriptEventChannel => "javascript_event_channel",
102            TrackedLimit::V8SessionFrames => "v8_session_frames",
103            TrackedLimit::SidecarStdinFrames => "sidecar_stdin_frames",
104            TrackedLimit::SidecarStdoutFrames => "sidecar_stdout_frames",
105            TrackedLimit::CompletedSidecarResponses => "completed_sidecar_responses",
106            TrackedLimit::PendingProcessEvents => "pending_process_events",
107            TrackedLimit::PendingSidecarResponses => "pending_sidecar_responses",
108            TrackedLimit::OutboundSidecarRequests => "outbound_sidecar_requests",
109            TrackedLimit::VmProcesses => "vm_processes",
110            TrackedLimit::VmOpenFds => "vm_open_fds",
111            TrackedLimit::VmPipes => "vm_pipes",
112            TrackedLimit::VmPtys => "vm_ptys",
113            TrackedLimit::VmSockets => "vm_sockets",
114            TrackedLimit::VmConnections => "vm_connections",
115            TrackedLimit::VmSocketBufferedBytes => "vm_socket_buffered_bytes",
116            TrackedLimit::VmSocketDatagramQueueLen => "vm_socket_datagram_queue_len",
117            TrackedLimit::VmFilesystemBytes => "vm_filesystem_bytes",
118            TrackedLimit::VmInodes => "vm_inodes",
119            TrackedLimit::VmRecursiveFsDepth => "vm_recursive_fs_depth",
120            TrackedLimit::VmRecursiveFsEntries => "vm_recursive_fs_entries",
121            TrackedLimit::V8HeapBytes => "v8_heap_bytes",
122            TrackedLimit::V8CpuTimeMs => "v8_cpu_time_ms",
123            TrackedLimit::V8WallClockMs => "v8_wall_clock_ms",
124            TrackedLimit::WasmFuelMs => "wasm_fuel_ms",
125            TrackedLimit::WasmMemoryBytes => "wasm_memory_bytes",
126        }
127    }
128
129    pub fn category(self) -> LimitCategory {
130        match self {
131            TrackedLimit::JavascriptEventChannel
132            | TrackedLimit::V8SessionFrames
133            | TrackedLimit::SidecarStdinFrames
134            | TrackedLimit::SidecarStdoutFrames
135            | TrackedLimit::CompletedSidecarResponses
136            | TrackedLimit::PendingProcessEvents
137            | TrackedLimit::PendingSidecarResponses
138            | TrackedLimit::OutboundSidecarRequests => LimitCategory::Queue,
139            TrackedLimit::VmProcesses
140            | TrackedLimit::VmOpenFds
141            | TrackedLimit::VmPipes
142            | TrackedLimit::VmPtys
143            | TrackedLimit::VmSockets
144            | TrackedLimit::VmConnections
145            | TrackedLimit::VmSocketBufferedBytes
146            | TrackedLimit::VmSocketDatagramQueueLen
147            | TrackedLimit::VmFilesystemBytes
148            | TrackedLimit::VmInodes
149            | TrackedLimit::VmRecursiveFsDepth
150            | TrackedLimit::VmRecursiveFsEntries => LimitCategory::Resource,
151            TrackedLimit::V8HeapBytes | TrackedLimit::WasmMemoryBytes => LimitCategory::Memory,
152            TrackedLimit::V8CpuTimeMs | TrackedLimit::V8WallClockMs | TrackedLimit::WasmFuelMs => {
153                LimitCategory::Cpu
154            }
155        }
156    }
157}
158
159/// A near-capacity event for one limit, delivered to the global warning sink at
160/// the same edge as the `tracing::warn!`. This is the structured payload a host
161/// hook (e.g. agentOS `onLimitWarning`) is built from.
162#[derive(Debug, Clone)]
163pub struct LimitWarning {
164    pub name: TrackedLimit,
165    pub category: LimitCategory,
166    pub observed: usize,
167    pub capacity: usize,
168    pub fill_percent: usize,
169}
170
171type LimitWarningHandler = Arc<dyn Fn(&LimitWarning) + Send + Sync>;
172
173fn warning_handler_slot() -> &'static Mutex<Option<LimitWarningHandler>> {
174    static HANDLER: OnceLock<Mutex<Option<LimitWarningHandler>>> = OnceLock::new();
175    HANDLER.get_or_init(|| Mutex::new(None))
176}
177
178/// Install a process-global sink that is invoked on the same edge-triggered,
179/// hysteresis-gated boundary as the `tracing::warn!` whenever a tracked limit
180/// crosses [`WARN_FILL_PERCENT`]. The sidecar uses this to forward limit warnings
181/// to the host as structured events (the `onLimitWarning` hook). The handler must
182/// be cheap and non-blocking; it runs on the producer's thread.
183pub fn set_limit_warning_handler(handler: Box<dyn Fn(&LimitWarning) + Send + Sync>) {
184    if let Ok(mut slot) = warning_handler_slot().lock() {
185        *slot = Some(Arc::from(handler));
186    }
187}
188
189fn dispatch_warning(warning: &LimitWarning) {
190    // Clone the handler Arc out and DROP the registry mutex before invoking it,
191    // so the handler never runs while we hold the global lock. The sink can be
192    // reached while a kernel lock is held (e.g. fd_tables -> warning_mutex ->
193    // handler); keeping the invocation outside the mutex avoids a lock-order
194    // hazard if the handler ever does non-trivial work.
195    let handler = match warning_handler_slot().lock() {
196        Ok(slot) => slot.as_ref().cloned(),
197        Err(_) => None,
198    };
199    if let Some(handler) = handler {
200        handler(warning);
201    }
202}
203
204/// Emit a structured/logged warning for a limit that has already been exhausted.
205/// Use this for runtime caps such as CPU or heap exhaustion where there is no
206/// continuously sampled queue depth to observe before the terminal edge.
207pub fn warn_limit_exhausted(name: TrackedLimit, observed: usize, capacity: usize) {
208    let fill_percent = observed
209        .saturating_mul(100)
210        .checked_div(capacity)
211        .unwrap_or(0);
212    let category = name.category();
213    tracing::warn!(
214        limit = name.as_str(),
215        category = category.as_str(),
216        observed,
217        capacity,
218        fill_percent,
219        "bounded limit exhausted"
220    );
221    dispatch_warning(&LimitWarning {
222        name,
223        category,
224        observed,
225        capacity,
226        fill_percent,
227    });
228}
229
230/// Live usage gauge for a single bounded queue.
231///
232/// Cloneable handles share one gauge through an [`Arc`]; the registry keeps a
233/// [`Weak`] so a gauge is auto-pruned from snapshots once its queue is dropped.
234#[derive(Debug)]
235pub struct QueueGauge {
236    name: TrackedLimit,
237    category: LimitCategory,
238    capacity: usize,
239    depth: AtomicUsize,
240    high_water: AtomicUsize,
241    warned: AtomicBool,
242}
243
244impl QueueGauge {
245    fn new(name: TrackedLimit, capacity: usize, category: LimitCategory) -> Self {
246        Self {
247            name,
248            category,
249            capacity,
250            depth: AtomicUsize::new(0),
251            high_water: AtomicUsize::new(0),
252            warned: AtomicBool::new(false),
253        }
254    }
255
256    /// Stable limit name (used in logs and snapshots).
257    pub fn name(&self) -> TrackedLimit {
258        self.name
259    }
260
261    /// The class of bounded resource this gauge tracks.
262    pub fn category(&self) -> LimitCategory {
263        self.category
264    }
265
266    /// Configured queue capacity (slots). `0` means unbounded / untracked.
267    pub fn capacity(&self) -> usize {
268        self.capacity
269    }
270
271    /// Current observed depth.
272    pub fn depth(&self) -> usize {
273        self.depth.load(Ordering::Acquire)
274    }
275
276    /// Highest depth observed over the gauge's lifetime.
277    pub fn high_water(&self) -> usize {
278        self.high_water.load(Ordering::Acquire)
279    }
280
281    /// Fill fraction (0–100) at the given depth. Saturates rather than dividing
282    /// by zero for untracked (capacity 0) queues.
283    fn fill_percent(&self, depth: usize) -> usize {
284        depth
285            .saturating_mul(100)
286            .checked_div(self.capacity)
287            .unwrap_or(0)
288    }
289
290    /// Record a new depth: refresh the high-water mark and emit an edge-triggered
291    /// near-capacity warning (or recovery debug line).
292    fn evaluate(&self, depth: usize) {
293        self.high_water.fetch_max(depth, Ordering::AcqRel);
294        if self.capacity == 0 {
295            return;
296        }
297        let percent = self.fill_percent(depth);
298        if percent >= WARN_FILL_PERCENT {
299            if !self.warned.swap(true, Ordering::AcqRel) {
300                tracing::warn!(
301                    limit = self.name.as_str(),
302                    category = self.category.as_str(),
303                    observed = depth,
304                    capacity = self.capacity,
305                    fill_percent = percent,
306                    "bounded limit near capacity"
307                );
308                // Same edge as the log: notify the structured warning sink so the
309                // host can surface it (e.g. an onLimitWarning hook). Edge-triggered
310                // + hysteresis keep this from firing more than once per crossing.
311                dispatch_warning(&LimitWarning {
312                    name: self.name,
313                    category: self.category,
314                    observed: depth,
315                    capacity: self.capacity,
316                    fill_percent: percent,
317                });
318            }
319        } else if percent <= REARM_FILL_PERCENT && self.warned.swap(false, Ordering::AcqRel) {
320            tracing::debug!(
321                limit = self.name.as_str(),
322                category = self.category.as_str(),
323                depth,
324                capacity = self.capacity,
325                fill_percent = percent,
326                "bounded limit drained back below threshold"
327            );
328        }
329    }
330
331    /// Report the queue's exact current depth (for queues whose backing channel
332    /// exposes its live length, e.g. a Tokio mpsc via `max_capacity - capacity`).
333    pub fn observe_depth(&self, depth: usize) {
334        self.depth.store(depth, Ordering::Release);
335        self.evaluate(depth);
336    }
337
338    /// Account for one item entering the queue (for manually-tracked queues).
339    pub fn record_enqueue(&self) {
340        let depth = self.depth.fetch_add(1, Ordering::AcqRel) + 1;
341        self.evaluate(depth);
342    }
343
344    /// Account for one item leaving the queue. Saturates at zero so a stray
345    /// dequeue can never underflow the depth counter. Re-evaluates so a gauge
346    /// that latched "warned" while full re-arms once the queue drains back below
347    /// the re-arm threshold, even if the producer has since gone idle.
348    pub fn record_dequeue(&self) {
349        let mut current = self.depth.load(Ordering::Acquire);
350        loop {
351            if current == 0 {
352                return;
353            }
354            match self.depth.compare_exchange_weak(
355                current,
356                current - 1,
357                Ordering::AcqRel,
358                Ordering::Acquire,
359            ) {
360                Ok(_) => {
361                    self.evaluate(current - 1);
362                    break;
363                }
364                Err(actual) => current = actual,
365            }
366        }
367    }
368}
369
370/// Immutable view of a tracked limit's usage, returned by [`queue_snapshot`].
371#[derive(Debug, Clone, PartialEq, Eq)]
372pub struct QueueSnapshot {
373    pub name: TrackedLimit,
374    pub category: LimitCategory,
375    pub depth: usize,
376    pub high_water: usize,
377    pub capacity: usize,
378    pub fill_percent: usize,
379}
380
381/// Process-global registry of every live [`QueueGauge`].
382#[derive(Default)]
383pub struct QueueRegistry {
384    gauges: Mutex<Vec<Weak<QueueGauge>>>,
385}
386
387impl QueueRegistry {
388    /// The shared registry. All `secure-exec` bounded queues register here so
389    /// their usage can be inspected from one place.
390    pub fn global() -> &'static QueueRegistry {
391        static REGISTRY: OnceLock<QueueRegistry> = OnceLock::new();
392        REGISTRY.get_or_init(QueueRegistry::default)
393    }
394
395    /// Register a new bounded limit and return its gauge. Dropping the returned
396    /// `Arc` (and all clones) removes the limit from future snapshots.
397    pub fn register(&self, name: TrackedLimit, capacity: usize) -> Arc<QueueGauge> {
398        let category = name.category();
399        let gauge = Arc::new(QueueGauge::new(name, capacity, category));
400        let mut gauges = self.gauges.lock().expect("queue registry mutex poisoned");
401        gauges.retain(|weak| weak.strong_count() > 0);
402        gauges.push(Arc::downgrade(&gauge));
403        gauge
404    }
405
406    /// Snapshot the live usage of every registered queue, pruning dead entries.
407    pub fn snapshot(&self) -> Vec<QueueSnapshot> {
408        let mut gauges = self.gauges.lock().expect("queue registry mutex poisoned");
409        gauges.retain(|weak| weak.strong_count() > 0);
410        gauges
411            .iter()
412            .filter_map(Weak::upgrade)
413            .map(|gauge| {
414                let depth = gauge.depth();
415                QueueSnapshot {
416                    name: gauge.name(),
417                    category: gauge.category(),
418                    depth,
419                    high_water: gauge.high_water(),
420                    capacity: gauge.capacity(),
421                    fill_percent: gauge.fill_percent(depth),
422                }
423            })
424            .collect()
425    }
426}
427
428/// Register a bounded queue (the [`LimitCategory::Queue`] case) with the global
429/// registry. Convenience over [`QueueRegistry::global`] + [`QueueRegistry::register`].
430pub fn register_queue(name: TrackedLimit, capacity: usize) -> Arc<QueueGauge> {
431    debug_assert_eq!(name.category(), LimitCategory::Queue);
432    QueueRegistry::global().register(name, capacity)
433}
434
435/// Register a non-queue bounded limit (a saturating resource or memory envelope)
436/// with the global registry, so it shares the same approach-warning + snapshot
437/// machinery as queues. Observe usage with [`QueueGauge::observe_depth`].
438pub fn register_limit(name: TrackedLimit, capacity: usize) -> Arc<QueueGauge> {
439    QueueRegistry::global().register(name, capacity)
440}
441
442/// Snapshot every registered queue from the global registry.
443pub fn queue_snapshot() -> Vec<QueueSnapshot> {
444    QueueRegistry::global().snapshot()
445}
446
447/// Emit a `debug!` line for every registered queue. Useful for an on-demand dump
448/// of the queue chain when diagnosing a stall.
449pub fn log_queue_snapshot() {
450    for stat in queue_snapshot() {
451        tracing::debug!(
452            limit = stat.name.as_str(),
453            category = stat.category.as_str(),
454            depth = stat.depth,
455            high_water = stat.high_water,
456            capacity = stat.capacity,
457            fill_percent = stat.fill_percent,
458            "limit usage"
459        );
460    }
461}
462
463/// A `std::sync::mpsc::SyncSender` that feeds a [`QueueGauge`] as items flow
464/// through it, so a queue whose backing channel cannot report its own length
465/// still participates in the centralized tracker.
466///
467/// `send` keeps the underlying blocking-backpressure semantics; it just records
468/// the enqueue first so near-capacity warnings fire as the queue fills.
469#[derive(Debug)]
470pub struct TrackedSyncSender<T> {
471    inner: SyncSender<T>,
472    gauge: Arc<QueueGauge>,
473}
474
475impl<T> Clone for TrackedSyncSender<T> {
476    fn clone(&self) -> Self {
477        Self {
478            inner: self.inner.clone(),
479            gauge: Arc::clone(&self.gauge),
480        }
481    }
482}
483
484impl<T> TrackedSyncSender<T> {
485    /// Blocking send: record the enqueue, then hand off to the bounded channel
486    /// (which parks the caller until a slot is free: clean backpressure).
487    pub fn send(&self, value: T) -> Result<(), SendError<T>> {
488        self.gauge.record_enqueue();
489        self.inner.send(value)
490    }
491
492    /// Non-blocking send: record the enqueue only on success. Lets a caller with
493    /// its own deadline poll instead of parking indefinitely on a full queue.
494    pub fn try_send(&self, value: T) -> Result<(), TrySendError<T>> {
495        match self.inner.try_send(value) {
496            Ok(()) => {
497                self.gauge.record_enqueue();
498                Ok(())
499            }
500            Err(error) => Err(error),
501        }
502    }
503
504    /// The gauge backing this sender.
505    pub fn gauge(&self) -> &Arc<QueueGauge> {
506        &self.gauge
507    }
508}
509
510/// Receiver half of a [`tracked_sync_channel`]; records a dequeue for every
511/// item it yields so the gauge depth tracks the real backlog.
512#[derive(Debug)]
513pub struct TrackedReceiver<T> {
514    inner: Receiver<T>,
515    gauge: Arc<QueueGauge>,
516}
517
518impl<T> TrackedReceiver<T> {
519    /// Blocking receive that decrements the gauge for the item it returns.
520    pub fn recv(&self) -> Result<T, RecvError> {
521        let value = self.inner.recv()?;
522        self.gauge.record_dequeue();
523        Ok(value)
524    }
525}
526
527/// Create a bounded `std::sync::mpsc` sync-channel whose depth is tracked by a
528/// registered [`QueueGauge`]. Drop-in for `std::sync::mpsc::sync_channel` plus
529/// centralized usage tracking + near-capacity warnings.
530pub fn tracked_sync_channel<T>(
531    name: TrackedLimit,
532    capacity: usize,
533) -> (TrackedSyncSender<T>, TrackedReceiver<T>) {
534    let (tx, rx) = std::sync::mpsc::sync_channel(capacity);
535    let gauge = register_queue(name, capacity);
536    (
537        TrackedSyncSender {
538            inner: tx,
539            gauge: Arc::clone(&gauge),
540        },
541        TrackedReceiver { inner: rx, gauge },
542    )
543}
544
545#[cfg(test)]
546mod tests {
547    use super::*;
548
549    #[test]
550    fn gauge_tracks_depth_and_high_water() {
551        let gauge = QueueGauge::new(
552            TrackedLimit::JavascriptEventChannel,
553            10,
554            LimitCategory::Queue,
555        );
556        assert_eq!(gauge.depth(), 0);
557        gauge.record_enqueue();
558        gauge.record_enqueue();
559        assert_eq!(gauge.depth(), 2);
560        assert_eq!(gauge.high_water(), 2);
561        gauge.record_dequeue();
562        assert_eq!(gauge.depth(), 1);
563        // High-water never regresses.
564        assert_eq!(gauge.high_water(), 2);
565        // Dequeue never underflows below zero.
566        gauge.record_dequeue();
567        gauge.record_dequeue();
568        assert_eq!(gauge.depth(), 0);
569    }
570
571    #[test]
572    fn gauge_warn_flag_is_edge_triggered_with_hysteresis() {
573        let gauge = QueueGauge::new(TrackedLimit::V8SessionFrames, 10, LimitCategory::Queue);
574        // Below 80%: not warned.
575        gauge.observe_depth(7);
576        assert!(!gauge.warned.load(Ordering::Acquire));
577        // Cross 80%: warned.
578        gauge.observe_depth(8);
579        assert!(gauge.warned.load(Ordering::Acquire));
580        // Still near full: stays armed (single warning, not re-fired).
581        gauge.observe_depth(9);
582        assert!(gauge.warned.load(Ordering::Acquire));
583        // Drain to <=50%: re-arms.
584        gauge.observe_depth(5);
585        assert!(!gauge.warned.load(Ordering::Acquire));
586    }
587
588    #[test]
589    fn gauge_rearms_on_dequeue_drain() {
590        // record_enqueue/record_dequeue gauges (TrackedSyncSender/Receiver) must
591        // also re-arm as they drain, not only stay latched after the producer idles.
592        let gauge = QueueGauge::new(TrackedLimit::SidecarStdoutFrames, 10, LimitCategory::Queue);
593        for _ in 0..9 {
594            gauge.record_enqueue(); // climbs to 90% -> warned
595        }
596        assert_eq!(gauge.depth(), 9);
597        assert!(gauge.warned.load(Ordering::Acquire));
598        for _ in 0..6 {
599            gauge.record_dequeue(); // drains to 30% (<=50%) -> re-arm on dequeue
600        }
601        assert_eq!(gauge.depth(), 3);
602        assert!(!gauge.warned.load(Ordering::Acquire));
603    }
604
605    #[test]
606    fn tracked_channel_reports_usage_through_registry() {
607        let (tx, rx) = tracked_sync_channel::<u32>(TrackedLimit::SidecarStdoutFrames, 4);
608        tx.send(1).unwrap();
609        tx.send(2).unwrap();
610
611        let snapshot = queue_snapshot();
612        let entry = snapshot
613            .iter()
614            .find(|stat| stat.name == TrackedLimit::SidecarStdoutFrames)
615            .expect("registered queue should appear in snapshot");
616        assert_eq!(entry.depth, 2);
617        assert_eq!(entry.capacity, 4);
618        assert_eq!(entry.high_water, 2);
619        assert_eq!(entry.fill_percent, 50);
620        assert_eq!(entry.category, LimitCategory::Queue);
621
622        assert_eq!(rx.recv().unwrap(), 1);
623        assert_eq!(tx.gauge().depth(), 1);
624
625        // Dropping the channel removes it from later snapshots.
626        drop(tx);
627        drop(rx);
628        assert!(queue_snapshot()
629            .iter()
630            .all(|stat| stat.name != TrackedLimit::SidecarStdoutFrames));
631    }
632
633    #[test]
634    fn warning_sink_fires_once_per_crossing() {
635        let captured: Arc<Mutex<Vec<LimitWarning>>> = Arc::new(Mutex::new(Vec::new()));
636        let sink = Arc::clone(&captured);
637        // The handler is global; filter by our unique name so a gauge from a
638        // concurrently-running test can never pollute this assertion.
639        set_limit_warning_handler(Box::new(move |warning| {
640            if warning.name == TrackedLimit::VmPipes {
641                sink.lock().expect("sink mutex").push(warning.clone());
642            }
643        }));
644
645        let gauge = register_limit(TrackedLimit::VmPipes, 10);
646        gauge.observe_depth(7); // below 80%: no warning
647        assert!(captured.lock().unwrap().is_empty());
648        gauge.observe_depth(9); // crosses 80%: fires once
649        gauge.observe_depth(10); // still near full: must NOT re-fire (edge-triggered)
650
651        let warnings = captured.lock().unwrap();
652        assert_eq!(
653            warnings.len(),
654            1,
655            "warning sink must fire once per crossing"
656        );
657        assert_eq!(warnings[0].category, LimitCategory::Resource);
658        assert_eq!(warnings[0].capacity, 10);
659        assert!(warnings[0].fill_percent >= WARN_FILL_PERCENT);
660    }
661
662    #[test]
663    fn exhausted_warning_sink_fires_immediately() {
664        let captured: Arc<Mutex<Vec<LimitWarning>>> = Arc::new(Mutex::new(Vec::new()));
665        let sink = Arc::clone(&captured);
666        set_limit_warning_handler(Box::new(move |warning| {
667            if warning.name == TrackedLimit::V8CpuTimeMs {
668                sink.lock().expect("sink mutex").push(warning.clone());
669            }
670        }));
671
672        warn_limit_exhausted(TrackedLimit::V8CpuTimeMs, 30_000, 30_000);
673
674        let warnings = captured.lock().unwrap();
675        assert_eq!(warnings.len(), 1);
676        assert_eq!(warnings[0].category, LimitCategory::Cpu);
677        assert_eq!(warnings[0].fill_percent, 100);
678    }
679}
agentos_bridge/queue_tracker.rs

agentos_bridge/
queue_tracker.rs
