locode_engine/
queue.rs

1//! The mid-run input queue (ADR-0028).
2//!
3//! A **handle taken before the run**, not a command. [`Session::run`] takes
4//! `&mut self` and the caller awaits it, so a frontend's command loop is not
5//! polled while a run is in flight — the same constraint that made
6//! `cancel_handle()` a pre-run clone (ADR-0018). A frontend clones this handle
7//! before starting a run and pushes into it; the loop drains it between
8//! dispatch and the next sample.
9//!
10//! [`Session::run`]: crate::Session::run
11
12use std::sync::{Arc, Mutex, PoisonError};
13
14/// The preamble that marks text as having arrived **mid-run**.
15///
16/// A mid-run message is not the same speech act as a fresh prompt: it may amend
17/// an earlier instruction, be an aside, or be an instruction for after the
18/// current work ("when you finish this, also…"). The model can only act
19/// correctly — including changing course inside the running loop — if it knows
20/// which. Unmarked, it reads as though it had been there from the start.
21///
22/// Grok Build ships the same marker and the same wording
23/// (`INTERJECTION_WIRE_PREFIX`), and its PTY tests assert the distinction this
24/// module preserves: the mid-run path carries the preamble, and a queued
25/// message that lands as an ordinary next-turn prompt does not.
26pub const MID_RUN_PREAMBLE: &str = "The user sent a message while you were working";
27
28/// A clonable handle to one session's pending mid-run input.
29///
30/// Clones share one queue. Cheap to clone and safe to hold across an await —
31/// the lock is only ever taken inside these methods, never returned.
32#[derive(Debug, Clone, Default)]
33pub struct InputQueue {
34    items: Arc<Mutex<Vec<String>>>,
35}
36
37impl InputQueue {
38    /// An empty queue.
39    #[must_use]
40    pub fn new() -> Self {
41        Self::default()
42    }
43
44    /// Queue a message typed while the run is in flight.
45    pub fn push(&self, text: impl Into<String>) {
46        self.lock().push(text.into());
47    }
48
49    /// The pending items, for a frontend to render. Does not consume.
50    #[must_use]
51    pub fn pending(&self) -> Vec<String> {
52        self.lock().clone()
53    }
54
55    /// Whether anything is waiting.
56    #[must_use]
57    pub fn is_empty(&self) -> bool {
58        self.lock().is_empty()
59    }
60
61    /// Take everything, leaving the queue empty.
62    #[must_use]
63    pub fn take_all(&self) -> Vec<String> {
64        std::mem::take(&mut *self.lock())
65    }
66
67    /// Drop every pending item.
68    ///
69    /// Called on cancel: the message was written for a run the user then
70    /// abandoned, so delivering it into the next turn would be surprising. This
71    /// covers **undelivered** items only — anything already drained is in
72    /// `history` and cancel rolls nothing back (ADR-0028).
73    pub fn clear(&self) {
74        self.lock().clear();
75    }
76
77    /// The queued text as one marked block, or `None` when nothing is waiting.
78    ///
79    /// Multiple items concatenate rather than arriving as separate turns: the
80    /// common case is a user typing several lines before sending, not issuing
81    /// competing instructions.
82    #[must_use]
83    pub(crate) fn take_marked(&self) -> Option<String> {
84        let items = self.take_all();
85        if items.is_empty() {
86            return None;
87        }
88        Some(format!("{MID_RUN_PREAMBLE}:\n\n{}", items.join("\n\n")))
89    }
90
91    fn lock(&self) -> std::sync::MutexGuard<'_, Vec<String>> {
92        // The guard only ever protects a Vec and is never held across an await,
93        // so recovering from a poisoned lock is strictly better than panicking
94        // inside the loop (denied lints forbid unwrap here anyway).
95        self.items.lock().unwrap_or_else(PoisonError::into_inner)
96    }
97}
98
99#[cfg(test)]
100mod tests {
101    use super::*;
102
103    #[test]
104    fn clones_share_one_queue() {
105        let a = InputQueue::new();
106        let b = a.clone();
107        a.push("from a");
108        assert_eq!(b.pending(), vec!["from a".to_string()]);
109        assert!(!b.is_empty());
110    }
111
112    #[test]
113    fn multiple_items_concatenate_under_one_marker() {
114        let q = InputQueue::new();
115        q.push("first line");
116        q.push("second line");
117        let marked = q.take_marked().expect("something queued");
118        assert!(marked.starts_with(MID_RUN_PREAMBLE));
119        assert!(marked.contains("first line"));
120        assert!(marked.contains("second line"));
121        assert_eq!(
122            marked.matches(MID_RUN_PREAMBLE).count(),
123            1,
124            "one marker for the whole batch, not one per item"
125        );
126        assert!(q.is_empty(), "taking consumes");
127    }
128
129    #[test]
130    fn an_empty_queue_yields_nothing_to_inject() {
131        assert!(InputQueue::new().take_marked().is_none());
132    }
133
134    #[test]
135    fn clear_drops_undelivered_items() {
136        let q = InputQueue::new();
137        q.push("abandoned");
138        q.clear();
139        assert!(q.is_empty());
140        assert!(q.take_marked().is_none());
141    }
142}
locode_engine/queue.rs

locode_engine/
queue.rs
