car_server_core/coder/
steering.rs

1//! Operator guidance delivered at native model-turn boundaries.
2//! The session snapshot owns durability; this inbox only coordinates the live loop.
3
4use std::collections::VecDeque;
5use std::sync::{Arc, Mutex};
6
7use super::session::{CoderEventKind, EventSink};
8
9#[derive(Debug, Default)]
10struct Pending {
11    open: bool,
12    messages: VecDeque<String>,
13}
14
15#[derive(Debug, Default)]
16pub struct SteeringInbox(Mutex<Pending>);
17
18impl SteeringInbox {
19    pub fn is_open(&self) -> bool {
20        self.0.lock().expect("steering inbox poisoned").open
21    }
22
23    pub fn has_pending(&self) -> bool {
24        !self
25            .0
26            .lock()
27            .expect("steering inbox poisoned")
28            .messages
29            .is_empty()
30    }
31    pub fn enter<'a>(self: &Arc<Self>, sink: &'a EventSink) -> SteeringScope<'a> {
32        self.0.lock().expect("steering inbox poisoned").open = true;
33        SteeringScope {
34            inbox: self.clone(),
35            sink,
36        }
37    }
38
39    /// Persist before acknowledgement. Holding the inbox lock makes admission
40    /// atomic with loop completion, and a failed write never queues guidance.
41    pub fn enqueue(
42        &self,
43        text: String,
44        persist: impl FnOnce() -> Result<(), String>,
45    ) -> Result<(), String> {
46        let mut pending = self.0.lock().expect("steering inbox poisoned");
47        if !pending.open {
48            return Err("This task is not accepting steering. Native tasks accept guidance while running; use a follow-up conversation after the task finishes.".into());
49        }
50        if pending.messages.len() >= 16 {
51            return Err("Too many queued instructions. Wait for the agent to receive them before sending more.".into());
52        }
53        persist()?;
54        pending.messages.push_back(text);
55        Ok(())
56    }
57
58    pub fn take(&self) -> Vec<String> {
59        self.0
60            .lock()
61            .expect("steering inbox poisoned")
62            .messages
63            .drain(..)
64            .collect()
65    }
66}
67
68pub struct SteeringScope<'a> {
69    inbox: Arc<SteeringInbox>,
70    sink: &'a EventSink,
71}
72
73impl Drop for SteeringScope<'_> {
74    fn drop(&mut self) {
75        let mut pending = self.inbox.0.lock().expect("steering inbox poisoned");
76        pending.open = false;
77        for text in pending.messages.drain(..) {
78            self.sink.emit(CoderEventKind::OperatorGuidance {
79                text,
80                status: "not_applied".into(),
81            });
82        }
83    }
84}
85
86#[cfg(test)]
87mod tests {
88    use super::*;
89
90    #[test]
91    fn steering_requires_live_scope_and_durable_acceptance() {
92        let inbox = Arc::new(SteeringInbox::default());
93        assert!(inbox.enqueue("before".into(), || Ok(())).is_err());
94        let (sink, events) = EventSink::collecting("steering");
95        let scope = inbox.enter(&sink);
96        assert!(inbox
97            .enqueue("lost write".into(), || Err("disk full".into()))
98            .is_err());
99        assert!(inbox.take().is_empty());
100        inbox
101            .enqueue("retain unfinished work".into(), || Ok(()))
102            .unwrap();
103        drop(scope);
104        assert!(!inbox.is_open());
105        assert!(inbox.enqueue("too late".into(), || Ok(())).is_err());
106        assert!(events.lock().unwrap().iter().any(|event| matches!(&event.kind,
107            CoderEventKind::OperatorGuidance { text, status } if text == "retain unfinished work" && status == "not_applied")));
108    }
109}
car_server_core/coder/steering.rs

car_server_core/coder/
steering.rs
