oxicode_agent/agent_loop/
queues.rs

1//! Queue management for agent loop.
2//!
3//! Provides capacity-limited push, drain, and clear operations for the
4//! steering and follow-up message queues used by the agent loop.
5
6/// Maximum number of messages in the steering queue before back-pressure is applied.
7const STEERING_QUEUE_CAPACITY: usize = 256;
8
9/// Maximum number of messages in the follow-up queue before back-pressure is applied.
10const FOLLOW_UP_QUEUE_CAPACITY: usize = 64;
11
12/// Clear the steering queue.
13pub(crate) fn clear_steering_queue(loop_ref: &super::AgentLoop) {
14    loop_ref.steering_queue.write().clear();
15}
16
17/// Clear the follow-up queue.
18pub(crate) fn clear_follow_up_queue(loop_ref: &super::AgentLoop) {
19    loop_ref.follow_up_queue.write().clear();
20}
21
22/// Clear both steering and follow-up queues.
23pub(crate) fn clear_all_queues(loop_ref: &super::AgentLoop) {
24    clear_steering_queue(loop_ref);
25    clear_follow_up_queue(loop_ref);
26}
27
28/// Drain and return all messages from the steering queue.
29pub(crate) fn drain_steering_queue(loop_ref: &super::AgentLoop) -> Vec<oxicode_ai::Message> {
30    let mut queue = loop_ref.steering_queue.write();
31    queue.drain(..).collect()
32}
33
34/// Drain and return all messages from the follow-up queue.
35pub(crate) fn drain_follow_up_queue(loop_ref: &super::AgentLoop) -> Vec<oxicode_ai::Message> {
36    let mut queue = loop_ref.follow_up_queue.write();
37    queue.drain(..).collect()
38}
39
40/// Try to push a message onto the steering queue.
41///
42/// Returns `true` if the message was accepted, `false` if the queue is at
43/// capacity (in which case the message is dropped and a warning is logged).
44pub(crate) fn try_push_steering(loop_ref: &super::AgentLoop, message: oxicode_ai::Message) -> bool {
45    let mut queue = loop_ref.steering_queue.write();
46    if queue.len() >= STEERING_QUEUE_CAPACITY {
47        tracing::warn!(
48            capacity = STEERING_QUEUE_CAPACITY,
49            "Steering queue at capacity — dropping message"
50        );
51        return false;
52    }
53    queue.push(message);
54    true
55}
56
57/// Try to push a message onto the follow-up queue.
58///
59/// Returns `true` if the message was accepted, `false` if the queue is at
60/// capacity (in which case the message is dropped and a warning is logged).
61pub(crate) fn try_push_follow_up(
62    loop_ref: &super::AgentLoop,
63    message: oxicode_ai::Message,
64) -> bool {
65    let mut queue = loop_ref.follow_up_queue.write();
66    if queue.len() >= FOLLOW_UP_QUEUE_CAPACITY {
67        tracing::warn!(
68            capacity = FOLLOW_UP_QUEUE_CAPACITY,
69            "Follow-up queue at capacity — dropping message"
70        );
71        return false;
72    }
73    queue.push(message);
74    true
75}
76
77/// Returns the number of messages currently in the steering queue.
78#[allow(dead_code)]
79pub(crate) fn steering_queue_len(loop_ref: &super::AgentLoop) -> usize {
80    loop_ref.steering_queue.read().len()
81}
82
83/// Returns the number of messages currently in the follow-up queue.
84#[allow(dead_code)]
85pub(crate) fn follow_up_queue_len(loop_ref: &super::AgentLoop) -> usize {
86    loop_ref.follow_up_queue.read().len()
87}
oxicode_agent/agent_loop/queues.rs

oxicode_agent/agent_loop/
queues.rs
