fno_agents/
write_queue.rs

1//! Bounded-backpressure stdin queue (design module `write_queue.rs`).
2//!
3//! Inputs destined for an agent's PTY stdin are enqueued here rather than
4//! written directly, so a slow or wedged child cannot block the caller and a
5//! burst of `ask`/drive keystrokes cannot grow memory without bound. The queue
6//! is capacity-bounded; `enqueue` returns [`WriteQueueError::Full`] when at
7//! capacity so the caller surfaces backpressure rather than buffering forever.
8//!
9//! Runtime-agnostic by design (Wave 1): the worker's writer loop pops messages
10//! and writes them to the PTY master. Wave 3's daemon drives this from a tokio
11//! task; nothing here assumes an async runtime.
12
13use std::collections::VecDeque;
14
15/// A single unit of work for the PTY writer.
16#[derive(Debug, Clone, PartialEq, Eq)]
17pub enum WriteMsg {
18    /// Raw bytes to write to the child's stdin (already envelope-wrapped by the
19    /// caller for non-Claude providers; the queue is content-agnostic).
20    Bytes(Vec<u8>),
21    /// A terminal resize to apply before subsequent writes (drive resize, etc.).
22    Resize { rows: u16, cols: u16 },
23}
24
25impl WriteMsg {
26    /// Approximate in-memory weight, used only for diagnostics/observability.
27    pub fn byte_len(&self) -> usize {
28        match self {
29            WriteMsg::Bytes(b) => b.len(),
30            WriteMsg::Resize { .. } => 0,
31        }
32    }
33}
34
35#[derive(Debug, thiserror::Error, PartialEq, Eq)]
36pub enum WriteQueueError {
37    #[error("write queue at capacity ({capacity} messages); apply backpressure")]
38    Full { capacity: usize },
39}
40
41/// FIFO queue with a hard message-count capacity.
42#[derive(Debug)]
43pub struct WriteQueue {
44    inner: VecDeque<WriteMsg>,
45    capacity: usize,
46}
47
48impl WriteQueue {
49    /// Create a queue holding at most `capacity` messages. A capacity of 0 is
50    /// clamped to 1 so the queue is always usable.
51    pub fn new(capacity: usize) -> Self {
52        WriteQueue {
53            inner: VecDeque::new(),
54            capacity: capacity.max(1),
55        }
56    }
57
58    pub fn capacity(&self) -> usize {
59        self.capacity
60    }
61
62    pub fn len(&self) -> usize {
63        self.inner.len()
64    }
65
66    pub fn is_empty(&self) -> bool {
67        self.inner.is_empty()
68    }
69
70    pub fn is_full(&self) -> bool {
71        self.inner.len() >= self.capacity
72    }
73
74    /// Enqueue a message. Returns [`WriteQueueError::Full`] without mutating the
75    /// queue when at capacity (the message is returned to the caller untouched
76    /// so it can be retried after backpressure clears).
77    pub fn enqueue(&mut self, msg: WriteMsg) -> Result<(), (WriteMsg, WriteQueueError)> {
78        if self.is_full() {
79            return Err((
80                msg,
81                WriteQueueError::Full {
82                    capacity: self.capacity,
83                },
84            ));
85        }
86        self.inner.push_back(msg);
87        Ok(())
88    }
89
90    /// Pop the next message in FIFO order.
91    pub fn dequeue(&mut self) -> Option<WriteMsg> {
92        self.inner.pop_front()
93    }
94}
95
96#[cfg(test)]
97mod tests {
98    use super::*;
99
100    #[test]
101    fn fifo_order_preserved() {
102        let mut q = WriteQueue::new(4);
103        q.enqueue(WriteMsg::Bytes(b"a".to_vec())).unwrap();
104        q.enqueue(WriteMsg::Resize { rows: 24, cols: 80 }).unwrap();
105        q.enqueue(WriteMsg::Bytes(b"b".to_vec())).unwrap();
106        assert_eq!(q.dequeue(), Some(WriteMsg::Bytes(b"a".to_vec())));
107        assert_eq!(q.dequeue(), Some(WriteMsg::Resize { rows: 24, cols: 80 }));
108        assert_eq!(q.dequeue(), Some(WriteMsg::Bytes(b"b".to_vec())));
109        assert_eq!(q.dequeue(), None);
110    }
111
112    #[test]
113    fn full_returns_backpressure_and_does_not_drop_message() {
114        let mut q = WriteQueue::new(2);
115        q.enqueue(WriteMsg::Bytes(b"1".to_vec())).unwrap();
116        q.enqueue(WriteMsg::Bytes(b"2".to_vec())).unwrap();
117        let rejected = WriteMsg::Bytes(b"3".to_vec());
118        let err = q.enqueue(rejected.clone()).unwrap_err();
119        assert_eq!(
120            err.0, rejected,
121            "rejected message must be returned for retry"
122        );
123        assert_eq!(err.1, WriteQueueError::Full { capacity: 2 });
124        assert_eq!(q.len(), 2, "full queue must not have grown past capacity");
125
126        // Draining one frees a slot so the retry succeeds.
127        assert!(q.dequeue().is_some());
128        assert!(q.enqueue(rejected).is_ok());
129    }
130
131    #[test]
132    fn zero_capacity_clamps_to_one() {
133        let mut q = WriteQueue::new(0);
134        assert_eq!(q.capacity(), 1);
135        assert!(q.enqueue(WriteMsg::Bytes(b"x".to_vec())).is_ok());
136        assert!(q.is_full());
137    }
138}
fno_agents/write_queue.rs

fno_agents/
write_queue.rs
