agentyk_core/
input.rs

1//! Steering — input that arrives while a turn is already running.
2//!
3//! A turn can take minutes. In that time the person who started it learns
4//! things: the approach is wrong, the file was already fixed, stop after this
5//! test. Without somewhere to put that, their only options are to wait or to
6//! cancel and start over, losing everything the turn has done.
7//!
8//! An [`InputQueue`] is that somewhere. A caller holds a clone, pushes a
9//! message whenever it likes, and the executor drains it **between reasoning
10//! steps** — see [`InputQueue::drain`] for why that boundary and not any
11//! other. Drained messages are recorded as ordinary `input.message` events, so
12//! they enter history exactly like the message that opened the turn and a
13//! replay reconstructs the same conversation.
14//!
15//! Deliberately std-only (a `Mutex<VecDeque>`, no channel crate) so it lives
16//! in `agentyk-core` and works the same in any host.
17//!
18//! ```
19//! use agentyk_core::{input::InputQueue, message::Message};
20//!
21//! let queue = InputQueue::new();
22//! let steering = queue.clone();
23//! steering.push(Message::user("actually, skip the tests"));
24//! assert_eq!(queue.drain().len(), 1);
25//! assert!(queue.drain().is_empty(), "draining consumes");
26//! ```
27
28use std::collections::VecDeque;
29use std::sync::{Arc, Mutex};
30
31use crate::message::Message;
32
33/// Messages waiting to join a turn already in progress. Cheap to clone;
34/// every clone shares one queue.
35#[derive(Clone, Default)]
36pub struct InputQueue(Arc<Mutex<VecDeque<Message>>>);
37
38impl InputQueue {
39    /// An empty queue.
40    pub fn new() -> Self {
41        Self::default()
42    }
43
44    /// Add a message for the running turn to pick up at its next reasoning
45    /// step. Never blocks and never fails: a queue with no turn draining it
46    /// simply holds what it was given, and the next turn takes it.
47    pub fn push(&self, message: Message) {
48        self.0
49            .lock()
50            .unwrap_or_else(|error| error.into_inner())
51            .push_back(message);
52    }
53
54    /// Whether anything is waiting. Cheaper than draining for a host that
55    /// only wants to know.
56    pub fn is_empty(&self) -> bool {
57        self.0
58            .lock()
59            .unwrap_or_else(|error| error.into_inner())
60            .is_empty()
61    }
62
63    /// Take everything queued, in the order it was pushed.
64    ///
65    /// Executors call this **only when the next action is a reasoning step**,
66    /// never mid-batch. That is a protocol constraint, not a preference:
67    /// providers require every tool call to be answered by its result, so a
68    /// user message wedged between a `tool_use` and its `tool_result` makes
69    /// the exchange invalid. Steering therefore lands at the next model call —
70    /// which is also the first moment the model could act on it anyway.
71    pub fn drain(&self) -> Vec<Message> {
72        self.0
73            .lock()
74            .unwrap_or_else(|error| error.into_inner())
75            .drain(..)
76            .collect()
77    }
78}
79
80impl std::fmt::Debug for InputQueue {
81    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
82        f.debug_struct("InputQueue")
83            .field("queued", &self.0.lock().map(|q| q.len()).unwrap_or(0))
84            .finish()
85    }
86}
87
88#[cfg(test)]
89mod tests {
90    use super::*;
91
92    #[test]
93    fn clones_share_one_queue() {
94        let queue = InputQueue::new();
95        let handle = queue.clone();
96        assert!(queue.is_empty());
97
98        handle.push(Message::user("first"));
99        handle.push(Message::user("second"));
100        assert!(!queue.is_empty());
101
102        let drained = queue.drain();
103        assert_eq!(drained.len(), 2);
104        assert_eq!(drained[0].text(), "first");
105        assert_eq!(drained[1].text(), "second");
106        assert!(handle.is_empty(), "draining one clone drains them all");
107    }
108
109    #[test]
110    fn a_queue_nobody_is_draining_keeps_its_messages() {
111        let queue = InputQueue::new();
112        queue.push(Message::user("held"));
113        assert_eq!(queue.drain().len(), 1);
114    }
115}
agentyk_core/input.rs

agentyk_core/
input.rs
